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, PANDetails, SellerBankVerification 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"} # Strict 100% match requirement: names must match 100% # If spaces/delimiters differ, e.g. "VIGNESHKRISHNASAMY" vs "VIGNESH KRISHNASAMY" if "".join(norm1.split()) == "".join(norm2.split()): return {"matched": True, "score": 100.0, "reason": "Exact name match (case & whitespace insensitive)"} # Any non-100% similarity is rejected as name mismatch words1, words2 = norm1.split(), norm2.split() sort_ratio = SequenceMatcher(None, " ".join(sorted(words1)), " ".join(sorted(words2))).ratio() * 100 seq_ratio = SequenceMatcher(None, norm1, norm2).ratio() * 100 sim_score = round(max(sort_ratio, seq_ratio), 1) return { "matched": False, "score": sim_score, "reason": f"Name match score is {sim_score}%. Strict 100% match is required.", } 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"]) # If pan or pan_detailed was verified, persist directly to pan_details table if doc_type in ["pan", "pan_detailed"] and engine_result.is_success and engine_result.data: try: p_data = engine_result.data.get("pan_details") or engine_result.data addr = p_data.get("address_data") or {} PANDetails.objects.create( transaction=transaction, document_type=str(p_data.get("document_type") or "PAN"), document_id=str(p_data.get("document_id") or p_data.get("pan_number") or ""), name=str(p_data.get("name") or p_data.get("full_name") or ""), first_name=p_data.get("first_name"), last_name=p_data.get("last_name"), category=p_data.get("category"), date_of_birth=str(p_data.get("date_of_birth") or p_data.get("dob") or ""), gender=p_data.get("gender"), masked_aadhaar_number=p_data.get("masked_aadhaar_number"), aadhaar_linked=bool(p_data.get("aadhaar_linked") or p_data.get("aadhaar_seeding_status") == "LINKED"), taxpayer_profile=bool(p_data.get("taxpayer_profile")), email=p_data.get("email"), phone=p_data.get("phone"), address_line_1=addr.get("line_1"), street=addr.get("street"), city=addr.get("city"), state=addr.get("state"), pincode=addr.get("pincode"), address_data=addr, request_id=p_data.get("request_id") or engine_result.data.get("request_id"), vendor_txn_id=p_data.get("transaction_id") or engine_result.data.get("transaction_id"), full_response=engine_result.data, ) except Exception as pan_err: logger.error("Failed to store PAN record in pan_details table: %s", pan_err) # If bank verification was performed and seller_id is supplied, persist directly to seller_bank_verification table seller_id = input_data.get("seller_id") or request.data.get("seller_id") if doc_type in ["bank_account", "bank_verify_pd", "bank_verify_v5", "bank_pd", "bank_verify_full", "bank_verify_v4"] and seller_id: try: b_data = engine_result.data or {} last_err_code = engine_result.attempts[-1].get("error_code") if engine_result.attempts else None SellerBankVerification.objects.update_or_create( seller_id=str(seller_id), defaults={ "transaction": transaction, "account_number": str(b_data.get("account_number") or input_data.get("account_number") or ""), "ifsc": str(b_data.get("ifsc") or input_data.get("ifsc") or "").upper(), "account_ifsc": b_data.get("account_ifsc") or b_data.get("ifsc") or str(input_data.get("ifsc") or "").upper(), "beneficiary_name": b_data.get("beneficiary_name") or b_data.get("account_holder_name"), "is_valid": bool(b_data.get("is_valid", engine_result.is_success)), "account_exists": bool(b_data.get("account_exists", engine_result.is_success)), "failure_reason": b_data.get("failure_reason") or engine_result.error_message, "npci_error_code": b_data.get("npci_error_code") or last_err_code, "tranx_id": b_data.get("tranx_id") or b_data.get("transaction_id") or str(transaction.reference_id), "timestamp": str(b_data.get("timestamp") or ""), "user_consent": str(b_data.get("user_consent") or input_data.get("user_consent") or "true"), "vendor_name": engine_result.successful_vendor or "codeplus", "status": engine_result.status.value, "latency_ms": round(engine_result.total_latency_ms, 2), } ) except Exception as bank_err: logger.error("Failed to store bank record in seller_bank_verification table: %s", bank_err) 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 SellerBankVerifyView(APIView): """ POST /api/verify/seller/bank/ Dedicated composite endpoint for seller bank account verification. Verifies bank account via Codeplus Penny Drop v5 and automatically saves the complete record to seller_bank_verification table with seller_id as primary key: - seller_id (primarykey) - account_number - ifsc - account_ifsc - beneficiary_name - is_valid - account_exists - failure_reason - npci_error_code - tranx_id - timestamp - user_consent - vendor_name - status - latency_ms """ def post(self, request, *args, **kwargs): seller_id = str(request.data.get("seller_id") or "").strip() account_number = str(request.data.get("account_number") or "").strip() ifsc = str(request.data.get("ifsc") or "").strip().upper() user_consent = str(request.data.get("user_consent") or "true") preferred_vendor = request.data.get("preferred_vendor") if not seller_id: return Response({"is_success": False, "error": "seller_id is required"}, status=status.HTTP_400_BAD_REQUEST) if not account_number: return Response({"is_success": False, "error": "account_number is required"}, status=status.HTTP_400_BAD_REQUEST) if not ifsc: return Response({"is_success": False, "error": "ifsc is required"}, status=status.HTTP_400_BAD_REQUEST) # Run verification via waterfall engine engine_result = engine.verify( document_type="bank_verify_pd", data={ "seller_id": seller_id, "account_number": account_number, "ifsc": ifsc, "user_consent": user_consent, }, preferred_vendor=preferred_vendor, ) safe_input = mask_sensitive_payload({"seller_id": seller_id, "account_number": account_number, "ifsc": ifsc}) transaction = VerificationTransaction.objects.create( document_type="BANK_ACCOUNT", 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, ) 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), ) b_data = engine_result.data or {} is_valid = bool(b_data.get("is_valid", engine_result.is_success)) account_exists = bool(b_data.get("account_exists", engine_result.is_success)) beneficiary_name = b_data.get("beneficiary_name") or b_data.get("account_holder_name") or "" account_ifsc = b_data.get("account_ifsc") or ifsc last_err_code = engine_result.attempts[-1].get("error_code") if engine_result.attempts else None failure_reason = b_data.get("failure_reason") or engine_result.error_message npci_error_code = b_data.get("npci_error_code") or last_err_code tranx_id = b_data.get("tranx_id") or b_data.get("transaction_id") or str(transaction.reference_id) timestamp = str(b_data.get("timestamp") or "") vendor_name = engine_result.successful_vendor or "codeplus" res_status = engine_result.status.value latency_ms = round(engine_result.total_latency_ms, 2) # Save to SellerBankVerification table with seller_id as primary key SellerBankVerification.objects.update_or_create( seller_id=str(seller_id), defaults={ "transaction": transaction, "account_number": account_number, "ifsc": ifsc, "account_ifsc": account_ifsc, "beneficiary_name": beneficiary_name, "is_valid": is_valid, "account_exists": account_exists, "failure_reason": failure_reason, "npci_error_code": npci_error_code, "tranx_id": tranx_id, "timestamp": timestamp, "user_consent": user_consent, "vendor_name": vendor_name, "status": res_status, "latency_ms": latency_ms, } ) response_payload = { "seller_id": seller_id, "reference_id": str(transaction.reference_id), "is_success": is_valid, "status": res_status, "account_number": account_number, "ifsc": ifsc, "account_ifsc": account_ifsc, "beneficiary_name": beneficiary_name, "is_valid": is_valid, "account_exists": account_exists, "failure_reason": failure_reason, "npci_error_code": npci_error_code, "tranx_id": tranx_id, "timestamp": timestamp, "user_consent": user_consent, "vendor_name": vendor_name, "latency_ms": latency_ms, "attempts_count": len(engine_result.attempts), "attempts": engine_result.attempts, } transaction.response_payload = response_payload transaction.save(update_fields=["response_payload"]) http_code = status.HTTP_200_OK if is_valid else status.HTTP_422_UNPROCESSABLE_ENTITY return Response(response_payload, status=http_code) 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() business_name = (serializer.validated_data.get("business_name") or request.data.get("business_name") or "").strip() preferred_vendor = serializer.validated_data.get("preferred_vendor") start_time = time.time() # Step 1: Verify PAN via waterfall engine using pan_detailed API pan_result = engine.verify( document_type="pan_detailed", data={"pan_number": pan, "pan": pan}, preferred_vendor=preferred_vendor, ) all_attempts = list(pan_result.attempts) # Fallback to standard PAN if detailed PAN fails or is not enabled for vendor if not pan_result.is_success: logger.info("Detailed PAN lookup not successful, attempting standard PAN verification...") pan_basic_result = engine.verify( document_type="pan", data={"pan": pan, "pan_number": pan}, preferred_vendor=preferred_vendor, ) for att in pan_basic_result.attempts: att["attempt_number"] = len(all_attempts) + 1 all_attempts.append(att) if pan_basic_result.is_success: pan_result = pan_basic_result 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 and detailed attributes 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 # Do not restrict GSTIN to individual PAN. Entities/companies have their own PAN inside their GSTIN. # Just verify that the GSTIN exists and is active. # 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 and trade name against business name (as per GST), seller name, and verified PAN name gst_candidates = [n for n in [verified_gst_name, gst_details.get("legalName"), gst_details.get("tradeName")] if n] # 1. Primary: Match against entered business name (as per GST) if business_name: for candidate in gst_candidates: b_match = compare_names(business_name, candidate) if b_match["matched"]: gst_name_match = b_match break # 2. Fallback: Match against seller name or verified PAN name if not gst_name_match.get("matched", False): 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 = f"GSTIN Mismatch: {gst_error}" elif not pan_name_matched: overall_success = False overall_status = "NAME_MISMATCH" error_msg = f"PAN Name Mismatch: The entered seller name '{seller_name}' does not match" elif target_gstin and not gst_name_matched: overall_success = False overall_status = "GST_NAME_MISMATCH" error_msg = f"GST Name Mismatch: The entered seller name '{seller_name}' does not match" 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), ) # Step 5: Store full PAN detailed response in dedicated 'pan_details' table for future analysis try: addr = pan_data.get("address_data") or {} PANDetails.objects.create( transaction=tx, document_type=str(pan_data.get("document_type") or "PAN"), document_id=str(pan_data.get("document_id") or pan), name=str(pan_data.get("name") or verified_pan_name or seller_name), first_name=pan_data.get("first_name"), last_name=pan_data.get("last_name"), category=pan_data.get("category"), date_of_birth=str(pan_data.get("date_of_birth") or pan_data.get("dob") or ""), gender=pan_data.get("gender"), masked_aadhaar_number=pan_data.get("masked_aadhaar_number"), aadhaar_linked=bool(pan_data.get("aadhaar_linked") or pan_data.get("aadhaar_seeding_status") == "LINKED"), taxpayer_profile=bool(pan_data.get("taxpayer_profile")), email=pan_data.get("email"), phone=pan_data.get("phone"), address_line_1=addr.get("line_1"), street=addr.get("street"), city=addr.get("city"), state=addr.get("state"), pincode=addr.get("pincode"), address_data=addr, request_id=pan_data.get("request_id"), vendor_txn_id=pan_data.get("transaction_id"), full_response=pan_data, ) except Exception as db_err: logger.error("Failed to store record in pan_details table: %s", db_err) 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