from typing import Any, Dict, List, Optional, Tuple, Union from datetime import datetime import asyncio import time from fastapi import HTTPException, status, Request from sqlalchemy.orm import Session from src.modules.prompt.schemas.requests import AddressData, BankStatementData, VehicleDocumentData from src.modules.prompt.schemas.responses import AddressMatchResponse, MatchResult, ErrorResponse from src.modules.prompt.schemas.requests import MultiDocumentVerificationRequest from src.utils.azure_openai import azure_openai_client from src.utils.pdf_parser import get_pdf_parser # from src.utils.file_processor import FileProcessor # Removed - easyocr disabled from pydantic import BaseModel from src.core.logging import logger # ---------------- Performance Cache ----------------- # In-memory cache to avoid repeated Azure OpenAI calls for identical # (file_type, file_url, payload) combinations within a short TTL window. # Key: f"{file_type}|{file_url}|{sorted_payload_items}" -> (timestamp, openai_result_dict) CACHE_TTL_SECONDS = 0 # Disabled - process every request fresh MAX_CACHE_ENTRIES = 128 _verification_cache: Dict[str, Tuple[float, Dict]] = {} _verification_cache_lock = asyncio.Lock() def _build_cache_key(file_type: str, file_url: str, payload: Dict) -> str: try: items = sorted(payload.items()) return f"{file_type}|{file_url}|{items}" except Exception: return f"{file_type}|{file_url}|fallback" async def _get_cached_verification(key: str) -> Optional[Dict]: now = time.time() async with _verification_cache_lock: entry = _verification_cache.get(key) if entry: ts, data = entry if now - ts <= CACHE_TTL_SECONDS: return data else: _verification_cache.pop(key, None) return None async def _store_cached_verification(key: str, data: Dict): async with _verification_cache_lock: if key not in _verification_cache and len(_verification_cache) >= MAX_CACHE_ENTRIES: oldest_key = min(_verification_cache.items(), key=lambda kv: kv[1][0])[0] _verification_cache.pop(oldest_key, None) _verification_cache[key] = (time.time(), data) class VerificationRequest(BaseModel): file_type: str file_urls: list[str] data: Optional[AddressData | BankStatementData | VehicleDocumentData] def build_response_from_openai_result( openai_result: Dict, file_type: str, file_url: str ) -> AddressMatchResponse: """ Build AddressMatchResponse from raw OpenAI result. This is the same logic used in verify_address_with_document, extracted for reuse. Args: openai_result: Raw response dict from Azure OpenAI file_type: Document type (e.g., 'residence_documents', 'bank_statement') file_url: Source file URL(s) - can be comma-separated Returns: AddressMatchResponse with all fields populated """ analysis_results: List[MatchResult] = [] mismatch_details: List[str] = [] bank_statement_results = None vehicle_details = None # Process analysis results analysis_results_raw = openai_result.get("analysis_results", []) if not isinstance(analysis_results_raw, list): analysis_results_raw = [] # Capture vehicle-specific details if present if openai_result.get("vehicle_details"): vehicle_details = openai_result.get("vehicle_details") for result in analysis_results_raw: try: if not isinstance(result, dict): continue field_name = str(result.get("field_name") or "") provided_value = str(result.get("provided_value") or "") found_in_document = result.get("found_in_document") if result.get("found_in_document") is not None else None is_match = bool(result.get("is_match", False)) confidence = str(result.get("confidence") or "low") if not field_name: continue match_result = MatchResult( field_name=field_name, provided_value=provided_value, found_in_document=found_in_document, is_match=is_match, confidence=confidence ) analysis_results.append(match_result) if not is_match: expected_val = provided_value if provided_value else 'N/A' found_val = found_in_document if found_in_document else 'Not found' mismatch_details.append( f"{field_name}: Expected '{expected_val}', Found '{found_val}'" ) except Exception as result_error: logger.error(f"Failed to process field analysis: {str(result_error)}") mismatch_details.append(f"Failed to process field analysis: {str(result_error)}") continue # Add additional mismatch info if openai_result.get("mismatch_summary"): mismatch_details.extend(openai_result["mismatch_summary"]) # Determine overall match overall_match = openai_result.get("overall_match", False) if file_type != "bank_statement" else None # Create response message confidence_score = openai_result.get("confidence_score", "N/A") document_type = openai_result.get("document_type", "Unknown") is_hazy = openai_result.get("is_hazy", False) # Check if this is a content filter error if document_type == "Error" and "Content filter triggered" in mismatch_details: message = ( f"Please check the document and ensure it contains appropriate content." ) else: message = ( f"Verification completed. Overall match: {overall_match}. " f"Confidence: {confidence_score}. Document type: {document_type}. Hazy: {is_hazy}" ) # Build feedback classification fully_matched = [] partially_matched = [] not_matched = [] feedback_details = [] for result in analysis_results: field_name = result.field_name is_match = result.is_match # Check for partial_match in raw data raw_result = next((r for r in analysis_results_raw if r.get("field_name") == field_name), {}) partial_match = raw_result.get("partial_match", False) if is_match: fully_matched.append(field_name) feedback_details.append({ "field": field_name, "status": "fully_matched", "explanation": f"Field '{field_name}' matches the document." }) elif partial_match: partially_matched.append(field_name) feedback_details.append({ "field": field_name, "status": "partially_matched", "explanation": f"Field '{field_name}' partially matches." }) else: not_matched.append(field_name) feedback_details.append({ "field": field_name, "status": "not_matched", "explanation": f"Field '{field_name}' not found or no match." }) # Build rich feedback text (same logic as original controller) feedback_text_parts = [] # Check document type mismatch based on file_type (include "Unknown" types too) if document_type: if file_type in ('vehicle_documents', 'vehicle_registration', 'vehicle_insurance') and 'Vehicle' not in document_type: feedback_text_parts.append(f"This looks like a {document_type}, not a vehicle document.") elif file_type == 'residence_documents' and 'Residence' not in document_type and 'Address' not in document_type: feedback_text_parts.append(f"This looks like a {document_type}, not a residence document.") elif file_type == 'bank_statement' and 'Bank' not in document_type and 'Statement' not in document_type: feedback_text_parts.append(f"This looks like a {document_type}, not a bank statement.") # Add hazy warning if is_hazy: feedback_text_parts.append("The image is a bit blurry.") # Add matched fields if fully_matched: feedback_text_parts.append(f"We found these matching: {', '.join(sorted(fully_matched))}.") # Add partial match details with differences if partially_matched: diffs = [] for result in analysis_results: if result.field_name in partially_matched and result.found_in_document is not None: if str(result.found_in_document) != str(result.provided_value): diffs.append(f"{result.field_name} differs (expected '{result.provided_value}' vs found '{result.found_in_document}')") if diffs: feedback_text_parts.append("Differences: " + '; '.join(diffs) + ".") else: feedback_text_parts.append("Some fields are present but differ: " + ', '.join(sorted(partially_matched)) + ".") # Add missing fields if not_matched: feedback_text_parts.append(f"Missing or not found: {', '.join(sorted(not_matched))}.") # Add guidance based on file_type if file_type in ('vehicle_documents', 'vehicle_registration', 'vehicle_insurance'): feedback_text_parts.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") elif file_type == 'residence_documents': feedback_text_parts.append("Please upload a clear residence document showing the address details.") elif file_type == 'bank_statement': feedback_text_parts.append("Please upload a clear bank statement showing account details and transactions.") feedback_text = " ".join(feedback_text_parts) if feedback_text_parts else "No fields analyzed." # Preserve overall_feedback_text from Azure OpenAI if it exists (e.g., jailbreak errors) # Otherwise use the constructed feedback_text final_overall_feedback_text = openai_result.get("overall_feedback_text") or feedback_text return AddressMatchResponse( success=True, message=message, file_url=file_url, analysis_results=analysis_results, overall_match=overall_match, mismatch_details=mismatch_details, is_hazy=is_hazy, document_type_detected=document_type, bank_statement_results=bank_statement_results, vehicle_details=vehicle_details, aggregated_months_covered=openai_result.get("aggregated_months_covered"), aggregated_missing_months=openai_result.get("aggregated_missing_months"), aggregated_months_complete=openai_result.get("aggregated_months_complete"), original_file_urls=[u.strip() for u in file_url.split(',') if u.strip()], feedback={ "fully_matched": sorted(fully_matched), "partially_matched": sorted(partially_matched), "not_matched": sorted(not_matched) }, per_file_results=openai_result.get("per_file_results"), feedback_details=feedback_details, feedback_text=feedback_text, overall_feedback_text=final_overall_feedback_text, timestamp=datetime.now().isoformat() ) async def verify_address_with_document( db: Session, file_type: str, file_url: str, address_data: VerificationRequest, request: Request ) -> AddressMatchResponse: """Verification using separate params (file_type, file_url) and JSON body for address fields.""" try: # Build payload based on file_type address_dict = address_data.dict() # Analyze document with OpenAI (with caching layer) cache_key = _build_cache_key(file_type, file_url, address_dict) openai_result = await _get_cached_verification(cache_key) if openai_result is None: start_call = time.time() try: openai_result = await azure_openai_client.verify_document( file_type, address_dict, file_url ) # Only cache if LlamaParse extraction succeeded for better accuracy llamaparse_success = openai_result.get('_llamaparse_success', False) if llamaparse_success: await _store_cached_verification(cache_key, openai_result) print(f"PERF: OpenAI call (cache miss, cached) {file_type} {file_url} in {time.time() - start_call:.2f}s") else: logger.warning(f"LlamaParse failed - skipping cache to allow retry on next request") print(f"PERF: OpenAI call (cache miss, NOT cached - LlamaParse failed) {file_type} {file_url} in {time.time() - start_call:.2f}s") except Exception as openai_error: error_message = f"Failed to analyze document: {str(openai_error)}" gor_error = AddressMatchResponse( success=False, message=error_message, file_url=file_url, analysis_results=[], overall_match=False, mismatch_details=[f"Azure OpenAI API Error: {str(openai_error)}"], timestamp=datetime.now().isoformat() ) logger.critical(f"OPEN AI ERROR - {gor_error}") return gor_error else: logger.info(f"PERF: Cache hit {file_type} {file_url}") print(f"PERF: Cache hit {file_type} {file_url}") # Process OpenAI response safely analysis_results = [] mismatch_details = [] bank_statement_results = None vehicle_details = None # Debug: Print the OpenAI result structure print(f"DEBUG: OpenAI result type: {type(openai_result)}") print(f"DEBUG: OpenAI result keys: {openai_result.keys() if isinstance(openai_result, dict) else 'Not a dict'}") print(f"DEBUG: Analysis results: {openai_result.get('analysis_results', 'Not found')}") ''' # BANK STATEMENT SPECIFIC LOGIC - COMMENTED OUT # This code handles complex bank statement verification with multiple features: # - Multi-file URL processing and result padding # - Per-file feedback classification (fully/partially/not matched fields) # - Name matching (first_name, middle_name, last_name) # - Bank name and account type verification # - Final balance comparison # - Arithmetic consistency checks across months # - 6-month coverage window calculation and validation # - Month aggregation from balances and transactions # - Per-file and aggregated month coverage tracking if file_type == "bank_statement": bank_statement_results = openai_result.get("bank_statement_results", []) if not isinstance(bank_statement_results, list): bank_statement_results = [] # Override any hallucinated file_url values with the original input URLs (order preserved) original_urls = [u.strip() for u in file_url.split(',') if u.strip()] for idx, item in enumerate(bank_statement_results): if isinstance(item, dict): if original_urls: safe_url = original_urls[idx] if idx < len(original_urls) else original_urls[-1] item["file_url"] = safe_url elif not item.get("file_url"): item["file_url"] = file_url # Add per-file feedback classification for bank statements fully_fs: List[str] = [] partial_fs: List[str] = [] not_fs: List[str] = [] name_matches = item.get("name_matches") or {} for n_field, matched in name_matches.items(): if matched is True: fully_fs.append(n_field) elif matched is False: not_fs.append(n_field) # Bank name if item.get("bank_name_matches") is True: fully_fs.append("bank_name") elif item.get("bank_name_matches") is False: not_fs.append("bank_name") # Account type if item.get("account_type_matches") is True: fully_fs.append("account_type") elif item.get("account_type_matches") is False: not_fs.append("account_type") # Final balance if item.get("final_balance_matches") is True: fully_fs.append("final_balance") elif item.get("final_balance_matches") is False and item.get("statement_final_balance"): # treat as partial if a statement balance present but mismatch partial_fs.append("final_balance") # Arithmetic consistency (if present) inconsistencies = item.get("arithmetic_inconsistencies") or [] # If no inconsistencies and we have balances array -> fully matched arithmetic if isinstance(inconsistencies, list) and len(inconsistencies) == 0 and item.get("balances"): fully_fs.append("arithmetic") elif item.get("balances") and inconsistencies: partial_fs.append("arithmetic") # Deduplicate and remove overlaps fully_set = set(fully_fs) partial_set = set(partial_fs) - fully_set not_set = set(not_fs) - fully_set - partial_set # Build detailed explanations details: List[Dict[str, str]] = [] def add_detail(field: str, status: str, explanation: str): details.append({"field": field, "status": status, "explanation": explanation}) # Names for n_field in ["first_name", "middle_name", "last_name"]: if n_field in fully_set: add_detail(n_field, "fully_matched", f"{n_field.replace('_',' ').title()} matches exactly across the statement.") elif n_field in not_set: add_detail(n_field, "not_matched", f"{n_field.replace('_',' ').title()} not found or does not match on the statement.") # Bank name if "bank_name" in fully_set: add_detail("bank_name", "fully_matched", "Bank name matches exactly.") elif "bank_name" in not_set: add_detail("bank_name", "not_matched", "Bank name is missing or different.") # Account type if "account_type" in fully_set: add_detail("account_type", "fully_matched", "Account type matches provided type.") elif "account_type" in not_set: add_detail("account_type", "not_matched", "Account type not found or mismatched.") # Final balance if "final_balance" in fully_set: add_detail("final_balance", "fully_matched", "Statement final balance matches expected balance.") elif "final_balance" in partial_set: add_detail("final_balance", "partially_matched", "Statement final balance present but differs from expected.") elif "final_balance" in not_set: add_detail("final_balance", "not_matched", "Unable to verify final balance.") # Arithmetic if "arithmetic" in fully_set: add_detail("arithmetic", "fully_matched", "Monthly arithmetic consistent: opening - debits + credits equals closing for all months.") elif "arithmetic" in partial_set: add_detail("arithmetic", "partially_matched", "Some months show arithmetic discrepancies; review transaction totals.") elif "arithmetic" in not_set: add_detail("arithmetic", "not_matched", "Insufficient data to verify arithmetic consistency.") item["feedback"] = { "fully_matched": sorted(fully_set), "partially_matched": sorted(partial_set), "not_matched": sorted(not_set), "details": details } # Pad missing results if model returned fewer objects than input URLs if len(bank_statement_results) < len(original_urls): for pad_idx in range(len(bank_statement_results), len(original_urls)): bank_statement_results.append({ "file_url": original_urls[pad_idx], "months_covered": [], "missing_months": [], "balances": [], "transactions": [], "provided_final_balance": address_dict.get("balance"), "statement_final_balance": None, "final_balance_matches": False, "months_complete": False, "name_matches": { "first_name": False, "middle_name": False, "last_name": False }, "bank_name_matches": False, "account_type_matches": False, "errors": ["No analysis returned for this file."], "feedback": { "fully_matched": [], "partially_matched": [], "not_matched": [], "details": [] } }) # --- Enhanced Month Coverage Logic --- # Build month sets per file from months_covered, balances.month and transactions.date from datetime import datetime as _dt now = _dt.now() current_year = now.year current_month = now.month # Target window = current month plus previous 5 months (total 6 months) target_window: List[str] = [] for i in range(6): # oldest first y = current_year m = current_month - (5 - i) while m <= 0: y -= 1 m += 12 target_window.append(f"{y:04d}-{m:02d}") # Normalize each file's months_covered using fallback sources for item in bank_statement_results: if not isinstance(item, dict): continue month_set = set() # existing months_covered for m in (item.get("months_covered") or []): if isinstance(m, str) and len(m) >= 7: month_set.add(m[:7]) # balances month field for bal in (item.get("balances") or []): if isinstance(bal, dict): mm = bal.get("month") if isinstance(mm, str) and len(mm) >= 7: month_set.add(mm[:7]) # transactions date -> YYYY-MM for txn in (item.get("transactions") or []): if isinstance(txn, dict): dt_val = txn.get("date") if isinstance(dt_val, str) and len(dt_val) >= 7: month_set.add(dt_val[:7]) # overwrite months_covered sorted ascending item["months_covered"] = sorted(month_set) # compute missing_months for this file relative to target window item_missing = [m for m in target_window if m not in month_set] item["missing_months"] = item_missing item["months_complete"] = len(item_missing) == 0 # Aggregated months coverage restricted to target window aggregated_months_covered = [m for m in target_window if any(isinstance(r, dict) and m in (r.get("months_covered") or []) for r in bank_statement_results)] aggregated_months_missing = [m for m in target_window if m not in aggregated_months_covered] aggregated_months_complete = len(aggregated_months_missing) == 0 openai_result["aggregated_months_covered"] = aggregated_months_covered openai_result["aggregated_missing_months"] = aggregated_months_missing openai_result["aggregated_months_complete"] = aggregated_months_complete else: ''' # Generic document processing (works for all document types including bank_statement) if True: # Safely process analysis results for non-bank documents analysis_results_raw = openai_result.get("analysis_results", []) if not isinstance(analysis_results_raw, list): print(f"WARNING: analysis_results is not a list: {type(analysis_results_raw)}") analysis_results_raw = [] # Capture vehicle-specific details if present if openai_result.get("vehicle_details"): vehicle_details = openai_result.get("vehicle_details") for i, result in enumerate(analysis_results_raw): try: print(f"DEBUG: Processing result {i}: {result}") if not isinstance(result, dict): print(f"WARNING: Skipping non-dict result {i}: {result}") continue field_name = str(result.get("field_name") or "") provided_value = str(result.get("provided_value") or "") found_in_document = result.get("found_in_document") if result.get("found_in_document") is not None else None is_match = bool(result.get("is_match", False)) confidence = str(result.get("confidence") or "low") if not field_name: continue match_result = MatchResult( field_name=field_name, provided_value=provided_value, found_in_document=found_in_document, is_match=is_match, confidence=confidence ) analysis_results.append(match_result) if not is_match: expected_val = provided_value if provided_value else 'N/A' found_val = found_in_document if found_in_document else 'Not found' mismatch_details.append( f"{field_name}: Expected '{expected_val}', Found '{found_val}'" ) except Exception as result_error: logger.error(f"Failed to process field analysis: {str(result_error)}") mismatch_details.append(f"Failed to process field analysis: {str(result_error)}") continue # Add additional mismatch info if openai_result.get("mismatch_summary"): mismatch_details.extend(openai_result["mismatch_summary"]) # Determine overall match overall_match = openai_result.get("overall_match", False) if file_type != "bank_statement" else None # Create response message confidence_score = openai_result.get("confidence_score", "N/A") document_type = openai_result.get("document_type", "Unknown") is_hazy = openai_result.get("is_hazy", False) # if file_type == "bank_statement": # message = (f"Bank statement verification completed. Files processed: {len(bank_statement_results)}. " # f"Confidence: {confidence_score if confidence_score!='N/A' else 'n/a'}. Document type: {document_type}. Hazy: {is_hazy}") # else: # message = ( # f"Verification completed. Overall match: {overall_match}. " # f"Confidence: {confidence_score}. Document type: {document_type}. Hazy: {is_hazy}" # ) # Generic message for all document types (including bank_statement) message = ( f"Verification completed. Overall match: {overall_match}. " f"Confidence: {confidence_score}. Document type: {document_type}. Hazy: {is_hazy}" ) return AddressMatchResponse( success=True, message=message, file_url=file_url, analysis_results=analysis_results, overall_match=overall_match, mismatch_details=mismatch_details, is_hazy=is_hazy, document_type_detected=document_type, bank_statement_results=bank_statement_results, vehicle_details=vehicle_details, aggregated_months_covered=openai_result.get("aggregated_months_covered"), aggregated_missing_months=openai_result.get("aggregated_missing_months"), aggregated_months_complete=openai_result.get("aggregated_months_complete"), original_file_urls=[u.strip() for u in file_url.split(',') if u.strip()], timestamp=datetime.now().isoformat() ) except HTTPException: raise except Exception as e: print(f"ERROR: Unexpected error in verify_address_with_document: {str(e)}") fallback_results: List[MatchResult] = [] if file_url and file_type != "bank_statement": try: for field, value in address_data.dict().items(): if value: fallback_results.append(MatchResult( field_name=field, provided_value=str(value), found_in_document=None, is_match=False, confidence="error" )) except Exception: pass return AddressMatchResponse( success=False, message=f"Error verifying address: {str(e)}", file_url=file_url, analysis_results=fallback_results if file_type != "bank_statement" else [], overall_match=False if file_type != "bank_statement" else None, mismatch_details=[f"System Error: {str(e)}"], is_hazy=None, document_type_detected=None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, original_file_urls=[u.strip() for u in file_url.split(',') if u.strip()], timestamp=datetime.now().isoformat() ) async def verify_multi_documents( db: Session, request_data: MultiDocumentVerificationRequest, request: Request ) -> AddressMatchResponse: """Process multi-document verification using a unified request body. Internally reuses existing verify_address_with_document logic by joining file URLs. """ try: file_type = request_data.file_type # Join file urls with comma to leverage existing downstream logic (bank statements expect this format) # Cast HttpUrl objects to str explicitly to avoid join TypeError file_url_joined = ','.join([str(u) for u in request_data.file_urls]) # data is already parsed into BankStatementData or AddressData via Union in schema parsed = request_data.data # Special handling: for non-bank documents with multiple URLs, call analysis per file if file_type != 'bank_statement' and len(request_data.file_urls) > 1: # Concurrent per-file verification for non-bank multi-file requests batch_start = time.time() tasks = [verify_address_with_document(db, file_type, str(single_url), parsed, request) for single_url in request_data.file_urls] single_responses = await asyncio.gather(*tasks) aggregated_results: List[MatchResult] = [] aggregated_mismatches: List[str] = [] overall_match_all = True is_hazy_any: Optional[bool] = False doc_types: List[str] = [] per_file_payload: List[Dict] = [] for single_resp, single_url in zip(single_responses, request_data.file_urls): if single_resp.analysis_results: aggregated_results.extend(single_resp.analysis_results) if single_resp.mismatch_details: aggregated_mismatches.extend(single_resp.mismatch_details) if single_resp.is_hazy: is_hazy_any = True if single_resp.document_type_detected: doc_types.append(single_resp.document_type_detected) if single_resp.overall_match is False: overall_match_all = False pf_fully: List[str] = [] pf_partial: List[str] = [] pf_not: List[str] = [] if single_resp.analysis_results: for r in single_resp.analysis_results: if r.is_match: pf_fully.append(r.field_name) else: if r.found_in_document: pf_partial.append(r.field_name) else: pf_not.append(r.field_name) per_file_feedback = { 'fully_matched': sorted(set(pf_fully)), 'partially_matched': sorted(set(pf_partial) - set(pf_fully)), 'not_matched': sorted(set(pf_not) - set(pf_fully) - set(pf_partial)) } summary_parts: List[str] = [] if single_resp.document_type_detected and ('Vehicle' not in single_resp.document_type_detected): summary_parts.append(f"This looks like a {single_resp.document_type_detected}, not a vehicle document.") if single_resp.is_hazy: summary_parts.append("The image is a bit blurry.") if per_file_feedback.get('fully_matched'): summary_parts.append("We found these matching: " + ', '.join(per_file_feedback['fully_matched']) + ".") partials = per_file_feedback.get('partially_matched') or [] if partials: diffs: List[str] = [] for r in (single_resp.analysis_results or []): try: if r.field_name in partials and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") except Exception: continue if diffs: summary_parts.append("Differences: " + '; '.join(diffs) + ".") else: summary_parts.append("Some fields are present but differ: " + ', '.join(partials) + ".") missing = per_file_feedback.get('not_matched') or [] if missing: summary_parts.append("Missing or not found: " + ', '.join(missing) + ".") if file_type != 'real_estate_documents': summary_parts.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") user_friendly_feedback = ' '.join(summary_parts) per_file_payload.append({ 'file_url': str(single_url), 'analysis_results': [r.model_dump() for r in (single_resp.analysis_results or [])], 'mismatch_details': single_resp.mismatch_details or [], 'overall_match': single_resp.overall_match, 'is_hazy': single_resp.is_hazy, 'document_type_detected': single_resp.document_type_detected, 'feedback': per_file_feedback, 'feedback_text': user_friendly_feedback }) dedup: Dict[str, MatchResult] = {} for r in aggregated_results: existing = dedup.get(r.field_name) if existing is None: dedup[r.field_name] = r else: conf_order = {'high': 3, 'medium': 2, 'low': 1} def score(m: MatchResult): return (1 if m.is_match else 0, conf_order.get(m.confidence, 0)) if score(r) > score(existing): dedup[r.field_name] = r merged_results = list(dedup.values()) elapsed = time.time() - batch_start print(f"PERF: Concurrent multi-file verification ({len(request_data.file_urls)} files, type={file_type}) in {elapsed:.2f}s") base_response = AddressMatchResponse( success=True, message=f"Multi-file verification completed. Overall match: {overall_match_all}.", file_url=file_url_joined, analysis_results=merged_results, overall_match=overall_match_all, mismatch_details=sorted(set(aggregated_mismatches)), is_hazy=is_hazy_any, document_type_detected=doc_types[0] if doc_types else None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, original_file_urls=[str(u) for u in request_data.file_urls], per_file_results=per_file_payload, timestamp=datetime.now().isoformat() ) else: base_response = await verify_address_with_document(db, file_type, file_url_joined, parsed, request) # ensure original_file_urls populated if not base_response.original_file_urls: base_response.original_file_urls = [str(u) for u in request_data.file_urls] # Build feedback only for this unified endpoint without altering underlying verification logic fully: List[str] = [] partial: List[str] = [] not_matched: List[str] = [] if file_type == 'bank_statement' and base_response.bank_statement_results: # Aggregate across all statement result objects for item in base_response.bank_statement_results: if not isinstance(item, dict): continue name_matches = item.get('name_matches') or {} for k, v in name_matches.items(): if v: fully.append(k) else: not_matched.append(k) # Other boolean match fields if item.get('bank_name_matches') is True: fully.append('bank_name') elif item.get('bank_name_matches') is False: not_matched.append('bank_name') if item.get('account_type_matches') is True: fully.append('account_type') elif item.get('account_type_matches') is False: not_matched.append('account_type') if item.get('final_balance_matches') is True: fully.append('balance') elif item.get('final_balance_matches') is False: not_matched.append('balance') # Deduplicate fully = sorted(set(fully)) not_matched = sorted(set(not_matched) - set(fully)) else: # Use analysis_results list; partial = found_in_document present but is_match False if base_response.analysis_results: for r in base_response.analysis_results: if r.is_match: fully.append(r.field_name) else: if r.found_in_document: partial.append(r.field_name) else: not_matched.append(r.field_name) fully = sorted(set(fully)) partial = sorted(set(partial) - set(fully)) not_matched = sorted(set(not_matched) - set(fully) - set(partial)) # Attach feedback base_response.feedback = { 'fully_matched': fully, 'partially_matched': partial, 'not_matched': not_matched } # Build feedback_details for user-friendly explanations feedback_details: List[Dict[str, str]] = [] def explain(field: str, status: str, provided: Optional[str] = None, found: Optional[str] = None): if status == 'fully_matched': msg = f"Field '{field}' matches exactly." if provided else f"Field '{field}' verified." elif status == 'partially_matched': msg = f"Field '{field}' present but value differs (expected '{provided}' vs found '{found}')." if provided else f"Field '{field}' partially matched." else: msg = f"Field '{field}' not found or no match." if provided else f"Field '{field}' missing." feedback_details.append({"field": field, "status": status, "explanation": msg}) # We can use analysis_results for non-bank and bank aggregated sets for bank if file_type == 'bank_statement' and base_response.bank_statement_results: # derive provided vs found from each bank statement result item for item in base_response.bank_statement_results: fb = item.get('feedback') or {} for fld in fb.get('fully_matched', []): explain(fld, 'fully_matched') for fld in fb.get('partially_matched', []): explain(fld, 'partially_matched') for fld in fb.get('not_matched', []): explain(fld, 'not_matched') else: if base_response.analysis_results: for r in base_response.analysis_results: status = 'fully_matched' if r.field_name in fully else ('partially_matched' if r.field_name in partial else 'not_matched') explain(r.field_name, status, r.provided_value, r.found_in_document) base_response.feedback_details = feedback_details # If single-file (no per_file_results) and non-bank, build feedback_text if file_type != 'bank_statement' and not base_response.per_file_results: try: fully_local = base_response.feedback.get('fully_matched') if base_response.feedback else [] partial_local = base_response.feedback.get('partially_matched') if base_response.feedback else [] not_local = base_response.feedback.get('not_matched') if base_response.feedback else [] summary_bits: List[str] = [] if base_response.document_type_detected and 'Vehicle' not in base_response.document_type_detected: summary_bits.append(f"This looks like a {base_response.document_type_detected}, not a vehicle document.") if base_response.is_hazy: summary_bits.append("The image is a bit blurry.") if fully_local: summary_bits.append("Matched: " + ', '.join(fully_local) + ".") if partial_local: diffs: List[str] = [] for r in (base_response.analysis_results or []): if r.field_name in partial_local and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") if diffs: summary_bits.append("Differences: " + '; '.join(diffs) + ".") else: summary_bits.append("Some fields are present but differ: " + ', '.join(partial_local) + ".") if not_local: summary_bits.append("Missing or not found: " + ', '.join(not_local) + ".") if file_type != 'real_estate_documents': summary_bits.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") setattr(base_response, 'feedback_text', ' '.join(summary_bits)) except Exception: pass # Bank statement specific human-readable feedback (single or multi-file) if file_type == 'bank_statement' and base_response.bank_statement_results: try: bs_items = base_response.bank_statement_results or [] summary_bits: List[str] = [] # Balance match status any_balance_match = any(isinstance(it, dict) and it.get('final_balance_matches') for it in bs_items) if any_balance_match: summary_bits.append("Final balance matches the provided expected amount.") else: summary_bits.append("Final balance could not be verified or does not match.") # Name matches aggregation name_fields = ['first_name','middle_name','last_name'] matched_names = [] missing_names = [] for nf in name_fields: # consider matched if any file has nf True if any(isinstance(it, dict) and isinstance(it.get('name_matches'), dict) and it['name_matches'].get(nf) is True for it in bs_items): matched_names.append(nf) else: missing_names.append(nf) if matched_names: summary_bits.append("Name components matched: " + ', '.join(matched_names) + '.') if missing_names and len(missing_names) < len(name_fields): summary_bits.append("Name components not found: " + ', '.join(missing_names) + '.') elif len(missing_names) == len(name_fields): summary_bits.append("No name components matched.") # Bank name / account type if any(isinstance(it, dict) and it.get('bank_name_matches') for it in bs_items): summary_bits.append("Bank name matches.") else: summary_bits.append("Bank name not matched.") if any(isinstance(it, dict) and it.get('account_type_matches') for it in bs_items): summary_bits.append("Account type matches.") else: summary_bits.append("Account type not matched.") # Arithmetic consistency and months coverage inconsistencies_total = sum(len(it.get('arithmetic_inconsistencies') or []) for it in bs_items if isinstance(it, dict)) if inconsistencies_total == 0 and any(isinstance(it, dict) and it.get('balances') for it in bs_items): summary_bits.append("Arithmetic appears consistent across provided months.") elif inconsistencies_total > 0: summary_bits.append(f"Arithmetic issues detected in {inconsistencies_total} month(s).") # Months coverage aggregated if base_response.aggregated_months_complete is True: summary_bits.append("All required recent months appear covered.") else: missing_months = base_response.aggregated_missing_months or [] if missing_months: summary_bits.append("Missing months: " + ', '.join(missing_months) + '.') # Per file quick notes if len(bs_items) == 1: single = bs_items[0] if isinstance(single, dict): covered = single.get('months_covered') or [] if covered: summary_bits.append("Months covered: " + ', '.join(covered) + '.') else: # List total months covered across all agg_cov = base_response.aggregated_months_covered or [] if agg_cov: summary_bits.append("Aggregated months covered: " + ', '.join(agg_cov) + '.') base_response.feedback_text = ' '.join(summary_bits) except Exception: pass # Build an overall user-friendly summary at top level for non-bank files if file_type != 'bank_statement': summary_bits: List[str] = [] # If we had doc types collected above (non-bank multi-file case) try: if 'Vehicle' not in (doc_types[0] if 'doc_types' in locals() and doc_types else ''): if doc_types: summary_bits.append(f"Some files appear to be {', '.join(doc_types)} rather than a vehicle document.") if 'is_hazy_any' in locals() and is_hazy_any: summary_bits.append("One or more files are a bit blurry.") except Exception: pass if fully: summary_bits.append("Matched fields: " + ', '.join(fully) + ".") if partial: diffs: List[str] = [] for r in (base_response.analysis_results or []): try: if r.field_name in partial and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") except Exception: continue if diffs: summary_bits.append("Differences: " + '; '.join(diffs) + ".") else: summary_bits.append("Some fields are present but differ: " + ', '.join(partial) + ".") if not_matched: summary_bits.append("Missing or not found: " + ', '.join(not_matched) + ".") if file_type != 'real_estate_documents': summary_bits.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") base_response_dict = base_response.model_dump() base_response_dict['overall_feedback_text'] = ' '.join(summary_bits) # Reconstruct pydantic model with added field not declared; instead attach via dynamic attribute try: setattr(base_response, 'overall_feedback_text', base_response_dict['overall_feedback_text']) except Exception: pass return base_response except Exception as e: return AddressMatchResponse( success=False, message=f"Error verifying documents: {str(e)}", file_url=None, analysis_results=[], overall_match=False if request_data.file_type != 'bank_statement' else None, mismatch_details=[str(e)], is_hazy=None, document_type_detected=None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, # Ensure we always pass plain strings (HttpUrl objects need str conversion) original_file_urls=[str(u) for u in request_data.file_urls], timestamp=datetime.now().isoformat() ) #for celery async def verify_multi_documents_celery( db: Session, request_data: MultiDocumentVerificationRequest, request: Request ) -> AddressMatchResponse: """Process multi-document verification using a unified request body. Internally reuses existing verify_address_with_document logic by joining file URLs. """ try: file_type = request_data.file_type # Join file urls with comma to leverage existing downstream logic (bank statements expect this format) # Cast HttpUrl objects to str explicitly to avoid join TypeError file_url_joined = ','.join([str(u) for u in request_data.file_urls]) # data is already parsed into BankStatementData or AddressData via Union in schema parsed = request_data.data # Special handling: for non-bank documents with multiple URLs, call analysis per file if file_type != 'bank_statement' and len(request_data.file_urls) > 1: # Concurrent per-file verification for non-bank multi-file requests batch_start = time.time() tasks = [verify_address_with_document(db, file_type, str(single_url), parsed, request) for single_url in request_data.file_urls] single_responses = await asyncio.gather(*tasks) aggregated_results: List[MatchResult] = [] aggregated_mismatches: List[str] = [] overall_match_all = True is_hazy_any: Optional[bool] = False doc_types: List[str] = [] per_file_payload: List[Dict] = [] for single_resp, single_url in zip(single_responses, request_data.file_urls): if single_resp.analysis_results: aggregated_results.extend(single_resp.analysis_results) if single_resp.mismatch_details: aggregated_mismatches.extend(single_resp.mismatch_details) if single_resp.is_hazy: is_hazy_any = True if single_resp.document_type_detected: doc_types.append(single_resp.document_type_detected) if single_resp.overall_match is False: overall_match_all = False pf_fully: List[str] = [] pf_partial: List[str] = [] pf_not: List[str] = [] if single_resp.analysis_results: for r in single_resp.analysis_results: if r.is_match: pf_fully.append(r.field_name) else: if r.found_in_document: pf_partial.append(r.field_name) else: pf_not.append(r.field_name) per_file_feedback = { 'fully_matched': sorted(set(pf_fully)), 'partially_matched': sorted(set(pf_partial) - set(pf_fully)), 'not_matched': sorted(set(pf_not) - set(pf_fully) - set(pf_partial)) } summary_parts: List[str] = [] if single_resp.document_type_detected and ('Vehicle' not in single_resp.document_type_detected): summary_parts.append(f"This looks like a {single_resp.document_type_detected}, not a vehicle document.") if single_resp.is_hazy: summary_parts.append("The image is a bit blurry.") if per_file_feedback.get('fully_matched'): summary_parts.append("We found these matching: " + ', '.join(per_file_feedback['fully_matched']) + ".") partials = per_file_feedback.get('partially_matched') or [] if partials: diffs: List[str] = [] for r in (single_resp.analysis_results or []): try: if r.field_name in partials and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") except Exception: continue if diffs: summary_parts.append("Differences: " + '; '.join(diffs) + ".") else: summary_parts.append("Some fields are present but differ: " + ', '.join(partials) + ".") missing = per_file_feedback.get('not_matched') or [] if missing: summary_parts.append("Missing or not found: " + ', '.join(missing) + ".") if file_type != 'real_estate_documents': summary_parts.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") user_friendly_feedback = ' '.join(summary_parts) per_file_payload.append({ 'file_url': str(single_url), 'analysis_results': [r.model_dump() for r in (single_resp.analysis_results or [])], 'mismatch_details': single_resp.mismatch_details or [], 'overall_match': single_resp.overall_match, 'is_hazy': single_resp.is_hazy, 'document_type_detected': single_resp.document_type_detected, 'feedback': per_file_feedback, 'feedback_text': user_friendly_feedback }) dedup: Dict[str, MatchResult] = {} for r in aggregated_results: existing = dedup.get(r.field_name) if existing is None: dedup[r.field_name] = r else: conf_order = {'high': 3, 'medium': 2, 'low': 1} def score(m: MatchResult): return (1 if m.is_match else 0, conf_order.get(m.confidence, 0)) if score(r) > score(existing): dedup[r.field_name] = r merged_results = list(dedup.values()) elapsed = time.time() - batch_start print(f"PERF: Concurrent multi-file verification ({len(request_data.file_urls)} files, type={file_type}) in {elapsed:.2f}s") base_response = AddressMatchResponse( success=True, message=f"Multi-file verification completed. Overall match: {overall_match_all}.", file_url=file_url_joined, analysis_results=merged_results, overall_match=overall_match_all, mismatch_details=sorted(set(aggregated_mismatches)), is_hazy=is_hazy_any, document_type_detected=doc_types[0] if doc_types else None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, original_file_urls=[str(u) for u in request_data.file_urls], per_file_results=per_file_payload, timestamp=datetime.now().isoformat() ) else: base_response = await verify_address_with_document(db, file_type, file_url_joined, parsed, request) # ensure original_file_urls populated if not base_response.original_file_urls: base_response.original_file_urls = [str(u) for u in request_data.file_urls] # Build feedback only for this unified endpoint without altering underlying verification logic fully: List[str] = [] partial: List[str] = [] not_matched: List[str] = [] if file_type == 'bank_statement' and base_response.bank_statement_results: # Aggregate across all statement result objects for item in base_response.bank_statement_results: if not isinstance(item, dict): continue name_matches = item.get('name_matches') or {} for k, v in name_matches.items(): if v: fully.append(k) else: not_matched.append(k) # Other boolean match fields if item.get('bank_name_matches') is True: fully.append('bank_name') elif item.get('bank_name_matches') is False: not_matched.append('bank_name') if item.get('account_type_matches') is True: fully.append('account_type') elif item.get('account_type_matches') is False: not_matched.append('account_type') if item.get('final_balance_matches') is True: fully.append('balance') elif item.get('final_balance_matches') is False: not_matched.append('balance') # Deduplicate fully = sorted(set(fully)) not_matched = sorted(set(not_matched) - set(fully)) else: # Use analysis_results list; partial = found_in_document present but is_match False if base_response.analysis_results: for r in base_response.analysis_results: if r.is_match: fully.append(r.field_name) else: if r.found_in_document: partial.append(r.field_name) else: not_matched.append(r.field_name) fully = sorted(set(fully)) partial = sorted(set(partial) - set(fully)) not_matched = sorted(set(not_matched) - set(fully) - set(partial)) # Attach feedback base_response.feedback = { 'fully_matched': fully, 'partially_matched': partial, 'not_matched': not_matched } # Build feedback_details for user-friendly explanations feedback_details: List[Dict[str, str]] = [] def explain(field: str, status: str, provided: Optional[str] = None, found: Optional[str] = None): if status == 'fully_matched': msg = f"Field '{field}' matches exactly." if provided else f"Field '{field}' verified." elif status == 'partially_matched': msg = f"Field '{field}' present but value differs (expected '{provided}' vs found '{found}')." if provided else f"Field '{field}' partially matched." else: msg = f"Field '{field}' not found or no match." if provided else f"Field '{field}' missing." feedback_details.append({"field": field, "status": status, "explanation": msg}) # We can use analysis_results for non-bank and bank aggregated sets for bank if file_type == 'bank_statement' and base_response.bank_statement_results: # derive provided vs found from each bank statement result item for item in base_response.bank_statement_results: fb = item.get('feedback') or {} for fld in fb.get('fully_matched', []): explain(fld, 'fully_matched') for fld in fb.get('partially_matched', []): explain(fld, 'partially_matched') for fld in fb.get('not_matched', []): explain(fld, 'not_matched') else: if base_response.analysis_results: for r in base_response.analysis_results: status = 'fully_matched' if r.field_name in fully else ('partially_matched' if r.field_name in partial else 'not_matched') explain(r.field_name, status, r.provided_value, r.found_in_document) base_response.feedback_details = feedback_details # If single-file (no per_file_results) and non-bank, build feedback_text if file_type != 'bank_statement' and not base_response.per_file_results: try: fully_local = base_response.feedback.get('fully_matched') if base_response.feedback else [] partial_local = base_response.feedback.get('partially_matched') if base_response.feedback else [] not_local = base_response.feedback.get('not_matched') if base_response.feedback else [] summary_bits: List[str] = [] if base_response.document_type_detected and 'Vehicle' not in base_response.document_type_detected: summary_bits.append(f"This looks like a {base_response.document_type_detected}, not a vehicle document.") if base_response.is_hazy: summary_bits.append("The image is a bit blurry.") if fully_local: summary_bits.append("Matched: " + ', '.join(fully_local) + ".") if partial_local: diffs: List[str] = [] for r in (base_response.analysis_results or []): if r.field_name in partial_local and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") if diffs: summary_bits.append("Differences: " + '; '.join(diffs) + ".") else: summary_bits.append("Some fields are present but differ: " + ', '.join(partial_local) + ".") if not_local: summary_bits.append("Missing or not found: " + ', '.join(not_local) + ".") if file_type != 'real_estate_documents': summary_bits.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") setattr(base_response, 'feedback_text', ' '.join(summary_bits)) except Exception: pass # Bank statement specific human-readable feedback (single or multi-file) if file_type == 'bank_statement' and base_response.bank_statement_results: try: bs_items = base_response.bank_statement_results or [] summary_bits: List[str] = [] # Balance match status any_balance_match = any(isinstance(it, dict) and it.get('final_balance_matches') for it in bs_items) if any_balance_match: summary_bits.append("Final balance matches the provided expected amount.") else: summary_bits.append("Final balance could not be verified or does not match.") # Name matches aggregation name_fields = ['first_name','middle_name','last_name'] matched_names = [] missing_names = [] for nf in name_fields: # consider matched if any file has nf True if any(isinstance(it, dict) and isinstance(it.get('name_matches'), dict) and it['name_matches'].get(nf) is True for it in bs_items): matched_names.append(nf) else: missing_names.append(nf) if matched_names: summary_bits.append("Name components matched: " + ', '.join(matched_names) + '.') if missing_names and len(missing_names) < len(name_fields): summary_bits.append("Name components not found: " + ', '.join(missing_names) + '.') elif len(missing_names) == len(name_fields): summary_bits.append("No name components matched.") # Bank name / account type if any(isinstance(it, dict) and it.get('bank_name_matches') for it in bs_items): summary_bits.append("Bank name matches.") else: summary_bits.append("Bank name not matched.") if any(isinstance(it, dict) and it.get('account_type_matches') for it in bs_items): summary_bits.append("Account type matches.") else: summary_bits.append("Account type not matched.") # Arithmetic consistency and months coverage inconsistencies_total = sum(len(it.get('arithmetic_inconsistencies') or []) for it in bs_items if isinstance(it, dict)) if inconsistencies_total == 0 and any(isinstance(it, dict) and it.get('balances') for it in bs_items): summary_bits.append("Arithmetic appears consistent across provided months.") elif inconsistencies_total > 0: summary_bits.append(f"Arithmetic issues detected in {inconsistencies_total} month(s).") # Months coverage aggregated if base_response.aggregated_months_complete is True: summary_bits.append("All required recent months appear covered.") else: missing_months = base_response.aggregated_missing_months or [] if missing_months: summary_bits.append("Missing months: " + ', '.join(missing_months) + '.') # Per file quick notes if len(bs_items) == 1: single = bs_items[0] if isinstance(single, dict): covered = single.get('months_covered') or [] if covered: summary_bits.append("Months covered: " + ', '.join(covered) + '.') else: # List total months covered across all agg_cov = base_response.aggregated_months_covered or [] if agg_cov: summary_bits.append("Aggregated months covered: " + ', '.join(agg_cov) + '.') base_response.feedback_text = ' '.join(summary_bits) except Exception: pass # Build an overall user-friendly summary at top level for non-bank files if file_type != 'bank_statement': summary_bits: List[str] = [] # If we had doc types collected above (non-bank multi-file case) try: if 'Vehicle' not in (doc_types[0] if 'doc_types' in locals() and doc_types else ''): if doc_types: summary_bits.append(f"Some files appear to be {', '.join(doc_types)} rather than a vehicle document.") if 'is_hazy_any' in locals() and is_hazy_any: summary_bits.append("One or more files are a bit blurry.") except Exception: pass if fully: summary_bits.append("Matched fields: " + ', '.join(fully) + ".") if partial: diffs: List[str] = [] for r in (base_response.analysis_results or []): try: if r.field_name in partial and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") except Exception: continue if diffs: summary_bits.append("Differences: " + '; '.join(diffs) + ".") else: summary_bits.append("Some fields are present but differ: " + ', '.join(partial) + ".") if not_matched: summary_bits.append("Missing or not found: " + ', '.join(not_matched) + ".") if file_type != 'real_estate_documents': summary_bits.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") base_response_dict = base_response.model_dump() base_response_dict['overall_feedback_text'] = ' '.join(summary_bits) # Reconstruct pydantic model with added field not declared; instead attach via dynamic attribute try: setattr(base_response, 'overall_feedback_text', base_response_dict['overall_feedback_text']) except Exception: pass return base_response except Exception as e: return AddressMatchResponse( success=False, message=f"Error verifying documents: {str(e)}", file_url=None, analysis_results=[], overall_match=False if request_data.file_type != 'bank_statement' else None, mismatch_details=[str(e)], is_hazy=None, document_type_detected=None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, # Ensure we always pass plain strings (HttpUrl objects need str conversion) original_file_urls=[str(u) for u in request_data.file_urls], timestamp=datetime.now().isoformat() ) #celery without db async def verify_multi_documents_celery_withoutdb( db: Session, request_data: MultiDocumentVerificationRequest, request: Request ) -> AddressMatchResponse: """Process multi-document verification using a unified request body. Internally reuses existing verify_address_with_document logic by joining file URLs. """ try: file_type = request_data.file_type # Join file urls with comma to leverage existing downstream logic (bank statements expect this format) # Cast HttpUrl objects to str explicitly to avoid join TypeError file_url_joined = ','.join([str(u) for u in request_data.file_urls]) # data is already parsed into BankStatementData or AddressData via Union in schema parsed = request_data.data # Special handling: for non-bank documents with multiple URLs, call analysis per file if file_type != 'bank_statement' and len(request_data.file_urls) > 1: # Concurrent per-file verification for non-bank multi-file requests batch_start = time.time() tasks = [verify_address_with_document(db, file_type, str(single_url), parsed, request) for single_url in request_data.file_urls] single_responses = await asyncio.gather(*tasks) aggregated_results: List[MatchResult] = [] aggregated_mismatches: List[str] = [] overall_match_all = True is_hazy_any: Optional[bool] = False doc_types: List[str] = [] per_file_payload: List[Dict] = [] for single_resp, single_url in zip(single_responses, request_data.file_urls): if single_resp.analysis_results: aggregated_results.extend(single_resp.analysis_results) if single_resp.mismatch_details: aggregated_mismatches.extend(single_resp.mismatch_details) if single_resp.is_hazy: is_hazy_any = True if single_resp.document_type_detected: doc_types.append(single_resp.document_type_detected) if single_resp.overall_match is False: overall_match_all = False pf_fully: List[str] = [] pf_partial: List[str] = [] pf_not: List[str] = [] if single_resp.analysis_results: for r in single_resp.analysis_results: if r.is_match: pf_fully.append(r.field_name) else: if r.found_in_document: pf_partial.append(r.field_name) else: pf_not.append(r.field_name) per_file_feedback = { 'fully_matched': sorted(set(pf_fully)), 'partially_matched': sorted(set(pf_partial) - set(pf_fully)), 'not_matched': sorted(set(pf_not) - set(pf_fully) - set(pf_partial)) } summary_parts: List[str] = [] if single_resp.document_type_detected and ('Vehicle' not in single_resp.document_type_detected): summary_parts.append(f"This looks like a {single_resp.document_type_detected}, not a vehicle document.") if single_resp.is_hazy: summary_parts.append("The image is a bit blurry.") if per_file_feedback.get('fully_matched'): summary_parts.append("We found these matching: " + ', '.join(per_file_feedback['fully_matched']) + ".") partials = per_file_feedback.get('partially_matched') or [] if partials: diffs: List[str] = [] for r in (single_resp.analysis_results or []): try: if r.field_name in partials and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") except Exception: continue if diffs: summary_parts.append("Differences: " + '; '.join(diffs) + ".") else: summary_parts.append("Some fields are present but differ: " + ', '.join(partials) + ".") missing = per_file_feedback.get('not_matched') or [] if missing: summary_parts.append("Missing or not found: " + ', '.join(missing) + ".") if file_type != 'real_estate_documents': summary_parts.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") user_friendly_feedback = ' '.join(summary_parts) per_file_payload.append({ 'file_url': str(single_url), 'analysis_results': [r.model_dump() for r in (single_resp.analysis_results or [])], 'mismatch_details': single_resp.mismatch_details or [], 'overall_match': single_resp.overall_match, 'is_hazy': single_resp.is_hazy, 'document_type_detected': single_resp.document_type_detected, 'feedback': per_file_feedback, 'feedback_text': user_friendly_feedback }) dedup: Dict[str, MatchResult] = {} for r in aggregated_results: existing = dedup.get(r.field_name) if existing is None: dedup[r.field_name] = r else: conf_order = {'high': 3, 'medium': 2, 'low': 1} def score(m: MatchResult): return (1 if m.is_match else 0, conf_order.get(m.confidence, 0)) if score(r) > score(existing): dedup[r.field_name] = r merged_results = list(dedup.values()) elapsed = time.time() - batch_start print(f"PERF: Concurrent multi-file verification ({len(request_data.file_urls)} files, type={file_type}) in {elapsed:.2f}s") base_response = AddressMatchResponse( success=True, message=f"Multi-file verification completed. Overall match: {overall_match_all}.", file_url=file_url_joined, analysis_results=merged_results, overall_match=overall_match_all, mismatch_details=sorted(set(aggregated_mismatches)), is_hazy=is_hazy_any, document_type_detected=doc_types[0] if doc_types else None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, original_file_urls=[str(u) for u in request_data.file_urls], per_file_results=per_file_payload, timestamp=datetime.now().isoformat() ) else: base_response = await verify_address_with_document(db, file_type, file_url_joined, parsed, request) # ensure original_file_urls populated if not base_response.original_file_urls: base_response.original_file_urls = [str(u) for u in request_data.file_urls] # Build feedback only for this unified endpoint without altering underlying verification logic fully: List[str] = [] partial: List[str] = [] not_matched: List[str] = [] if file_type == 'bank_statement' and base_response.bank_statement_results: # Aggregate across all statement result objects for item in base_response.bank_statement_results: if not isinstance(item, dict): continue name_matches = item.get('name_matches') or {} for k, v in name_matches.items(): if v: fully.append(k) else: not_matched.append(k) # Other boolean match fields if item.get('bank_name_matches') is True: fully.append('bank_name') elif item.get('bank_name_matches') is False: not_matched.append('bank_name') if item.get('account_type_matches') is True: fully.append('account_type') elif item.get('account_type_matches') is False: not_matched.append('account_type') if item.get('final_balance_matches') is True: fully.append('balance') elif item.get('final_balance_matches') is False: not_matched.append('balance') # Deduplicate fully = sorted(set(fully)) not_matched = sorted(set(not_matched) - set(fully)) else: # Use analysis_results list; partial = found_in_document present but is_match False if base_response.analysis_results: for r in base_response.analysis_results: if r.is_match: fully.append(r.field_name) else: if r.found_in_document: partial.append(r.field_name) else: not_matched.append(r.field_name) fully = sorted(set(fully)) partial = sorted(set(partial) - set(fully)) not_matched = sorted(set(not_matched) - set(fully) - set(partial)) # Attach feedback base_response.feedback = { 'fully_matched': fully, 'partially_matched': partial, 'not_matched': not_matched } # Build feedback_details for user-friendly explanations feedback_details: List[Dict[str, str]] = [] def explain(field: str, status: str, provided: Optional[str] = None, found: Optional[str] = None): if status == 'fully_matched': msg = f"Field '{field}' matches exactly." if provided else f"Field '{field}' verified." elif status == 'partially_matched': msg = f"Field '{field}' present but value differs (expected '{provided}' vs found '{found}')." if provided else f"Field '{field}' partially matched." else: msg = f"Field '{field}' not found or no match." if provided else f"Field '{field}' missing." feedback_details.append({"field": field, "status": status, "explanation": msg}) # We can use analysis_results for non-bank and bank aggregated sets for bank if file_type == 'bank_statement' and base_response.bank_statement_results: # derive provided vs found from each bank statement result item for item in base_response.bank_statement_results: fb = item.get('feedback') or {} for fld in fb.get('fully_matched', []): explain(fld, 'fully_matched') for fld in fb.get('partially_matched', []): explain(fld, 'partially_matched') for fld in fb.get('not_matched', []): explain(fld, 'not_matched') else: if base_response.analysis_results: for r in base_response.analysis_results: status = 'fully_matched' if r.field_name in fully else ('partially_matched' if r.field_name in partial else 'not_matched') explain(r.field_name, status, r.provided_value, r.found_in_document) base_response.feedback_details = feedback_details # If single-file (no per_file_results) and non-bank, build feedback_text if file_type != 'bank_statement' and not base_response.per_file_results: try: fully_local = base_response.feedback.get('fully_matched') if base_response.feedback else [] partial_local = base_response.feedback.get('partially_matched') if base_response.feedback else [] not_local = base_response.feedback.get('not_matched') if base_response.feedback else [] summary_bits: List[str] = [] if base_response.document_type_detected and 'Vehicle' not in base_response.document_type_detected: summary_bits.append(f"This looks like a {base_response.document_type_detected}, not a vehicle document.") if base_response.is_hazy: summary_bits.append("The image is a bit blurry.") if fully_local: summary_bits.append("Matched: " + ', '.join(fully_local) + ".") if partial_local: diffs: List[str] = [] for r in (base_response.analysis_results or []): if r.field_name in partial_local and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") if diffs: summary_bits.append("Differences: " + '; '.join(diffs) + ".") else: summary_bits.append("Some fields are present but differ: " + ', '.join(partial_local) + ".") if not_local: summary_bits.append("Missing or not found: " + ', '.join(not_local) + ".") if file_type != 'real_estate_documents': summary_bits.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") setattr(base_response, 'feedback_text', ' '.join(summary_bits)) except Exception: pass # Bank statement specific human-readable feedback (single or multi-file) if file_type == 'bank_statement' and base_response.bank_statement_results: try: bs_items = base_response.bank_statement_results or [] summary_bits: List[str] = [] # Balance match status any_balance_match = any(isinstance(it, dict) and it.get('final_balance_matches') for it in bs_items) if any_balance_match: summary_bits.append("Final balance matches the provided expected amount.") else: summary_bits.append("Final balance could not be verified or does not match.") # Name matches aggregation name_fields = ['first_name','middle_name','last_name'] matched_names = [] missing_names = [] for nf in name_fields: # consider matched if any file has nf True if any(isinstance(it, dict) and isinstance(it.get('name_matches'), dict) and it['name_matches'].get(nf) is True for it in bs_items): matched_names.append(nf) else: missing_names.append(nf) if matched_names: summary_bits.append("Name components matched: " + ', '.join(matched_names) + '.') if missing_names and len(missing_names) < len(name_fields): summary_bits.append("Name components not found: " + ', '.join(missing_names) + '.') elif len(missing_names) == len(name_fields): summary_bits.append("No name components matched.") # Bank name / account type if any(isinstance(it, dict) and it.get('bank_name_matches') for it in bs_items): summary_bits.append("Bank name matches.") else: summary_bits.append("Bank name not matched.") if any(isinstance(it, dict) and it.get('account_type_matches') for it in bs_items): summary_bits.append("Account type matches.") else: summary_bits.append("Account type not matched.") # Arithmetic consistency and months coverage inconsistencies_total = sum(len(it.get('arithmetic_inconsistencies') or []) for it in bs_items if isinstance(it, dict)) if inconsistencies_total == 0 and any(isinstance(it, dict) and it.get('balances') for it in bs_items): summary_bits.append("Arithmetic appears consistent across provided months.") elif inconsistencies_total > 0: summary_bits.append(f"Arithmetic issues detected in {inconsistencies_total} month(s).") # Months coverage aggregated if base_response.aggregated_months_complete is True: summary_bits.append("All required recent months appear covered.") else: missing_months = base_response.aggregated_missing_months or [] if missing_months: summary_bits.append("Missing months: " + ', '.join(missing_months) + '.') # Per file quick notes if len(bs_items) == 1: single = bs_items[0] if isinstance(single, dict): covered = single.get('months_covered') or [] if covered: summary_bits.append("Months covered: " + ', '.join(covered) + '.') else: # List total months covered across all agg_cov = base_response.aggregated_months_covered or [] if agg_cov: summary_bits.append("Aggregated months covered: " + ', '.join(agg_cov) + '.') base_response.feedback_text = ' '.join(summary_bits) except Exception: pass # Build an overall user-friendly summary at top level for non-bank files if file_type != 'bank_statement': summary_bits: List[str] = [] # If we had doc types collected above (non-bank multi-file case) try: if 'Vehicle' not in (doc_types[0] if 'doc_types' in locals() and doc_types else ''): if doc_types: summary_bits.append(f"Some files appear to be {', '.join(doc_types)} rather than a vehicle document.") if 'is_hazy_any' in locals() and is_hazy_any: summary_bits.append("One or more files are a bit blurry.") except Exception: pass if fully: summary_bits.append("Matched fields: " + ', '.join(fully) + ".") if partial: diffs: List[str] = [] for r in (base_response.analysis_results or []): try: if r.field_name in partial and r.found_in_document is not None and str(r.found_in_document) != str(r.provided_value): diffs.append(f"{r.field_name} differs (expected '{r.provided_value}' vs found '{r.found_in_document}')") except Exception: continue if diffs: summary_bits.append("Differences: " + '; '.join(diffs) + ".") else: summary_bits.append("Some fields are present but differ: " + ', '.join(partial) + ".") if not_matched: summary_bits.append("Missing or not found: " + ', '.join(not_matched) + ".") if file_type != 'real_estate_documents': summary_bits.append("Please upload a clear vehicle registration or insurance document showing the vehicle number, model, mileage, and value.") base_response_dict = base_response.model_dump() base_response_dict['overall_feedback_text'] = ' '.join(summary_bits) # Reconstruct pydantic model with added field not declared; instead attach via dynamic attribute try: setattr(base_response, 'overall_feedback_text', base_response_dict['overall_feedback_text']) except Exception: pass return base_response except Exception as e: return AddressMatchResponse( success=False, message=f"Error verifying documents: {str(e)}", file_url=None, analysis_results=[], overall_match=False if request_data.file_type != 'bank_statement' else None, mismatch_details=[str(e)], is_hazy=None, document_type_detected=None, bank_statement_results=None, aggregated_months_covered=None, aggregated_missing_months=None, aggregated_months_complete=None, # Ensure we always pass plain strings (HttpUrl objects need str conversion) original_file_urls=[str(u) for u in request_data.file_urls], timestamp=datetime.now().isoformat() )