pan_verification.py•3.87 kB
"""PAN verification tool implementation."""
import hashlib
import time
from typing import Any, Dict
from pydantic import ValidationError as PydanticValidationError
from src.clients.kyc_api_client import KYCAPIClient
from src.models.requests import PANVerificationRequest
from src.models.responses import PANVerificationResponse
from src.tools.base_tool import BaseTool
from src.utils.logger import get_logger
logger = get_logger(__name__)
class PANVerificationTool(BaseTool):
"""
Tool for PAN card verification with name and DOB matching.
"""
def __init__(self, api_client: KYCAPIClient):
"""
Initialize PAN verification tool.
Args:
api_client: KYC API client instance
"""
self.api_client = api_client
def get_name(self) -> str:
"""Return tool name."""
return "verify_pan"
async def execute(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""
Execute PAN verification.
Args:
params: Dictionary containing:
- pan: 10-character PAN number
- name_as_per_pan: Full name as per PAN card
- date_of_birth: Date of birth in DD/MM/YYYY format
- consent: User consent ('Y' or 'y')
- reason: Reason for verification
Returns:
Verification result with match status
"""
logger.info("pan_verification_started", pan=params.get("pan"))
try:
# Validate input using Pydantic model
request = PANVerificationRequest(**params)
# Prepare API request payload
api_payload = {
"@entity": "in.co.sandbox.kyc.pan_verification.request",
"pan": request.pan,
"name_as_per_pan": request.name_as_per_pan,
"date_of_birth": request.date_of_birth,
"consent": request.consent,
"reason": request.reason,
}
# Call KYC API
response = await self.api_client.post(
endpoint="/kyc/pan/verify", data=api_payload
)
# Parse and validate response
data = response.get("data", {})
result = PANVerificationResponse(**data)
# Format response
output = {
"pan": result.pan,
"category": result.category,
"status": result.status,
"remarks": result.remarks,
"name_match": result.name_as_per_pan_match,
"dob_match": result.date_of_birth_match,
"aadhaar_seeding_status": result.aadhaar_seeding_status,
"verified_at": int(time.time()),
}
logger.info(
"pan_verification_completed",
pan=result.pan,
status=result.status,
name_match=result.name_as_per_pan_match,
)
return output
except PydanticValidationError as e:
logger.error("pan_verification_validation_error", error=str(e))
raise ValueError(f"Invalid input parameters: {str(e)}")
except Exception as e:
logger.error("pan_verification_failed", error=str(e), exc_info=True)
raise
def get_cache_key(self, params: Dict[str, Any]) -> str:
"""
Generate cache key from PAN, name, and DOB.
Args:
params: Input parameters
Returns:
Cache key
"""
key_data = f"{params.get('pan')}|{params.get('name_as_per_pan')}|{params.get('date_of_birth')}"
hash_key = hashlib.sha256(key_data.encode()).hexdigest()
return f"pan_verify:{hash_key}"
def get_cache_ttl(self) -> int:
"""Get cache TTL for PAN verification."""
return 3600 # 1 hour