diff --git a/docverify/settings.py b/docverify/settings.py index 8f7d773..6f199f4 100644 --- a/docverify/settings.py +++ b/docverify/settings.py @@ -116,7 +116,8 @@ REST_FRAMEWORK = { # Vendor API Configurations # ────────────────────────────────────────────────────────────────────────────── # Codeplus API Credentials -CODEPLUS_BASE_URL = os.environ.get("CODEPLUS_BASE_URL", "https://api.codeplus.in") -CODEPLUS_API_KEY = os.environ.get("CODEPLUS_API_KEY", "") -CODEPLUS_CLIENT_ID = os.environ.get("CODEPLUS_CLIENT_ID", "") -CODEPLUS_CLIENT_SECRET = os.environ.get("CODEPLUS_CLIENT_SECRET", "") +CODEPLUS_BASE_URL = os.environ.get("CODEPLUS_BASE_URL", "https://apihub.services") +CODEPLUS_CLIENT_API_ID = os.environ.get("CODEPLUS_CLIENT_API_ID", "CAPI-YYR0YJIE0VMTDENY") +CODEPLUS_CLIENT_API_SECRET = os.environ.get("CODEPLUS_CLIENT_API_SECRET", "s24Ts9wFsgl64mFzA61OqG0RDLRs2HFCpmOZBmqs") + + diff --git a/verifier/engine/base.py b/verifier/engine/base.py index 9e4ed1b..b81a1fd 100644 --- a/verifier/engine/base.py +++ b/verifier/engine/base.py @@ -6,15 +6,27 @@ from typing import Any, Dict, List, Optional class DocumentType(str, Enum): PAN = "pan" + PAN_V5 = "pan_v5" + PAN_VERIFY = "pan_verify" + PAN_ADVANCED = "pan_advanced" + PAN_DETAILED = "pan_detailed" AADHAAR = "aadhaar" DRIVING_LICENSE = "driving_license" VOTER_ID = "voter_id" PASSPORT = "passport" BANK_ACCOUNT = "bank_account" GSTIN = "gstin" + GST = "gst" + GST_VERIFY = "gst_verify" + GST_BY_PAN = "gst_by_pan" + FETCH_GST = "fetch_gst" + MSME = "msme" + MSME_BY_PAN = "msme_by_pan" + UDYAM = "udyam" CUSTOM = "custom" + class VerificationStatus(str, Enum): SUCCESS = "SUCCESS" FAILED = "FAILED" diff --git a/verifier/tests.py b/verifier/tests.py index b9d7fc5..4a41ef5 100644 --- a/verifier/tests.py +++ b/verifier/tests.py @@ -104,7 +104,13 @@ class CodeplusVendorTests(TestCase): """Tests Codeplus vendor adapter configuration and handling.""" def setUp(self): - self.codeplus = CodeplusVendor(config={"base_url": "https://api.codeplus.in", "api_key": "test_key"}) + self.codeplus = CodeplusVendor( + config={ + "base_url": "https://apihub.services", + "client_api_id": "CAPI-YYR0YJIE0VMTDENY", + "client_api_secret": "test_secret_123", + } + ) def test_codeplus_metadata(self): self.assertEqual(self.codeplus.name, "codeplus") @@ -114,8 +120,397 @@ class CodeplusVendorTests(TestCase): def test_codeplus_headers(self): headers = self.codeplus.get_headers() - self.assertEqual(headers["Authorization"], "Bearer test_key") - self.assertEqual(headers["x-api-key"], "test_key") + self.assertEqual(headers["X-Client-Api-ID"], "CAPI-YYR0YJIE0VMTDENY") + self.assertEqual(headers["X-Client-Api-Secret"], "test_secret_123") + self.assertEqual(headers["Content-Type"], "application/json") + + def test_codeplus_pan_v5_success(self): + """Test parsing valid PAN response via v5 Recommended API.""" + mock_response_payload = { + "success": True, + "message": "PAN verification completed", + "data": { + "transactionId": "TXN_1787136465045_demo12345", + "service": "PAN_VERIFICATION", + "success": True, + "resultCode": 101, + "status": "SUCCESS", + "message": "PAN verified successfully", + "providerTransactionId": "TXN_1787136465045_demo12345", + "pan": "ABCDE1234F", + "panType": "Individual", + "fullName": "RAHUL KUMAR", + "firstName": "RAHUL", + "middleName": None, + "lastName": "KUMAR", + "gender": "male", + "dob": "15/06/1995", + "aadhaarNumber": "XXXXXXXX1234", + "aadhaarLinked": True, + "addressDetails": { + "building_name": "12/45", + "locality": "Central Area", + "street_name": "MG Road", + "pincode": "641001", + "city": "COIMBATORE", + "state": "Tamil Nadu", + "country": "India", + }, + "clientRefNum": "REQ_1787136465034_demo12345", + }, + "requestId": "REQ_1787136465034_demo12345", + "timestamp": "2026-08-19T10:47:45.735Z", + "statusCode": 200, + "status": "SUCCESS", + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("pan", {"pan": "ABCDE1234F"}) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertEqual(result.data["full_name"], "RAHUL KUMAR") + self.assertEqual(result.data["dob"], "15/06/1995") + self.assertEqual(result.data["pan_type"], "Individual") + self.assertEqual(result.data["address_details"]["city"], "COIMBATORE") + self.assertTrue(result.data["aadhaar_linked"]) + + def test_codeplus_pan_advanced_success(self): + """Test parsing valid PAN response via v4 Advanced API (HTTP 200, code 1000).""" + mock_response_payload = { + "request_id": "8a42f1c7-63d9-4e15-b820-7c91a5d34628", + "transaction_id": "d5b73e91-2a46-4c08-9f61-83e2b7a51439", + "status": 200, + "data": { + "code": "1000", + "message": "Valid PAN.", + "pan_data": { + "document_type": "PAN", + "document_id": "ABCDE1234F", + "name": "RAHUL KUMAR", + "first_name": "RAHUL", + "last_name": "KUMAR", + "category": "P", + "category_type": "Individual", + "is_individual": True, + "individual_tax_compliance_status": "Operative", + "aadhaar_linked": True, + }, + }, + "timestamp": 1787031203055, + "path": "/pan-api/fetch-advanced", + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("pan_advanced", {"pan": "ABCDE1234F"}) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertEqual(result.data["name"], "RAHUL KUMAR") + self.assertEqual(result.data["document_id"], "ABCDE1234F") + self.assertTrue(result.data["aadhaar_linked"]) + + def test_codeplus_pan_advanced_error_nonexistent(self): + """Test parsing non-existent PAN (code 1004).""" + mock_error_payload = { + "request_id": "23bc5fa5-a02e-4246-aefc-3c455bcd1853", + "transaction_id": "542f956c-db4f-48f5-8d6b-d1877a0b930a", + "status": 200, + "data": { + "code": "1004", + "message": "PAN does not exist.", + }, + "timestamp": 1787031223795, + "path": "/pan-api/fetch-advanced", + } + + class MockResponse: + status_code = 400 + def json(self): + return mock_error_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("pan_advanced", {"pan": "INVALID999Z"}) + + self.assertFalse(result.is_success) + self.assertEqual(result.error_code, "1004") + self.assertEqual(result.error_message, "PAN does not exist.") + + def test_codeplus_pan_detailed_success(self): + """Test parsing valid Detailed PAN response (HTTP 200, code 1000 with address/email/phone).""" + mock_response_payload = { + "request_id": "7f3a1c92-5b84-4d21-a7e6-91c2f0d84b56", + "transaction_id": "b6e8d4a1-2c73-49f5-9a10-5f7c3e821d64", + "status": 200, + "data": { + "code": "1000", + "message": "Valid PAN.", + "pan_data": { + "document_type": "PAN", + "document_id": "ABCDE1234F", + "name": "RAHUL KUMAR", + "first_name": "RAHUL", + "last_name": "KUMAR", + "category": "P", + "date_of_birth": "1995-06-15", + "masked_aadhaar_number": "XXXXXXXX1234", + "address_data": { + "line_1": "12/45", + "city": "COIMBATORE", + "state": "TAMIL NADU", + "pincode": "641001", + }, + "email": "ra******@example.com", + "phone": "98XXXXXX10", + "gender": "MALE", + "aadhaar_linked": True, + "taxpayer_profile": True, + }, + }, + "timestamp": 1787028660489, + "path": "/pan-api/fetch-detailed", + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("pan_detailed", {"pan_number": "ABCDE1234F"}) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertEqual(result.data["name"], "RAHUL KUMAR") + self.assertEqual(result.data["email"], "ra******@example.com") + self.assertEqual(result.data["phone"], "98XXXXXX10") + self.assertEqual(result.data["date_of_birth"], "1995-06-15") + self.assertEqual(result.data["address_data"]["city"], "COIMBATORE") + + def test_codeplus_pan_detailed_error(self): + """Test parsing invalid Detailed PAN error (HTTP 400 with error block).""" + mock_error_payload = { + "request_id": "b89afa37-8282-4cf7-827f-985700f84ab8", + "transaction_id": "d796efe7-26e2-41c0-a1e7-2f7387b2d5aa", + "status": 400, + "error": { + "code": "INVALID_PAN", + "message": "Invalid PAN number.", + }, + "timestamp": 1787028631589, + "path": "/pan-api/fetch-detailed", + } + + class MockResponse: + status_code = 400 + def json(self): + return mock_error_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("pan_detailed", {"pan_number": "INVALID999Z"}) + + self.assertFalse(result.is_success) + self.assertEqual(result.error_code, "INVALID_PAN") + self.assertEqual(result.error_message, "Invalid PAN number.") + + def test_codeplus_bank_verify_success(self): + """Test successful Bank Account verification.""" + mock_response_payload = { + "request_id": "b1a2c3d4-e5f6-7890-abcd-ef1234567890", + "transaction_id": "tx-bank-998877", + "status": 200, + "data": { + "code": "1000", + "message": "Bank account verified successfully.", + "bank_data": { + "account_number": "123456789012", + "ifsc": "HDFC0001234", + "name_at_bank": "PRIYA SHARMA", + "micr": "641240002", + "bank_name": "HDFC BANK", + "branch": "MAIN BRANCH", + "city": "COIMBATORE", + "state": "TAMIL NADU", + "pincode": "641001", + "account_exists": True, + }, + }, + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify( + "bank_account", + {"account_number": "123456789012", "ifsc": "HDFC0001234"}, + ) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertEqual(result.data["beneficiary_name"], "PRIYA SHARMA") + self.assertEqual(result.data["bank_name"], "HDFC BANK") + self.assertEqual(result.data["micr"], "641240002") + self.assertEqual(result.data["city"], "COIMBATORE") + + def test_codeplus_bank_verify_missing_fields(self): + """Test bank verification validation for missing account or IFSC.""" + result_missing_acc = self.codeplus.verify("bank_account", {"ifsc": "HDFC0001234"}) + self.assertFalse(result_missing_acc.is_success) + self.assertEqual(result_missing_acc.error_code, "MISSING_ACCOUNT_NUMBER") + + result_missing_ifsc = self.codeplus.verify("bank_account", {"account_number": "123456"}) + self.assertFalse(result_missing_ifsc.is_success) + self.assertEqual(result_missing_ifsc.error_code, "MISSING_IFSC") + + def test_codeplus_fetch_gst_success(self): + """Test successful fetch GST by PAN.""" + mock_response_payload = { + "request_id": "b27c9087-f061-4d38-b7aa-96498f8e667e", + "status": 200, + "data": { + "code": "1002", + "message": "Found 3 records", + "results": [ + { + "document_type": "GSTIN", + "document_id": "20AXXXXXXXXXXX", + "status": "Active", + "state": "Jharkhand", + "state_code": "20", + }, + { + "document_type": "GSTIN", + "document_id": "27AXXXXXXXXXXX", + "status": "Active", + "state": "Maharashtra", + "state_code": "27", + }, + { + "document_type": "GSTIN", + "document_id": "29AXXXXXXXXXXX", + "status": "Active", + "state": "Karnataka", + "state_code": "29", + }, + ], + }, + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("gst_by_pan", {"pan": "ABCDE1234F"}) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertEqual(result.data["records_count"], 3) + self.assertEqual(len(result.data["results"]), 3) + self.assertEqual(result.data["results"][0]["document_id"], "20AXXXXXXXXXXX") + + def test_codeplus_fetch_gst_missing_pan(self): + """Test validation for missing PAN in GST by PAN.""" + result = self.codeplus.verify("gst_by_pan", {}) + self.assertFalse(result.is_success) + self.assertEqual(result.error_code, "MISSING_PAN") + + def test_codeplus_fetch_msme_no_record(self): + """Test MSME by PAN returning code 1011 (No record found).""" + mock_response_payload = { + "request_id": "cd7a7566-3122-4942-a110-d2b0ab81224d", + "transaction_id": "87186c96-f51d-436b-9fce-3a9d1ab8c44c", + "status": 200, + "data": { + "code": "1011", + "message": "No record found.", + }, + "timestamp": 1741349527562, + "path": "/msme-api/udyam/fetch-by-pan", + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("msme", {"pan": "ABCDE1234F"}) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertFalse(result.data["has_msme"]) + self.assertEqual(result.data["code"], "1011") + self.assertEqual(result.data["message"], "No record found.") + + def test_codeplus_fetch_msme_missing_pan(self): + """Test validation for missing PAN in MSME.""" + result = self.codeplus.verify("msme", {}) + self.assertFalse(result.is_success) + self.assertEqual(result.error_code, "MISSING_PAN") + + def test_codeplus_gst_verify_v5_success(self): + """Test successful GST verification v5.""" + mock_response_payload = { + "success": True, + "message": "GST verification completed", + "data": { + "service": "GST_VERIFICATION", + "gstNumber": "27ABCDE1234F1Z5", + "legalName": "ABCDE PRIVATE LIMITED", + "registrationStatus": "ACTIVE", + }, + "statusCode": 200, + } + + class MockResponse: + status_code = 200 + def json(self): + return mock_response_payload + + from unittest.mock import patch + with patch.object(self.codeplus.session, "post", return_value=MockResponse()): + result = self.codeplus.verify("gstin", {"gst_number": "27ABCDE1234F1Z5"}) + + self.assertTrue(result.is_success) + self.assertEqual(result.status, VerificationStatus.SUCCESS) + self.assertEqual(result.data["gst_number"], "27ABCDE1234F1Z5") + self.assertEqual(result.data["legal_name"], "ABCDE PRIVATE LIMITED") + self.assertEqual(result.data["registration_status"], "ACTIVE") + + def test_codeplus_gst_verify_v5_missing_gst(self): + """Test validation for missing GST number.""" + result = self.codeplus.verify("gstin", {}) + self.assertFalse(result.is_success) + self.assertEqual(result.error_code, "MISSING_GST_NUMBER") + + + + + + class VerificationAPITests(TestCase): diff --git a/verifier/vendors/codeplus.py b/verifier/vendors/codeplus.py index 032f7a1..61513cc 100644 --- a/verifier/vendors/codeplus.py +++ b/verifier/vendors/codeplus.py @@ -16,9 +16,9 @@ logger = logging.getLogger(__name__) class CodeplusVendor(BaseVendor): """ - Codeplus Verification Adapter. - Implements verification workflows for PAN, Aadhaar, Driving License, Voter ID, - Bank Accounts, GSTIN, and custom documents via Codeplus APIs. + 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" @@ -27,58 +27,69 @@ class CodeplusVendor(BaseVendor): 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.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://api.codeplus.in") + or os.environ.get("CODEPLUS_BASE_URL", "https://apihub.services") ).rstrip("/") - self.api_key = ( - self.config.get("api_key") - or getattr(settings, "CODEPLUS_API_KEY", None) - or os.environ.get("CODEPLUS_API_KEY", "") + # 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_id = ( - self.config.get("client_id") - or getattr(settings, "CODEPLUS_CLIENT_ID", None) - or os.environ.get("CODEPLUS_CLIENT_ID", "") + 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.client_secret = ( - self.config.get("client_secret") - or getattr(settings, "CODEPLUS_CLIENT_SECRET", None) - or os.environ.get("CODEPLUS_CLIENT_SECRET", "") - ) - - self.timeout = int(self.config.get("timeout", 15)) + self.timeout = int(self.config.get("timeout", 20)) self.session = requests.Session() def get_headers(self) -> Dict[str, str]: - """Construct standard HTTP headers for Codeplus API calls.""" + """ + 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.api_key: - headers["Authorization"] = f"Bearer {self.api_key}" - headers["x-api-key"] = self.api_key - - if self.client_id: - headers["x-client-id"] = self.client_id - if self.client_secret: - headers["x-client-secret"] = self.client_secret + 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 @@ -93,219 +104,1065 @@ class CodeplusVendor(BaseVendor): """ 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.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) # ────────────────────────────────────────────────────────────────────────── - # Document-Specific Verification Methods - # (Structured for easy payload/endpoint mapping once examples are provided) + # PAN Verification (v5 Recommended, v4 Advanced, & v4 Detailed APIs) # ────────────────────────────────────────────────────────────────────────── def verify_pan(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ - PAN Card Verification. - Expected data inputs: - - pan_number / pan: PAN number (e.g. ABCDE1234F) - - name / full_name: (Optional) Name to compare - - dob: (Optional) Date of Birth (YYYY-MM-DD or DD/MM/YYYY) + 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') """ - endpoint = f"{self.base_url}/v1/pan/verify" - payload = { - "pan": data.get("pan_number") or data.get("pan"), - "name": data.get("name") or data.get("full_name"), - "dob": data.get("dob"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="PAN") + mode = str(data.get("mode", "")).lower() + version = str(data.get("version", "")).lower() - def verify_aadhaar(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: - """ - Aadhaar Verification (OTP generation, verification, or OCR). - """ - endpoint = f"{self.base_url}/v1/aadhaar/verify" - payload = { - "aadhaar_number": data.get("aadhaar_number") or data.get("aadhaar"), - "otp": data.get("otp"), - "client_ref_id": data.get("client_ref_id"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="Aadhaar") + 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) - def verify_driving_license(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: - """ - Driving License Verification. - """ - endpoint = f"{self.base_url}/v1/dl/verify" - payload = { - "dl_number": data.get("dl_number") or data.get("license_number"), - "dob": data.get("dob"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="Driving License") + # Default to v5 Recommended API + v5_result = self.verify_pan_v5(data, files) + if v5_result.is_success: + return v5_result - def verify_voter_id(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: - """ - Voter ID (EPIC) Verification. - """ - endpoint = f"{self.base_url}/v1/voter-id/verify" - payload = { - "epic_number": data.get("epic_number") or data.get("voter_id"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="Voter ID") + # 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 - def verify_passport(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: - """ - Passport Verification. - """ - endpoint = f"{self.base_url}/v1/passport/verify" - payload = { - "passport_number": data.get("passport_number"), - "dob": data.get("dob"), - "file_number": data.get("file_number"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="Passport") + return v5_result - def verify_bank_account(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: + def verify_pan_v5(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: """ - Bank Account / Penny Drop Verification. + 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}/v1/bank/verify" - payload = { - "account_number": data.get("account_number") or data.get("bank_account"), - "ifsc": data.get("ifsc"), - "name": data.get("name"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="Bank Account") + 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() - def verify_gstin(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: - """ - GSTIN Business Verification. - """ - endpoint = f"{self.base_url}/v1/gstin/verify" - payload = { - "gstin": data.get("gstin"), - } - return self._send_request(endpoint, payload=payload, files=files, doc_label="GSTIN") + 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, + ) - def verify_generic(self, data: Dict[str, Any], files: Optional[Dict[str, Any]] = None) -> VendorResult: - """Generic document verification fallback.""" - endpoint = f"{self.base_url}/v1/document/verify" - return self._send_request(endpoint, payload=data, files=files, doc_label="Generic Document") - - # ────────────────────────────────────────────────────────────────────────── - # Internal HTTP Dispatcher & Response Normalizer - # ────────────────────────────────────────────────────────────────────────── - - def _send_request( - self, - endpoint: str, - payload: Dict[str, Any], - files: Optional[Dict[str, Any]] = None, - doc_label: str = "Document", - ) -> VendorResult: - """ - Executes HTTP call to Codeplus and maps response into standardized VendorResult. - Catches errors to trigger fallback waterfall if Codeplus fails. - """ + payload = {"pan": pan_number} headers = self.get_headers() - # Clean payload of None values - clean_payload = {k: v for k, v in payload.items() if v is not None} - try: - if files: - headers.pop("Content-Type", None) - response = self.session.post( - endpoint, - data=clean_payload, - files=files, - headers=headers, - timeout=self.timeout, - ) - else: - response = self.session.post( - endpoint, - json=clean_payload, - headers=headers, - timeout=self.timeout, - ) + response = self.session.post( + endpoint, + json=payload, + headers=headers, + timeout=self.timeout, + ) try: raw_json = response.json() except Exception: - raw_json = {"text": response.text} + raw_json = {"text": response.text, "status_code": response.status_code} - # Check HTTP status - if response.status_code >= 400: - error_msg = ( - raw_json.get("message") - or raw_json.get("error") - or f"Codeplus HTTP {response.status_code} Error" - ) - logger.warning( - "Codeplus %s verification failed with HTTP %d: %s", - doc_label, - response.status_code, - error_msg, - ) - 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=str(error_msg), - retryable=response.status_code >= 500 or response.status_code in (429, 408), - ) + 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 {} - # Codeplus success validation (can be tuned once exact example response is shared) - status_field = str(raw_json.get("status", "")).upper() - is_successful = status_field in ("SUCCESS", "VALID", "COMPLETED", "TRUE", "1") or raw_json.get("valid") is True + 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_successful or ("data" in raw_json and "error" not in raw_json): - extracted_data = raw_json.get("data") if isinstance(raw_json.get("data"), dict) else raw_json + 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=extracted_data, + data=normalized_result, raw_response=raw_json, ) - else: - msg = raw_json.get("message") or raw_json.get("error") or "Document verification returned invalid status" - return VendorResult( - vendor_name=self.name, - is_success=False, - status=VerificationStatus.FAILED, - data=raw_json.get("data", {}), - raw_response=raw_json, - error_code="INVALID_STATUS", - error_message=str(msg), - retryable=False, - ) + + # 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 %s call timed out: %s", doc_label, 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 timed out: {exc}", + error_message=f"Request to Codeplus PAN Verification (v5) timed out: {exc}", retryable=True, ) except requests.exceptions.RequestException as exc: - logger.warning("Codeplus %s network exception: %s", doc_label, 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: {exc}", + 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.") + 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"), + "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 Verification (Full Details) 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, )