vickytechkey
4e1ea0fcf2
All checks were successful
Deploy Beta (NATIVE) / deploy (push) Successful in 34s
1170 lines
49 KiB
Python
1170 lines
49 KiB
Python
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.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.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": "<PAN_NUMBER>"
|
|
}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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": "<PAN_NUMBER>"}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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": "<PAN_NUMBER>"}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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 Verification (Full Details) API:
|
|
Endpoint: POST https://apihub.services/v4/bank-verify
|
|
Payload:
|
|
{
|
|
"account_number": "<ACCOUNT_NUMBER>",
|
|
"ifsc": "<IFSC_CODE>"
|
|
}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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": "<PAN_NUMBER>"
|
|
}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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": "<GSTIN>"
|
|
}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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": "<PAN_NUMBER>",
|
|
"detailed_response": true/false # (optional)
|
|
}
|
|
Headers:
|
|
X-Client-Api-ID: <id>
|
|
X-Client-Api-Secret: <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,
|
|
)
|