import logging import re import time from difflib import SequenceMatcher from rest_framework import status from rest_framework.response import Response from rest_framework.views import APIView # Import engine and vendors to ensure default registration import verifier.vendors # noqa: F401 from verifier.engine import engine, registry from .models import VerificationTransaction, VendorAttemptLog from .serializers import ( VerifyDocumentRequestSerializer, VerifySellerPanGstRequestSerializer, VerificationTransactionSerializer, ) logger = logging.getLogger(__name__) def normalize_name(name: str) -> str: """Normalizes names by stripping honorifics, titles, special characters, and excess whitespace.""" if not name: return "" name = name.upper().strip() for p in [r"\bMR\.?\b", r"\bMRS\.?\b", r"\bMS\.?\b", r"\bSHRI\b", r"\bSMT\.?\b", r"\bDR\.?\b", r"\bM/S\b"]: name = re.sub(p, "", name, flags=re.IGNORECASE) name = re.sub(r"[^A-Z\s]", " ", name) return " ".join(name.split()) def compare_names(name1: str, name2: str) -> dict: """ Intelligent name matching for Indian KYC. Matches exact, permuted, initials, and fuzzy token sets while rejecting different first names with common surnames. """ norm1, norm2 = normalize_name(name1), normalize_name(name2) if not norm1 or not norm2: return {"matched": False, "score": 0.0, "reason": "Missing or empty name"} if norm1 == norm2: return {"matched": True, "score": 100.0, "reason": "Exact name match"} words1, words2 = norm1.split(), norm2.split() # Reject complete first name mismatch (e.g. PRIYA SHARMA vs ROHIT SHARMA) if len(words1) >= 2 and len(words2) >= 2: if words1[0] != words2[0] and words1[-1] != words2[0] and words1[0] != words2[-1]: if len(words1[0]) >= 3 and len(words2[0]) >= 3: fn_sim = SequenceMatcher(None, words1[0], words2[0]).ratio() if fn_sim < 0.65: return { "matched": False, "score": round(fn_sim * 100, 1), "reason": f"First name mismatch ('{words1[0]}' vs '{words2[0]}')", } set1, set2 = set(words1), set(words2) if set1 == set2: return {"matched": True, "score": 98.0, "reason": "Permuted words match"} common = set1.intersection(set2) if len(common) == min(len(set1), len(set2)): return {"matched": True, "score": 95.0, "reason": "Subset token match"} # Initial check (e.g. 'RAHUL K' vs 'RAHUL KUMAR' or 'V SANKAR' vs 'VIGNESH SANKAR') if len(words1) == len(words2): initial_match = True for w1, w2 in zip(words1, words2): if w1 != w2 and not (len(w1) == 1 and w2.startswith(w1)) and not (len(w2) == 1 and w1.startswith(w2)): initial_match = False break if initial_match: return {"matched": True, "score": 92.0, "reason": "Initials abbreviation match"} sort_ratio = SequenceMatcher(None, " ".join(sorted(words1)), " ".join(sorted(words2))).ratio() * 100 seq_ratio = SequenceMatcher(None, norm1, norm2).ratio() * 100 best_score = max(sort_ratio, seq_ratio) matched = best_score >= 80.0 return { "matched": matched, "score": round(best_score, 1), "reason": "Fuzzy token match" if matched else "Name similarity score below 80% threshold", } def mask_sensitive_payload(payload: dict) -> dict: """Mask sensitive inputs like full aadhaar numbers or passwords for audit safety.""" masked = dict(payload) sensitive_keys = ["aadhaar_number", "aadhaar", "otp", "password", "client_secret"] for key in sensitive_keys: if key in masked and isinstance(masked[key], str): val = masked[key] if len(val) > 4: masked[key] = f"{val[:2]}****{val[-2:]}" else: masked[key] = "****" return masked class DocumentVerifyView(APIView): """ POST /api/verify/ Verifies a document using the vendor waterfall failover engine. If the first vendor encounters an error/failure, the engine automatically cascades to the next available vendor. """ def post(self, request, *args, **kwargs): serializer = VerifyDocumentRequestSerializer(data=request.data) if not serializer.is_valid(): return Response( {"is_success": False, "errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST, ) doc_type = serializer.validated_data["document_type"] input_data = serializer.validated_data["data"] preferred_vendor = serializer.validated_data.get("preferred_vendor") files = request.FILES if hasattr(request, "FILES") else None # Execute waterfall verification engine_result = engine.verify( document_type=doc_type, data=input_data, files=files, preferred_vendor=preferred_vendor, ) # Persist transaction and audit attempts in DB safe_input = mask_sensitive_payload(input_data) transaction = VerificationTransaction.objects.create( document_type=doc_type, status=engine_result.status.value, successful_vendor=engine_result.successful_vendor, input_payload=safe_input, final_data=engine_result.data if engine_result.is_success else {}, attempts_count=len(engine_result.attempts), total_latency_ms=engine_result.total_latency_ms, ) # Persist attempt logs for attempt in engine_result.attempts: VendorAttemptLog.objects.create( transaction=transaction, attempt_number=attempt.get("attempt_number", 1), vendor_name=attempt.get("vendor_name", "unknown"), status=attempt.get("status", "UNKNOWN"), is_success=attempt.get("is_success", False), data=attempt.get("data", {}), raw_response=attempt.get("raw_response"), error_code=attempt.get("error_code"), error_message=attempt.get("error_message"), latency_ms=attempt.get("latency_ms", 0.0), ) response_payload = { "reference_id": str(transaction.reference_id), "is_success": engine_result.is_success, "status": engine_result.status.value, "document_type": doc_type, "successful_vendor": engine_result.successful_vendor, "data": engine_result.data, "attempts_count": len(engine_result.attempts), "attempts": engine_result.attempts, "error_message": engine_result.error_message, "total_latency_ms": round(engine_result.total_latency_ms, 2), } # Store complete client response payload in database record transaction.response_payload = response_payload transaction.save(update_fields=["response_payload"]) http_status = status.HTTP_200_OK if engine_result.is_success else status.HTTP_422_UNPROCESSABLE_ENTITY return Response(response_payload, status=http_status) class SellerPanGstVerifyView(APIView): """ POST /api/verify/seller/ Composite verification endpoint for seller onboarding: 1. Verifies the seller's PAN against government / vendor records. 2. Validates that the seller's entered full name matches the verified PAN cardholder name. 3. Fetches all active GST registrations linked to that PAN. 4. Automatically persists the full verification transaction and vendor attempt logs in the database. """ def post(self, request, *args, **kwargs): serializer = VerifySellerPanGstRequestSerializer(data=request.data) if not serializer.is_valid(): return Response( {"is_success": False, "errors": serializer.errors}, status=status.HTTP_400_BAD_REQUEST, ) seller_name = serializer.validated_data["seller_name"].strip() pan = serializer.validated_data["pan"].strip().upper() entered_gstin = (serializer.validated_data.get("gstin") or "").strip().upper() preferred_vendor = serializer.validated_data.get("preferred_vendor") start_time = time.time() # Step 1: Verify PAN via waterfall engine pan_result = engine.verify( document_type="pan", data={"pan": pan}, preferred_vendor=preferred_vendor, ) all_attempts = list(pan_result.attempts) if not pan_result.is_success: total_latency = (time.time() - start_time) * 1000 tx = VerificationTransaction.objects.create( document_type="SELLER_PAN_GST", status="FAILED", successful_vendor=None, input_payload={"seller_name": seller_name, "pan": pan, "gstin": entered_gstin}, final_data={}, attempts_count=len(all_attempts), total_latency_ms=total_latency, ) for att in all_attempts: VendorAttemptLog.objects.create( transaction=tx, attempt_number=att.get("attempt_number", 1), vendor_name=att.get("vendor_name", "unknown"), status=att.get("status", "UNKNOWN"), is_success=att.get("is_success", False), data=att.get("data", {}), raw_response=att.get("raw_response"), error_code=att.get("error_code"), error_message=att.get("error_message"), latency_ms=att.get("latency_ms", 0.0), ) res_payload = { "reference_id": str(tx.reference_id), "is_success": False, "status": "FAILED", "document_type": "SELLER_PAN_GST", "error_message": f"PAN verification failed: {pan_result.error_message}", "name_matched": False, "gst_records": [], "attempts": all_attempts, "total_latency_ms": round(total_latency, 2), } tx.response_payload = res_payload tx.save(update_fields=["response_payload"]) return Response(res_payload, status=status.HTTP_422_UNPROCESSABLE_ENTITY) # PAN is valid, extract verified name pan_data = pan_result.data or {} verified_pan_name = ( pan_data.get("name") or pan_data.get("full_name") or pan_data.get("name_on_card") or "" ) # Step 2: Compare seller name with verified PAN cardholder name match_result = compare_names(seller_name, verified_pan_name) # Step 3: Pull GST linked with the PAN from PAN number gst_result = engine.verify( document_type="fetch_gst", data={"pan": pan}, preferred_vendor=preferred_vendor, ) for att in gst_result.attempts: att["attempt_number"] = len(all_attempts) + 1 all_attempts.append(att) gst_records = [] if gst_result.is_success and gst_result.data: gst_records = gst_result.data.get("results") or gst_result.data.get("gstins") or [] # Find active GSTINs active_gstins = [ (g.get("document_id") or g.get("gstin")) for g in gst_records if isinstance(g, dict) and g.get("status", "").lower() in ["active", "valid"] ] primary_gstin = active_gstins[0] if active_gstins else ( (gst_records[0].get("document_id") or gst_records[0].get("gstin")) if gst_records and isinstance(gst_records[0], dict) else None ) target_gstin = entered_gstin or primary_gstin or "" gstin_matched = False gst_details = {} verified_gst_name = "" gst_name_match = {"matched": True, "score": 100.0, "reason": "No GST provided"} gst_error = None if target_gstin: # Check if entered GSTIN exists in fetched GST records for this PAN matching_gst_record = None for g in gst_records: if isinstance(g, dict): doc_id = (g.get("document_id") or g.get("gstin") or "").upper() if doc_id == target_gstin: matching_gst_record = g gstin_matched = True break # Validate that 15-character GSTIN contains the 10-character PAN at chars 3-12 if len(target_gstin) == 15 and target_gstin[2:12].upper() != pan.upper(): gstin_matched = False gst_error = f"GSTIN '{target_gstin}' does not match the provided PAN card '{pan}'." # Check if matching record has legal_name if matching_gst_record and matching_gst_record.get("legal_name"): verified_gst_name = matching_gst_record.get("legal_name") gst_details = matching_gst_record else: # Call gst_verify to get legal_name and active status try: gst_verify_result = engine.verify( document_type="gst_verify", data={"gst_number": target_gstin}, preferred_vendor=preferred_vendor, ) for att in gst_verify_result.attempts: att["attempt_number"] = len(all_attempts) + 1 all_attempts.append(att) if gst_verify_result.is_success and gst_verify_result.data: gst_details = gst_verify_result.data.get("gst_details") or gst_verify_result.data verified_gst_name = ( gst_verify_result.data.get("legal_name") or gst_details.get("legalName") or gst_details.get("tradeName") or "" ) gstin_matched = True except Exception as gst_err: logger.warning("Optional gst_verify call failed: %s", gst_err) # If verified_gst_name wasn't returned by upstream, fallback to matching record or verified_pan_name if PAN matches GSTIN if not verified_gst_name and gstin_matched: verified_gst_name = verified_pan_name # Compare GST legal name against seller full name and verified PAN name if verified_gst_name: gst_name_match = compare_names(seller_name, verified_gst_name) if not gst_name_match["matched"] and verified_pan_name: pan_to_gst_match = compare_names(verified_pan_name, verified_gst_name) if pan_to_gst_match["matched"]: gst_name_match = pan_to_gst_match total_latency = (time.time() - start_time) * 1000 pan_name_matched = bool(match_result.get("matched", False)) gst_name_matched = bool(gst_name_match.get("matched", False)) if target_gstin else True if gst_error: overall_success = False overall_status = "GSTIN_PAN_MISMATCH" error_msg = gst_error elif not pan_name_matched: overall_success = False overall_status = "NAME_MISMATCH" error_msg = f"Name mismatch: The entered name '{seller_name}' does not match PAN record '{verified_pan_name}' ({match_result.get('reason', '')})" elif target_gstin and not gst_name_matched: overall_success = False overall_status = "GST_NAME_MISMATCH" error_msg = f"Name mismatch: The entered name '{seller_name}' does not match GST legal name '{verified_gst_name}' ({gst_name_match.get('reason', '')})" else: overall_success = pan_result.is_success and pan_name_matched overall_status = "SUCCESS" if overall_success else "FAILED" error_msg = None combined_data = { "pan_number": pan, "verified_pan_name": verified_pan_name, "seller_name": seller_name, "name_matched": pan_name_matched, "pan_name_matched": pan_name_matched, "pan_match_score": match_result.get("score", 0.0), "pan_match_reason": match_result.get("reason", ""), "pan_details": pan_data, "gst_number": target_gstin, "verified_gst_name": verified_gst_name, "gst_name_matched": gst_name_matched, "gst_match_score": gst_name_match.get("score", 0.0), "gst_match_reason": gst_name_match.get("reason", ""), "gst_details": gst_details, "gst_records": gst_records, "active_gstins": active_gstins, "primary_gstin": primary_gstin, "entered_gstin": entered_gstin, "gstin_matched": gstin_matched, "is_seller_verified": overall_success, } # Step 4: Persist verification transaction & attempt logs in database tx = VerificationTransaction.objects.create( document_type="SELLER_PAN_GST", status=overall_status, successful_vendor=pan_result.successful_vendor, input_payload={"seller_name": seller_name, "pan": pan, "gstin": entered_gstin}, final_data=combined_data, attempts_count=len(all_attempts), total_latency_ms=total_latency, ) for att in all_attempts: VendorAttemptLog.objects.create( transaction=tx, attempt_number=att.get("attempt_number", 1), vendor_name=att.get("vendor_name", "unknown"), status=att.get("status", "UNKNOWN"), is_success=att.get("is_success", False), data=att.get("data", {}), raw_response=att.get("raw_response"), error_code=att.get("error_code"), error_message=att.get("error_message"), latency_ms=att.get("latency_ms", 0.0), ) res_payload = { "reference_id": str(tx.reference_id), "is_success": overall_success, "status": overall_status, "document_type": "SELLER_PAN_GST", "successful_vendor": pan_result.successful_vendor, "is_seller_verified": overall_success, "data": combined_data, "attempts_count": len(all_attempts), "attempts": all_attempts, "error_message": error_msg, "total_latency_ms": round(total_latency, 2), } tx.response_payload = res_payload tx.save(update_fields=["response_payload"]) http_code = status.HTTP_200_OK if overall_success else status.HTTP_422_UNPROCESSABLE_ENTITY return Response(res_payload, status=http_code) class VendorListView(APIView): """ GET /api/vendors/ Lists all active verification vendors registered with the engine in priority order. """ def get(self, request, *args, **kwargs): vendors = registry.get_all_vendors() vendor_data = [ { "name": v.name, "display_name": getattr(v, "display_name", v.name), "priority": v.priority, "is_active": v.is_active, "supported_documents": v.supported_documents, } for v in vendors ] return Response({"count": len(vendor_data), "vendors": vendor_data}) class TransactionDetailView(APIView): """ GET /api/transactions// Fetch complete audit trail and fallback logs for a transaction. """ def get(self, request, reference_id, *args, **kwargs): try: tx = VerificationTransaction.objects.prefetch_related("attempts").get(reference_id=reference_id) except VerificationTransaction.DoesNotExist: return Response( {"error": f"Transaction '{reference_id}' not found"}, status=status.HTTP_404_NOT_FOUND, ) serializer = VerificationTransactionSerializer(tx) return Response(serializer.data) class TransactionListView(APIView): """ GET /api/transactions/ Lists recent verification transactions (up to 50) for dashboard/portal monitoring. """ def get(self, request, *args, **kwargs): limit = min(int(request.query_params.get("limit", 25)), 50) txs = VerificationTransaction.objects.prefetch_related("attempts").all()[:limit] serializer = VerificationTransactionSerializer(txs, many=True) return Response({"count": len(serializer.data), "results": serializer.data}) from django.views.generic import TemplateView class PortalView(TemplateView): """ Renders the interactive testing & inspection studio for all verification features. """ template_name = "verifier/portal.html" def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) vendors = registry.get_all_vendors() context["vendors"] = vendors context["active_vendor_count"] = sum(1 for v in vendors if v.is_active) context["recent_transactions"] = VerificationTransaction.objects.prefetch_related("attempts").all()[:10] return context