import logging import os from typing import Any, Dict, Optional import requests from django.conf import settings from verifier.engine.base import ( BaseVendor, DocumentType, VendorResult, VerificationStatus, ) logger = logging.getLogger(__name__) class CodeplusVendor(BaseVendor): """ Codeplus Verification Adapter (apihub.services). Implements verification workflows for PAN Advanced, Aadhaar, Driving License, Voter ID, Passport, Bank Account, GSTIN, and custom documents. """ name: str = "codeplus" display_name: str = "Codeplus Verification Services" priority: int = 10 # Lower number = higher priority in waterfall is_active: bool = True supported_documents = [ DocumentType.PAN.value, DocumentType.PAN_V5.value, DocumentType.PAN_VERIFY.value, DocumentType.PAN_ADVANCED.value, DocumentType.PAN_DETAILED.value, DocumentType.AADHAAR.value, DocumentType.DRIVING_LICENSE.value, DocumentType.VOTER_ID.value, DocumentType.PASSPORT.value, DocumentType.BANK_ACCOUNT.value, DocumentType.BANK_VERIFY_PD.value, DocumentType.BANK_VERIFY_V5.value, DocumentType.BANK_PD.value, DocumentType.GSTIN.value, DocumentType.GST.value, DocumentType.GST_VERIFY.value, DocumentType.GST_BY_PAN.value, DocumentType.FETCH_GST.value, DocumentType.MSME.value, DocumentType.MSME_BY_PAN.value, DocumentType.UDYAM.value, DocumentType.CUSTOM.value, ] def __init__(self, config: Optional[Dict[str, Any]] = None): super().__init__(config=config) # Base URL: default https://apihub.services self.base_url = ( self.config.get("base_url") or getattr(settings, "CODEPLUS_BASE_URL", None) or os.environ.get("CODEPLUS_BASE_URL", "https://apihub.services") ).rstrip("/") # Client Credentials self.client_api_id = ( self.config.get("client_api_id") or self.config.get("client_id") or getattr(settings, "CODEPLUS_CLIENT_API_ID", None) or os.environ.get("CODEPLUS_CLIENT_API_ID", "CAPI-YYR0YJIE0VMTDENY") ) self.client_api_secret = ( self.config.get("client_api_secret") or self.config.get("api_secret") or getattr(settings, "CODEPLUS_CLIENT_API_SECRET", None) or os.environ.get("CODEPLUS_CLIENT_API_SECRET", "") ) self.timeout = int(self.config.get("timeout", 20)) self.session = requests.Session() def get_headers(self) -> Dict[str, str]: """ Construct required HTTP headers for Codeplus / apihub.services API calls: - X-Client-Api-ID - X-Client-Api-Secret - Content-Type: application/json """ headers = { "Content-Type": "application/json", "Accept": "application/json", } if self.client_api_id: headers["X-Client-Api-ID"] = self.client_api_id if self.client_api_secret: headers["X-Client-Api-Secret"] = self.client_api_secret return headers def execute_verification( self, document_type: str, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None, ) -> VendorResult: """ Dispatches verification to specific document handlers. """ handlers = { DocumentType.PAN.value: self.verify_pan, DocumentType.PAN_V5.value: self.verify_pan_v5, DocumentType.PAN_VERIFY.value: self.verify_pan_v5, DocumentType.PAN_ADVANCED.value: self.verify_pan_advanced, DocumentType.PAN_DETAILED.value: self.verify_pan_detailed, DocumentType.AADHAAR.value: self.verify_aadhaar, DocumentType.DRIVING_LICENSE.value: self.verify_driving_license, DocumentType.VOTER_ID.value: self.verify_voter_id, DocumentType.PASSPORT.value: self.verify_passport, DocumentType.BANK_ACCOUNT.value: self.verify_bank_account, DocumentType.BANK_VERIFY_PD.value: self.verify_bank_verify_pd_v5, DocumentType.BANK_VERIFY_V5.value: self.verify_bank_verify_pd_v5, DocumentType.BANK_PD.value: self.verify_bank_verify_pd_v5, DocumentType.GSTIN.value: self.verify_gstin, DocumentType.GST.value: self.verify_gstin, DocumentType.GST_VERIFY.value: self.verify_gstin, DocumentType.GST_BY_PAN.value: self.fetch_gst_by_pan, DocumentType.FETCH_GST.value: self.fetch_gst_by_pan, DocumentType.MSME.value: self.fetch_msme_by_pan, DocumentType.MSME_BY_PAN.value: self.fetch_msme_by_pan, DocumentType.UDYAM.value: self.fetch_msme_by_pan, } handler = handlers.get(document_type, self.verify_generic) return handler(data, files) # ────────────────────────────────────────────────────────────────────────── # PAN Verification (v5 Recommended, v4 Advanced, & v4 Detailed APIs) # ────────────────────────────────────────────────────────────────────────── def verify_pan(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Dispatches to: - verify_pan_v5 (Recommended default) - verify_pan_detailed (if mode == 'detailed' or version == 'v4_detailed') - verify_pan_advanced (if mode == 'advanced' or version == 'v4_advanced') """ mode = str(data.get("mode", "")).lower() version = str(data.get("version", "")).lower() if data.get("detailed") is True or mode == "detailed" or version == "v4_detailed": return self.verify_pan_detailed(data, files) if mode == "advanced" or version == "v4_advanced": return self.verify_pan_advanced(data, files) # Default to v5 Recommended API v5_result = self.verify_pan_v5(data, files) if v5_result.is_success: return v5_result # If v5 encounters an error/retryable failure, fallback locally to v4 advanced if v5_result.retryable: v4_result = self.verify_pan_advanced(data, files) if v4_result.is_success: return v4_result return v5_result def verify_pan_v5(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Pan Verification (v5) Recommended API: Endpoint: POST https://apihub.services/v5/pan-verify Payload: { "pan": "" } Headers: X-Client-Api-ID: X-Client-Api-Secret: Returns full holder details: name, dob, gender, aadhaar linking, address details, pan type. """ endpoint = f"{self.base_url}/v5/pan-verify" pan_number = ( data.get("pan") or data.get("pan_number") or data.get("document_id") or "" ).strip().upper() if not pan_number: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_PAN", error_message="pan is required for PAN verification", retryable=False, ) payload = {"pan": pan_number} headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} error_block = raw_json.get("error") if isinstance(raw_json.get("error"), dict) else {} is_success = ( response.status_code == 200 and not error_block and ( raw_json.get("success") is True or raw_json.get("status") == "SUCCESS" or raw_json.get("statusCode") == 200 or data_block.get("success") is True or data_block.get("status") == "SUCCESS" ) ) if is_success: normalized_result = { "document_type": "PAN", "pan": data_block.get("pan") or pan_number, "pan_type": data_block.get("panType"), "full_name": data_block.get("fullName") or data_block.get("name"), "first_name": data_block.get("firstName"), "middle_name": data_block.get("middleName"), "last_name": data_block.get("lastName"), "gender": data_block.get("gender"), "dob": data_block.get("dob"), "aadhaar_number": data_block.get("aadhaarNumber"), "aadhaar_linked": data_block.get("aadhaarLinked"), "address_details": data_block.get("addressDetails"), "service": data_block.get("service", "PAN_VERIFICATION"), "result_code": data_block.get("resultCode"), "client_ref_num": data_block.get("clientRefNum"), "provider_transaction_id": data_block.get("providerTransactionId"), "transaction_id": data_block.get("transactionId") or raw_json.get("transactionId"), "request_id": raw_json.get("requestId") or raw_json.get("request_id"), "timestamp": raw_json.get("timestamp"), "pan_details": data_block, } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling is_server_error = response.status_code >= 500 if error_block: err_code = str(error_block.get("code") or f"HTTP_{response.status_code}") err_message = error_block.get("message") or "PAN verification failed" elif data_block: err_code = str(data_block.get("resultCode") or data_block.get("code") or f"HTTP_{response.status_code}") err_message = data_block.get("message") or raw_json.get("message") or "PAN verification failed" else: err_code = f"HTTP_{response.status_code}" err_message = raw_json.get("message") or "PAN verification failed" logger.warning( "Codeplus PAN Verification (v5) failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus PAN Verification (v5) timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus PAN Verification (v5) timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus PAN Verification (v5) network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus PAN v5 API: {exc}", retryable=True, ) def verify_pan_advanced(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ PAN Advanced API: Endpoint: POST https://apihub.services/v4/pan-advanced Payload: {"pan": ""} Headers: X-Client-Api-ID: X-Client-Api-Secret: """ endpoint = f"{self.base_url}/v4/pan-advanced" pan_number = ( data.get("pan") or data.get("pan_number") or data.get("document_id") or "" ).strip().upper() if not pan_number: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_PAN", error_message="PAN number is required in input data", retryable=False, ) payload = {"pan": pan_number} headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} code = str(data_block.get("code") or "") message = data_block.get("message") or raw_json.get("message") # Success condition: HTTP 200 and Code 1000 ("Valid PAN.") if response.status_code == 200 and code == "1000": pan_data = data_block.get("pan_data") or data_block normalized_result = { "document_type": pan_data.get("document_type", "PAN"), "document_id": pan_data.get("document_id", pan_number), "name": pan_data.get("name"), "first_name": pan_data.get("first_name"), "last_name": pan_data.get("last_name"), "category": pan_data.get("category"), "category_type": pan_data.get("category_type"), "is_individual": pan_data.get("is_individual"), "individual_tax_compliance_status": pan_data.get("individual_tax_compliance_status"), "aadhaar_linked": pan_data.get("aadhaar_linked"), "pan_details": pan_data, "request_id": raw_json.get("request_id"), "transaction_id": raw_json.get("transaction_id"), } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling (e.g. code 1004 "PAN does not exist") is_server_error = response.status_code >= 500 err_code = code or f"HTTP_{response.status_code}" err_message = message or f"PAN verification failed (Code: {err_code})" logger.warning( "Codeplus PAN Advanced verification failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus PAN verification timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus PAN API timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus PAN verification network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus PAN API: {exc}", retryable=True, ) def verify_pan_detailed(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Detailed PAN API: Endpoint: POST https://apihub.services/v4/pan-detailed Payload: {"pan_number": ""} Headers: X-Client-Api-ID: X-Client-Api-Secret: Returns comprehensive details: name, dob, address data, email, mobile, gender, aadhaar_linked, taxpayer_profile """ endpoint = f"{self.base_url}/v4/pan-detailed" pan_number = ( data.get("pan_number") or data.get("pan") or data.get("document_id") or "" ).strip().upper() if not pan_number: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_PAN", error_message="pan_number is required in input data", retryable=False, ) payload = {"pan_number": pan_number} headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} error_block = raw_json.get("error") if isinstance(raw_json.get("error"), dict) else {} code = str(data_block.get("code") or "") # Success condition: HTTP 200 and Code 1000 ("Valid PAN.") or valid data block is_code_success = code in ("1000", "SUCCESS", "200") is_body_success = response.status_code == 200 and not error_block and ("data" in raw_json and bool(data_block)) if is_code_success or is_body_success: pan_data = data_block.get("pan_data") or data_block normalized_result = { "document_type": pan_data.get("document_type", "PAN"), "document_id": pan_data.get("document_id", pan_number), "name": pan_data.get("name"), "first_name": pan_data.get("first_name"), "last_name": pan_data.get("last_name"), "category": pan_data.get("category"), "date_of_birth": pan_data.get("date_of_birth"), "masked_aadhaar_number": pan_data.get("masked_aadhaar_number"), "address_data": pan_data.get("address_data"), "email": pan_data.get("email"), "phone": pan_data.get("phone"), "gender": pan_data.get("gender"), "aadhaar_linked": pan_data.get("aadhaar_linked"), "taxpayer_profile": pan_data.get("taxpayer_profile"), "pan_details": pan_data, "request_id": raw_json.get("request_id"), "transaction_id": raw_json.get("transaction_id"), } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling (e.g. HTTP 400 with {"error": {"code": "INVALID_PAN", "message": "Invalid PAN number."}}) is_server_error = response.status_code >= 500 if error_block: err_code = str(error_block.get("code") or f"HTTP_{response.status_code}") err_message = error_block.get("message") or f"Detailed PAN verification failed" elif data_block: err_code = str(data_block.get("code") or f"HTTP_{response.status_code}") err_message = data_block.get("message") or f"Detailed PAN verification failed" else: err_code = f"HTTP_{response.status_code}" err_message = raw_json.get("message") or "Detailed PAN verification failed" logger.warning( "Codeplus PAN Detailed verification failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus PAN Detailed verification timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus PAN Detailed API timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus PAN Detailed verification network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus PAN Detailed API: {exc}", retryable=True, ) # ────────────────────────────────────────────────────────────────────────── # Other Document Verification Handlers (Stubs ready for additional endpoints) # ────────────────────────────────────────────────────────────────────────── def verify_aadhaar(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: endpoint = f"{self.base_url}/v4/aadhaar-verify" return self._send_generic_request(endpoint, payload=data, doc_label="Aadhaar") def verify_driving_license(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: endpoint = f"{self.base_url}/v4/dl-verify" return self._send_generic_request(endpoint, payload=data, doc_label="Driving License") def verify_voter_id(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: endpoint = f"{self.base_url}/v4/voter-id-verify" return self._send_generic_request(endpoint, payload=data, doc_label="Voter ID") def verify_passport(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: endpoint = f"{self.base_url}/v4/passport-verify" return self._send_generic_request(endpoint, payload=data, doc_label="Passport") def verify_bank_account(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Bank Account Verification Dispatcher: - Defaults to Bank Verify (Penny Drop - v5) Recommended API (`verify_bank_verify_pd_v5`) - If mode or version is 'v4', 'detailed', or 'full', routes to `verify_bank_account_v4` - If v5 encounters an upstream retryable failure (5xx or timeout), falls back to v4 """ mode = str(data.get("mode", "")).lower() version = str(data.get("version", "")).lower() if version == "v4" or mode in ("v4", "detailed", "full"): return self.verify_bank_account_v4(data, files) # Primary execution: Bank Verify (Penny Drop - v5) v5_result = self.verify_bank_verify_pd_v5(data, files) if v5_result.is_success: return v5_result # Fallback locally to v4 if upstream server error / timeout or if upstream returns v4 response structure is_v4_payload = isinstance(v5_result.raw_response, dict) and "bank_data" in v5_result.raw_response.get("data", {}) if v5_result.retryable or is_v4_payload: logger.info("Codeplus Bank Verify fallback to v4 triggered (retryable=%s, is_v4=%s)...", v5_result.retryable, is_v4_payload) v4_result = self.verify_bank_account_v4(data, files) if v4_result.is_success: return v4_result return v5_result def verify_bank_verify_pd_v5(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Bank Verify (Penny Drop - v5): Endpoint: POST https://apihub.services/v5/bank-verify-pd Payload: { "account_number": "", "ifsc": "" } Headers: X-Client-Api-ID: X-Client-Api-Secret: Content-Type: application/json Description: Verifies a bank account using account number and IFSC code via penny drop. Returns validity, account holder name, and verified bank account details. Response (HTTP 200): { "status": "success", "result": { "failure_reason": null, "account_number": "********7270", "user": null, "is_valid": true, "account_holder_name": "MANOJ PRASAD", "account_ifsc": "UTIB0000373", "npci_error_code": null }, "tranx_id": "9eeb9d58-04f8-4f8b-9608-99731850993f", "timestamp": "2026-05-22T05:10:43.561554+00:00", "chargeble": "true", "user_consent": "true" } """ endpoint = f"{self.base_url}/v5/bank-verify-pd" account_number = str( data.get("account_number") or data.get("bank_account") or data.get("account_no") or data.get("account") or "" ).strip() ifsc = str( data.get("ifsc") or data.get("ifsc_code") or "" ).strip().upper() if not account_number: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_ACCOUNT_NUMBER", error_message="account_number is required for Bank Verify (Penny Drop - v5)", retryable=False, ) if not ifsc: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_IFSC", error_message="ifsc code is required for Bank Verify (Penny Drop - v5)", retryable=False, ) payload = { "account_number": account_number, "ifsc": ifsc, } headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} api_status = str(raw_json.get("status", "")).lower() result_block = raw_json.get("result") if isinstance(raw_json.get("result"), dict) else {} is_valid = result_block.get("is_valid") account_holder_name = result_block.get("account_holder_name") or "" failure_reason = result_block.get("failure_reason") npci_error_code = result_block.get("npci_error_code") masked_acc = result_block.get("account_number") or ( f"********{account_number[-4:]}" if len(account_number) >= 4 else account_number ) account_ifsc = result_block.get("account_ifsc") or ifsc if response.status_code == 200: # Success condition: status == 'success' and is_valid is True if api_status == "success" and (is_valid is True or (is_valid is None and account_holder_name and not failure_reason)): normalized_result = { "document_type": "BANK_ACCOUNT", "verification_mode": "penny_drop_v5", "account_number": account_number, "masked_account_number": masked_acc, "ifsc": ifsc, "account_ifsc": account_ifsc, "beneficiary_name": account_holder_name, "account_holder_name": account_holder_name, "is_valid": True, "account_exists": True, "failure_reason": None, "npci_error_code": None, "user": result_block.get("user"), "tranx_id": raw_json.get("tranx_id"), "transaction_id": raw_json.get("tranx_id"), "timestamp": raw_json.get("timestamp"), "chargeble": raw_json.get("chargeble"), "user_consent": raw_json.get("user_consent"), "bank_details": result_block, } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # HTTP 200 with verification failure or account invalid err_code = str(npci_error_code or "INVALID_BANK_ACCOUNT") err_message = failure_reason or raw_json.get("message") or "Bank account verification failed" normalized_result = { "document_type": "BANK_ACCOUNT", "verification_mode": "penny_drop_v5", "account_number": account_number, "masked_account_number": masked_acc, "ifsc": ifsc, "account_ifsc": account_ifsc, "beneficiary_name": account_holder_name, "account_holder_name": account_holder_name, "is_valid": False, "account_exists": False, "failure_reason": failure_reason, "npci_error_code": npci_error_code, "user": result_block.get("user"), "tranx_id": raw_json.get("tranx_id"), "transaction_id": raw_json.get("tranx_id"), "timestamp": raw_json.get("timestamp"), "chargeble": raw_json.get("chargeble"), "user_consent": raw_json.get("user_consent"), "bank_details": result_block, } logger.warning( "Codeplus Bank Verify PD v5 returned failure: npci_error=%s, failure_reason=%s", npci_error_code, failure_reason, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, data=normalized_result, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=False, ) # Upstream HTTP error status (400, 500, etc.) is_server_error = response.status_code >= 500 err_code = str(npci_error_code or f"HTTP_{response.status_code}") err_message = failure_reason or raw_json.get("message") or raw_json.get("detail") or raw_json.get("error") or "Bank verification failed" if isinstance(err_message, dict): err_message = str(err_message) logger.warning( "Codeplus Bank Verify PD v5 failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=raw_json if isinstance(raw_json, dict) else {}, raw_response=raw_json, error_code=err_code, error_message=str(err_message), retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus Bank Verify PD v5 timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus Bank Verify PD v5 timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus Bank Verify PD v5 network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus Bank Verify PD v5 API: {exc}", retryable=True, ) def verify_bank_account_v4(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Bank Verification (Full Details - v4) API: Endpoint: POST https://apihub.services/v4/bank-verify Payload: { "account_number": "", "ifsc": "" } Headers: X-Client-Api-ID: X-Client-Api-Secret: Retrieves account holder name via penny drop routing, MICR code, city, pincode, etc. """ endpoint = f"{self.base_url}/v4/bank-verify" account_number = str( data.get("account_number") or data.get("bank_account") or data.get("account_no") or data.get("account") or "" ).strip() ifsc = str( data.get("ifsc") or data.get("ifsc_code") or "" ).strip().upper() if not account_number: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_ACCOUNT_NUMBER", error_message="account_number is required for bank verification", retryable=False, ) if not ifsc: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_IFSC", error_message="ifsc code is required for bank verification", retryable=False, ) payload = { "account_number": account_number, "ifsc": ifsc, } headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} error_block = raw_json.get("error") if isinstance(raw_json.get("error"), dict) else {} code = str(data_block.get("code") or "") # Success condition: HTTP 200 and either code 1000 or valid bank verification payload is_code_success = code in ("1000", "SUCCESS", "200") is_body_success = response.status_code == 200 and not error_block and ("data" in raw_json or "account_exists" in raw_json) if is_code_success or is_body_success: bank_data = data_block.get("bank_data") or data_block or raw_json beneficiary_name = ( bank_data.get("name_at_bank") or bank_data.get("beneficiary_name") or bank_data.get("account_holder_name") or bank_data.get("name") or data_block.get("name") ) normalized_result = { "document_type": "BANK_ACCOUNT", "account_number": account_number, "ifsc": ifsc, "beneficiary_name": beneficiary_name, "micr": bank_data.get("micr") or bank_data.get("micr_code"), "bank_name": bank_data.get("bank_name") or bank_data.get("bank"), "branch": bank_data.get("branch"), "city": bank_data.get("city"), "state": bank_data.get("state"), "pincode": bank_data.get("pincode") or bank_data.get("pin_code"), "account_exists": bank_data.get("account_exists", True), "bank_details": bank_data, "request_id": raw_json.get("request_id"), "transaction_id": raw_json.get("transaction_id"), } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling is_server_error = response.status_code >= 500 if error_block: err_code = str(error_block.get("code") or f"HTTP_{response.status_code}") err_message = error_block.get("message") or "Bank verification failed" elif data_block: err_code = str(data_block.get("code") or f"HTTP_{response.status_code}") err_message = data_block.get("message") or "Bank verification failed" else: err_code = f"HTTP_{response.status_code}" err_message = raw_json.get("message") or "Bank verification failed" logger.warning( "Codeplus Bank verification failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus Bank verification timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus Bank verification timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus Bank verification network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus Bank API: {exc}", retryable=True, ) def fetch_gst_by_pan(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Fetch GST By PAN API: Endpoint: POST https://apihub.services/v4/fetch-gst Payload: { "pan": "" } Headers: X-Client-Api-ID: X-Client-Api-Secret: Returns all GSTINs associated with a valid business PAN. """ endpoint = f"{self.base_url}/v4/fetch-gst" pan = str( data.get("pan") or data.get("pan_number") or data.get("document_id") or "" ).strip().upper() if not pan: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_PAN", error_message="pan is required to fetch GSTIN details", retryable=False, ) payload = {"pan": pan} headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} error_block = raw_json.get("error") if isinstance(raw_json.get("error"), dict) else {} code = str(data_block.get("code") or "") # Success condition: HTTP 200 and presence of results or valid code (e.g. 1002, 1000) is_success_code = code in ("1000", "1001", "1002", "SUCCESS") has_results = "results" in data_block or isinstance(raw_json.get("results"), list) if response.status_code == 200 and not error_block and (is_success_code or has_results or "data" in raw_json): results = data_block.get("results") if "results" in data_block else raw_json.get("results", []) normalized_result = { "document_type": "GST_BY_PAN", "pan": pan, "records_count": len(results), "message": data_block.get("message") or f"Found {len(results)} records", "results": results, "request_id": raw_json.get("request_id"), "transaction_id": raw_json.get("transaction_id"), } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling is_server_error = response.status_code >= 500 if error_block: err_code = str(error_block.get("code") or f"HTTP_{response.status_code}") err_message = error_block.get("message") or "GST by PAN fetch failed" elif data_block: err_code = str(data_block.get("code") or f"HTTP_{response.status_code}") err_message = data_block.get("message") or "GST by PAN fetch failed" else: err_code = f"HTTP_{response.status_code}" err_message = raw_json.get("message") or "GST by PAN fetch failed" logger.warning( "Codeplus Fetch GST by PAN failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus Fetch GST by PAN timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus Fetch GST by PAN timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus Fetch GST by PAN network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus Fetch GST API: {exc}", retryable=True, ) def verify_gstin(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ GST Verification (v5) API: Endpoint: POST https://apihub.services/v5/gst-verify Payload: { "gst_number": "" } Headers: X-Client-Api-ID: X-Client-Api-Secret: Returns legal business name and current GST registration status. """ gst_number = str( data.get("gst_number") or data.get("gstin") or data.get("gst") or data.get("document_id") or "" ).strip().upper() # If a 10-character PAN is supplied instead of a 15-character GSTIN, route to fetch_gst_by_pan if not gst_number and (data.get("pan") or data.get("pan_number")): return self.fetch_gst_by_pan(data, files) if len(gst_number) == 10 and gst_number[3] in "CPHFATBLJG": return self.fetch_gst_by_pan({"pan": gst_number}, files) if not gst_number: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_GST_NUMBER", error_message="gst_number is required for GST verification", retryable=False, ) endpoint = f"{self.base_url}/v5/gst-verify" payload = {"gst_number": gst_number} headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} error_block = raw_json.get("error") if isinstance(raw_json.get("error"), dict) else {} # Success condition: HTTP 200 and success == True or statusCode == 200 is_success = ( response.status_code == 200 and not error_block and (raw_json.get("success") is True or raw_json.get("statusCode") == 200 or "legalName" in data_block) ) if is_success: normalized_result = { "document_type": "GSTIN", "gst_number": data_block.get("gstNumber") or gst_number, "legal_name": data_block.get("legalName"), "registration_status": data_block.get("registrationStatus"), "service": data_block.get("service", "GST_VERIFICATION"), "message": raw_json.get("message", "GST verification completed"), "gst_details": data_block, "request_id": raw_json.get("request_id"), "transaction_id": raw_json.get("transaction_id"), } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling is_server_error = response.status_code >= 500 if error_block: err_code = str(error_block.get("code") or f"HTTP_{response.status_code}") err_message = error_block.get("message") or "GST verification failed" elif data_block: err_code = str(data_block.get("code") or f"HTTP_{response.status_code}") err_message = data_block.get("message") or raw_json.get("message") or "GST verification failed" else: err_code = f"HTTP_{response.status_code}" err_message = raw_json.get("message") or "GST verification failed" logger.warning( "Codeplus GST Verification (v5) failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus GST verification timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus GST verification timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus GST verification network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus GST API: {exc}", retryable=True, ) def fetch_msme_by_pan(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ Fetch MSME By PAN API: Endpoint: POST https://apihub.services/v4/msme-pan Payload: { "pan": "", "detailed_response": true/false # (optional) } Headers: X-Client-Api-ID: X-Client-Api-Secret: Retrieves the Udyam Registration Number linked to a PAN and full Udyam details. """ endpoint = f"{self.base_url}/v4/msme-pan" pan = str( data.get("pan") or data.get("pan_number") or data.get("document_id") or "" ).strip().upper() if not pan: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED, error_code="MISSING_PAN", error_message="pan is required to fetch MSME details", retryable=False, ) payload: Dict[str, Any] = {"pan": pan} if data.get("detailed_response") is True or data.get("detailed") is True: payload["detailed_response"] = True headers = self.get_headers() try: response = self.session.post( endpoint, json=payload, headers=headers, timeout=self.timeout, ) try: raw_json = response.json() except Exception: raw_json = {"text": response.text, "status_code": response.status_code} data_block = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else {} error_block = raw_json.get("error") if isinstance(raw_json.get("error"), dict) else {} code = str(data_block.get("code") or "") # Success condition: HTTP 200 and no error block if response.status_code == 200 and not error_block: # Code 1011 = "No record found." has_record = code != "1011" and ("udyam" in data_block or "results" in data_block or code in ("1000", "1002")) normalized_result = { "document_type": "MSME", "pan": pan, "has_msme": has_record, "code": code, "message": data_block.get("message") or ("MSME record found" if has_record else "No record found."), "udyam_data": data_block, "request_id": raw_json.get("request_id"), "transaction_id": raw_json.get("transaction_id"), } return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=normalized_result, raw_response=raw_json, ) # Failure / Error handling is_server_error = response.status_code >= 500 if error_block: err_code = str(error_block.get("code") or f"HTTP_{response.status_code}") err_message = error_block.get("message") or "MSME by PAN fetch failed" elif data_block: err_code = str(data_block.get("code") or f"HTTP_{response.status_code}") err_message = data_block.get("message") or "MSME by PAN fetch failed" else: err_code = f"HTTP_{response.status_code}" err_message = raw_json.get("message") or "MSME by PAN fetch failed" logger.warning( "Codeplus Fetch MSME by PAN failed: status_code=%d, code=%s, message=%s", response.status_code, err_code, err_message, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR if is_server_error else VerificationStatus.FAILED, data=data_block, raw_response=raw_json, error_code=err_code, error_message=err_message, retryable=is_server_error or response.status_code in (429, 408, 401, 403), ) except requests.exceptions.Timeout as exc: logger.warning("Codeplus Fetch MSME by PAN timed out: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="TIMEOUT", error_message=f"Request to Codeplus Fetch MSME by PAN timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: logger.warning("Codeplus Fetch MSME by PAN network exception: %s", exc) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code="NETWORK_ERROR", error_message=f"Network error connecting to Codeplus Fetch MSME API: {exc}", retryable=True, ) def verify_generic(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: endpoint = f"{self.base_url}/v4/document-verify" return self._send_generic_request(endpoint, payload=data, doc_label="Generic Document") def _send_generic_request( self, endpoint: str, payload: Dict[str, Any], doc_label: str = "Document", ) -> VendorResult: """Generic fallback dispatcher for endpoints awaiting specific contracts.""" headers = self.get_headers() clean_payload = {k: v for k, v in payload.items() if v is not None} try: response = self.session.post( endpoint, json=clean_payload, headers=headers, timeout=self.timeout, ) raw_json = response.json() if response.content else {} if response.status_code == 200: return VendorResult( vendor_name=self.name, is_success=True, status=VerificationStatus.SUCCESS, data=raw_json.get("data", raw_json), raw_response=raw_json, ) return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.FAILED if response.status_code < 500 else VerificationStatus.ERROR, raw_response=raw_json, error_code=f"HTTP_{response.status_code}", error_message=raw_json.get("message", f"Request failed with HTTP {response.status_code}"), retryable=response.status_code >= 500, ) except Exception as exc: return VendorResult( vendor_name=self.name, is_success=False, status=VerificationStatus.ERROR, error_code=exc.__class__.__name__, error_message=str(exc), retryable=True, )