import aiohttp import json import base64 from typing import Dict, List from datetime import datetime from src.core.config import settings from src.modules.prompt.schemas.responses import MatchResult from src.modules.prompt.prompts import ( address_document_prompt, lease_documents_prompt, realestate_document_prompt, vehicle_prompt, retirement_prompt, creditor_prompt, priority_unsecured_claim_prompt, insurance_prompt, executory_contract_prompt, third_party_claim_prompt, employment_prompt, net_income_prompt, gross_overtime_prompt, rental_income_prompt, royalties_income_prompt, pension_income_prompt, family_support_income_prompt, security_income_prompt, social_security_act_benefits_prompt, public_assistance_prompt, other_income_prompt, spouse_income_prompt, rent_home_records_prompt, utilities_records_prompt, transportation_records_prompt, vehicle_insurance_records_prompt, taxes_records_prompt, installment_payments_records_prompt, alimony_support_records_prompt, insurance_records_prompt, retirement_contributions_records_prompt, other_expenses_records_prompt, insider_payment_documents_prompt, paid_transferred_property_documents_prompt, paid_transfer_help_creditors_documents_prompt, sold_traded_transferred_property_documents_prompt, owe_money_to_creditors_documents_prompt, property_taken_as_collateral_debt_documents_prompt, court_action_documents_prompt, property_repossessed_documents_prompt, creditors_legal_process_claims_documents_prompt, gave_gifts_over_600_last_year_documents_prompt, financial_accounts_changed_last_year_documents_prompt, housing_mortgage_records_prompt, public_transportation_records_prompt, secured_debt_payments_records_prompt, other_necessary_expenses_records_prompt, priority_claims_records_prompt, charitable_contributions_records_prompt, additional_security_records_prompt, home_energy_costs_records_prompt, education_dependents_records_prompt, elderly_disabled_care_records_prompt, health_disability_insurance_records_prompt, telecommunications_records_prompt, healthcare_beyond_standards_records_prompt, childcare_expenses_records_prompt, education_expenses_records_prompt, court_ordered_payments_records_prompt, life_insurance_records_prompt, involuntary_deductions_records_prompt, vehicle_ownership_records_prompt, healthcare_expenses_records_prompt, housing_insurance_records_prompt, vehicle_operation_records_prompt, taxes_actual_records_prompt, spouse_business_pl_documents_prompt, self_employment_pl_documents_prompt, creditor_payment_documents_prompt, insider_benefit_documents_prompt, insider_payment_documents_prompt, insider_payment_documents_prompt_2, ) from src.core.logging import logger from src.utils.pdf_parser import get_pdf_parser from src.utils.pusher_config import pusher_client try: import fitz # PyMuPDF for PDF to image conversion except ImportError: # Graceful degradation if not installed fitz = None class AzureOpenAIClient: """Azure OpenAI client for document-driven structured verification and classification""" def __init__(self, max_tokens=None, temperature=None, top_p=None, response_format=None): self.endpoint = settings.AZURE_OPENAI_ENDPOINT self.api_key = settings.AZURE_OPENAI_KEY self.deployment = settings.AZURE_OPENAI_DEPLOYMENT self.api_version = settings.AZURE_OPENAI_VERSION self.url = f"{self.endpoint}/openai/deployments/{self.deployment}/chat/completions?api-version={self.api_version}" # Configurable API parameters with defaults self.max_tokens = max_tokens if max_tokens is not None else 2000 self.temperature = temperature if temperature is not None else 0.3 self.top_p = top_p if top_p is not None else 0.5 self.response_format = response_format if response_format is not None else {"type": "json_object"} async def verify_document(self, file_type: str, payload: Dict, file_url: str, extracted_text: str = None, request_id: str = None) -> Dict: """Generic document verification entry point. For now supports file_type == 'residence_documents' which performs name + address field matching, haze detection, and document type classification. Args: file_type: Document type for prompt selection payload: Data to verify against file_url: Document URL (can be empty if extracted_text provided) extracted_text: Pre-extracted text (from chunking). If provided, skips LlamaParse. """ logger.info(f"Verifying document: {file_type} | URL: {file_url}") # ===== LLAMAPARSE EXTRACTION - SKIP IF extracted_text ALREADY PROVIDED ===== llamaparse_success = False if extracted_text: # Text already provided (from chunking pipeline) llamaparse_success = True logger.info(f"[Chunk Mode] Using pre-extracted text ({len(extracted_text)} chars)") else: # Original flow - extract using LlamaParse logger.info(f"STARTING LlamaParse extraction for: {file_url}") try: # Create fresh parser instance to avoid connection pool conflicts parser = get_pdf_parser() logger.info(f"LlamaParse instance created, calling parse_url()...") documents = parser.parse_url(file_url) logger.info(f"LlamaParse parse_url() returned {len(documents)} documents") extracted_text = parser.get_text(documents) llamaparse_success = len(extracted_text) > 0 logger.info(f"LlamaParse SUCCESS: Extracted {len(extracted_text)} chars - Will be sent to Vision API") print(f"LLAMAPARSE EXTRACTED TEXT ({len(extracted_text)} chars): {extracted_text[:200]}...") except Exception as e: # LlamaParse is the ONLY text extraction method - no OCR fallback logger.error(f"LlamaParse extraction FAILED: {e}. No text extraction available.") print(f"LLAMAPARSE FAILED: {e}") extracted_text = None llamaparse_success = False # Determine prompt set if file_type == "residence_documents": logger.info(f"MATCHED: residence_documents") system_prompt = self._create_residence_document_prompt() user_message = self._create_residence_user_message(payload, file_url, extracted_text) elif file_type == "billing_address_document": logger.info(f"MATCHED: billing_address_document") system_prompt = self._create_residence_document_prompt() user_message = self._create_residence_user_message(payload, file_url, extracted_text) elif file_type == "vehicle_document": system_prompt = self._create_vehicle_document_prompt() user_message = self._create_vehicle_user_message(payload, file_url, extracted_text) elif file_type == "bank_statement": system_prompt = self._create_bank_statement_prompt() user_message = self._create_bank_statement_user_message(payload, file_url, extracted_text) elif file_type == "business_documents": logger.info(f"MATCHED: business_documents") system_prompt = self._create_business_document_prompt() user_message = self._create_business_user_message(payload, file_url, extracted_text) elif file_type == "address_documents": logger.info(f"MATCHED: address_documents") system_prompt = address_document_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "ssn_documents": logger.info(f"MATCHED: ssn_documents") system_prompt = self._create_address_document_prompt() user_message = self._create_address_user_message(payload, file_url, extracted_text) elif file_type == "lease_documents": logger.info(f"MATCHED: lease_documents") system_prompt = lease_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "real_estate_documents": logger.info(f"MATCHED: real_estate_documents") system_prompt = realestate_document_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "recreational_vehicle_documents": logger.info(f"MATCHED: recreational_vehicle_documents") system_prompt = vehicle_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "vehicle_documents": logger.info(f"MATCHED: vehicle_documents") system_prompt = vehicle_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "equipment_documents": logger.info(f"MATCHED: equipment_documents") system_prompt = vehicle_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "retirement_documents": logger.info(f"MATCHED: retirement_documents") system_prompt = retirement_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "creditor_documents": logger.info(f"MATCHED: creditor_documents") system_prompt = creditor_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "priority_unsecured_claim_documents": logger.info(f"MATCHED: priority_unsecured_claim_documents") system_prompt = priority_unsecured_claim_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "insurance_policies": logger.info(f"MATCHED: insurance_policies") system_prompt = insurance_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "executory_contract_documents": logger.info(f"MATCHED: executory_contract_documents") system_prompt = executory_contract_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "third_party_claim_documents": logger.info(f"MATCHED: third_party_claim_documents") system_prompt = third_party_claim_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type in ["employment_documents", "spouse_employment_documents"]: logger.info(f"MATCHED: employment_documents") system_prompt = employment_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "net_income_documents": logger.info(f"MATCHED: net_income_documents") system_prompt = net_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "gross_overtime_documents": logger.info(f"MATCHED: gross_overtime_documents") system_prompt = gross_overtime_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "rental_income_documents": logger.info(f"MATCHED: rental_income_documents") system_prompt = rental_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "royalties_income_documents": logger.info(f"MATCHED: royalties_income_documents") system_prompt = royalties_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "pension_income_documents": logger.info(f"Pension Income Prompt: {pension_income_prompt}") system_prompt = pension_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "family_support_income_documents": logger.info(f"MATCHED: family_support_income_documents") system_prompt = family_support_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "security_income_documents": logger.info(f"MATCHED: security_income_documents") system_prompt = security_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "social_security_act_benefits_documents": logger.info(f"MATCHED: social_security_act_benefits_documents") system_prompt = social_security_act_benefits_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "public_assistance_documents": logger.info(f"MATCHED: public_assistance_documents") system_prompt = public_assistance_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "other_income_documents": logger.info(f"MATCHED: other_income_documents") system_prompt = other_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "spouse_income_documents": logger.info(f"MATCHED: spouse_income_documents") system_prompt = spouse_income_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "rent_home_records": logger.info(f"MATCHED: rent_home_records") system_prompt = rent_home_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "utilities_records": logger.info(f"MATCHED: utilities_records") system_prompt = utilities_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "transportation_records": logger.info(f"MATCHED: transportation_records") system_prompt = transportation_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "vehicle_insurance_records": logger.info(f"MATCHED: vehicle_insurance_records") system_prompt = vehicle_insurance_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "taxes_records": logger.info(f"MATCHED: taxes_records") system_prompt = taxes_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "installment_payments_records": logger.info(f"MATCHED: installment_payments_records") system_prompt = installment_payments_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "alimony_support_records": logger.info(f"MATCHED: alimony_support_records") system_prompt = alimony_support_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "insurance_records": logger.info(f"MATCHED: insurance_records") system_prompt = insurance_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "retirement_contributions_records": logger.info(f"MATCHED: retirement_contributions_records") system_prompt = retirement_contributions_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "other_expenses_records": logger.info(f"MATCHED: other_expenses_records") system_prompt = other_expenses_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "insider_payment_documents": logger.info(f"MATCHED: insider_payment_documents") system_prompt = insider_payment_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "paid_transferred_property_documents": logger.info(f"MATCHED: paid_transferred_property_documents") system_prompt = paid_transferred_property_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "paid_transfer_help_creditors_documents": logger.info(f"MATCHED: paid_transfer_help_creditors_documents") system_prompt = paid_transfer_help_creditors_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "sold_traded_transferred_property_documents": logger.info(f"MATCHED: sold_traded_transferred_property_documents") system_prompt = sold_traded_transferred_property_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "owe_money_to_creditors_documents": logger.info(f"MATCHED: owe_money_to_creditors_documents") system_prompt = owe_money_to_creditors_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "property_taken_as_collateral_debt_documents": logger.info(f"MATCHED: property_taken_as_collateral_debt_documents") system_prompt = property_taken_as_collateral_debt_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "court_action_documents": logger.info(f"MATCHED: court_action_documents") system_prompt = court_action_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "property_repossessed_documents": logger.info(f"MATCHED: property_repossessed_documents") system_prompt = property_repossessed_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "creditors_legal_process_claims_documents": logger.info(f"MATCHED: creditors_legal_process_claims_documents") system_prompt = creditors_legal_process_claims_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "gave_gifts_over_600_last_year_documents": logger.info(f"MATCHED: gave_gifts_over_600_last_year_documents") system_prompt = gave_gifts_over_600_last_year_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "financial_accounts_changed_last_year_documents": logger.info(f"MATCHED: financial_accounts_changed_last_year_documents") system_prompt = financial_accounts_changed_last_year_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "housing_mortgage_records": logger.info(f"MATCHED: housing_mortgage_records") system_prompt = housing_mortgage_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "public_transportation_records": logger.info(f"MATCHED: public_transportation_records") system_prompt = public_transportation_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "secured_debt_payments_records": logger.info(f"MATCHED: secured_debt_payments_records") system_prompt = secured_debt_payments_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "other_necessary_expenses_records": logger.info(f"MATCHED: other_necessary_expenses_records") system_prompt = other_necessary_expenses_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "priority_claims_records": logger.info(f"MATCHED: priority_claims_records") system_prompt = priority_claims_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "charitable_contributions_records": logger.info(f"MATCHED: charitable_contributions_records") system_prompt = charitable_contributions_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "additional_security_records": logger.info(f"MATCHED: additional_security_records") system_prompt = additional_security_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "home_energy_costs_records": logger.info(f"MATCHED: home_energy_costs_records") system_prompt = home_energy_costs_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "education_dependents_records": logger.info(f"MATCHED: education_dependents_records") system_prompt = education_dependents_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "elderly_disabled_care_records": logger.info(f"MATCHED: elderly_disabled_care_records") system_prompt = elderly_disabled_care_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "health_disability_insurance_records": logger.info(f"MATCHED: health_disability_insurance_records") system_prompt = health_disability_insurance_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "telecommunications_records": logger.info(f"MATCHED: telecommunications_records") system_prompt = telecommunications_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "healthcare_beyond_standards_records": logger.info(f"MATCHED: healthcare_beyond_standards_records") system_prompt = healthcare_beyond_standards_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "childcare_expenses_records": logger.info(f"MATCHED: childcare_expenses_records") system_prompt = childcare_expenses_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "education_expenses_records": logger.info(f"MATCHED: education_expenses_records") system_prompt = education_expenses_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "court_ordered_payments_records": logger.info(f"MATCHED: court_ordered_payments_records") system_prompt = court_ordered_payments_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "life_insurance_records": logger.info(f"MATCHED: life_insurance_records") system_prompt = life_insurance_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "involuntary_deductions_records": logger.info(f"MATCHED: involuntary_deductions_records") system_prompt = involuntary_deductions_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "vehicle_ownership_records": logger.info(f"MATCHED: vehicle_ownership_records") system_prompt = vehicle_ownership_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "healthcare_expenses_records": logger.info(f"MATCHED: healthcare_expenses_records") system_prompt = healthcare_expenses_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "housing_insurance_records": logger.info(f"MATCHED: housing_insurance_records") system_prompt = housing_insurance_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "vehicle_operation_records": logger.info(f"MATCHED: vehicle_operation_records") system_prompt = vehicle_operation_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "taxes_actual_records": logger.info(f"MATCHED: taxes_actual_records") system_prompt = taxes_actual_records_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "spouse_business_pl_documents": logger.info(f"MATCHED: spouse_business_pl_documents") system_prompt = spouse_business_pl_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "self_employment_pl_documents": logger.info(f"MATCHED: self_employment_pl_documents") system_prompt = self_employment_pl_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "creditor_payment_documents": logger.info(f"MATCHED: creditor_payment_documents") system_prompt = creditor_payment_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "insider_benefit_documents": logger.info(f"MATCHED: insider_benefit_documents") system_prompt = insider_benefit_documents_prompt user_message = self._create_response_message(payload, file_url, extracted_text) elif file_type == "insider_payment_documents": logger.info(f"MATCHED: insider_payment_documents") system_prompt = insider_payment_documents_prompt_2 user_message = self._create_response_message(payload, file_url, extracted_text) else: logger.warning(f"Unsupported file type {file_type} for document verification.") system_prompt = self._create_generic_prompt(file_type) user_message = self._create_generic_user_message(payload, file_url) # Log which prompt is being used (helpful for debugging) prompt_preview = system_prompt[:200] if isinstance(system_prompt, str) else str(system_prompt)[:200] logger.info(f"[PROMPT SELECTED] file_type='{file_type}' | prompt_preview: {prompt_preview}...") print(f"\n{'='*60}") print(f"[PROMPT] file_type: {file_type}") print(f"[PROMPT] system_prompt length: {len(system_prompt) if isinstance(system_prompt, str) else 'N/A'}") print(f"[PROMPT] preview: {prompt_preview}...") print(f"{'='*60}\n") # PDF handling (supports comma-separated URLs). Convert first 2 pages of each PDF. if file_type == "bank_statement": urls = [u.strip() for u in file_url.split(',') if u.strip()] rebuilt: List[Dict] = [] # Preserve initial text blocks (meta + per file listing) => generated by user_message builder text_blocks = [b for b in user_message if b.get('type') == 'text'] rebuilt.extend(text_blocks) for idx, u in enumerate(urls): if u.lower().endswith('.pdf'): if fitz is None: raise Exception("PDF provided but PyMuPDF not installed. Add PyMuPDF to requirements.") try: pdf_images = await self._pdf_url_to_data_images(u, max_pages=2) for data_uri in pdf_images: rebuilt.append({"type": "image_url", "image_url": {"url": data_uri, "detail": "high"}}) except Exception as pdf_err: # Add a placeholder text block indicating failure for this file rebuilt.append({"type": "text", "text": f"FILE_INDEX {idx} URL {u} PDF_CONVERSION_ERROR {pdf_err}"}) else: # Non-PDF: keep original direct image reference rebuilt.append({"type": "image_url", "image_url": {"url": u, "detail": "high"}}) user_message = rebuilt else: # Single (non-bank) PDF case if file_url.lower().endswith('.pdf'): if fitz is None: raise Exception("PDF provided but PyMuPDF not installed. Add PyMuPDF to requirements.") try: pdf_images = await self._pdf_url_to_data_images(file_url, max_pages=2) # Handle both string and list message formats if isinstance(user_message, str): # String format: wrap it in text block text_blocks = [{"type": "text", "text": user_message}] else: # List format: extract text blocks text_blocks = [b for b in user_message if b.get('type') == 'text'] user_message = text_blocks + [ {"type": "image_url", "image_url": {"url": data_uri, "detail": "high"}} for data_uri in pdf_images ] except Exception as pdf_err: raise Exception(f"Failed to process PDF for vision: {pdf_err}") # Added temparature, top_p, seed for better response consistency body = { "model": self.deployment, "max_tokens": self.max_tokens, "temperature": self.temperature, "top_p": self.top_p, "response_format": self.response_format, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message} ] } headers = { "api-key": self.api_key, "Content-Type": "application/json" } try: logger.info(f"Calling Azure OpenAI Vision API (with LlamaParse-extracted text)...") async with aiohttp.ClientSession() as session: async with session.post(self.url, headers=headers, json=body) as response: if not response.ok: error_text = await response.text() logger.critical(f"Azure OpenAI API error for {file_type}: {response.status} - {error_text}") # Check for content filter jailbreak error if "jailbreak" in error_text.lower(): # Send Pusher notification - this is a document issue, not server issue if pusher_client and request_id: try: logger.info(f"[JAILBREAK] Sending Pusher notification for request_id: {request_id}") pusher_client.trigger('notification-channel', 'notification-event', { 'status': False, 'message':'Please verify the document', 'request_id': request_id }) logger.info(f"[JAILBREAK] Pusher notification sent successfully for request_id: {request_id}") except Exception as pusher_err: logger.error(f"[JAILBREAK] Pusher notification failed for request_id {request_id}: {pusher_err}") # Return structured error for frontend return { "success": True, # HTTP 200 - API call succeeded "error": True, # Explicit error flag for frontend "error_type": "content_filter", "message": "Verification failed. Content filter triggered.", "overall_feedback_text": "Please check the document.", "analysis_results": [], "overall_match": False, "confidence_score": "0%", "document_type": "Error", "mismatch_summary": ["Content filter triggered"], "is_hazy": False, "_llamaparse_success": llamaparse_success, "error_code": response.status } # Return structured error for other API errors return { "success": True, # HTTP 200 - API call succeeded "error": True, # Explicit error flag for frontend "error_type": "api_error", "message": f"Verification failed. Azure OpenAI API error: {response.status}", "overall_feedback_text": f"API Error: {error_text[:200]}", "analysis_results": [], "overall_match": False, "confidence_score": "0%", "document_type": "Error", "mismatch_summary": ["API request failed"], "is_hazy": False, "_llamaparse_success": llamaparse_success, "error_code": response.status } result = await response.json() content = result["choices"][0]["message"]["content"] # Parse the JSON response analysis_result = json.loads(content) # Add LlamaParse success flag to result analysis_result['_llamaparse_success'] = llamaparse_success logger.info(f"Verification complete") return analysis_result except Exception as e: # Check if it's a jailbreak filter error if "jailbreak" in str(e).lower(): logger.warning(f"Content filter triggered for {file_type}: jailbreak detected") # Send Pusher notification - this is a document issue, not server issue if pusher_client and request_id: try: logger.info(f"[JAILBREAK] Sending Pusher notification for request_id: {request_id}") pusher_client.trigger('notification-channel', 'notification-event', { 'status': False, 'message': 'Content filter triggered: Please check the document', 'request_id': request_id }) logger.info(f"[JAILBREAK] Pusher notification sent successfully for request_id: {request_id}") except Exception as pusher_err: logger.error(f"[JAILBREAK] Pusher notification failed for request_id {request_id}: {pusher_err}") # Return structured error payload for frontend return { "success": True, # HTTP 200 - API call succeeded "error": True, # Explicit error flag for frontend "error_type": "content_filter", "message": "Verification failed. Content filter triggered.", "overall_feedback_text": "Please check the document. The content was flagged by security filters.", "analysis_results": [], "overall_match": False, "confidence_score": "0%", "document_type": "Error", "mismatch_summary": ["Document verification blocked by content filter"], "is_hazy": False, "_llamaparse_success": llamaparse_success } logger.critical(f"Failed to analyze document: {str(e)}") # Return structured error payload for any other exception return { "success": True, # HTTP 200 - API call succeeded "error": True, # Explicit error flag for frontend "error_type": "system_error", "message": "Verification failed. Analysis error occurred.", "overall_feedback_text": f"Failed to analyze document: {str(e)}", "analysis_results": [], "overall_match": False, "confidence_score": "0%", "document_type": "Error", "mismatch_summary": [f"Analysis error: {str(e)}"], "is_hazy": False, "_llamaparse_success": llamaparse_success } def _create_realestate_document_prompt(self) -> str: return """ You are an expert KYC document analyzer. You will receive an image/PDF of ANY real estate supporting document, such as: - Property deed - Ownership certificate - Real estate transfer document - Title document - Property valuation statement Your job: 1. Extract any visible ownership information (owner name, share percentage, property value, etc.). 2. Match ONLY the provided fields. If a field was NOT provided, do not mark it as required. 3. If any field is not found in the document, set found_in_document to null and is_match to false. 4. Detect if the document image is hazy/blurred (is_hazy true/false). 5. Classify the document_type concisely (e.g., "Real Estate Deed", "Property Title Certificate", "Ownership Document", "Unknown"). 6. Provide overall_match = true only if all provided fields match. If a provided field is missing or mismatch, overall_match = false. 7. Provide confidence_score as a percentage. 8. Provide mismatch_summary listing only mismatched or missing fields. Return ONLY valid JSON as below: { "analysis_results": [ { "field_name": "first_name", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "confidence": "high|medium|low" }, { "field_name": "last_name", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "confidence": "high|medium|low" }, { "field_name": "name", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "confidence": "high|medium|low" }, { "field_name": "share", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "confidence": "high|medium|low" }, { "field_name": "value", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "confidence": "high|medium|low" } ], "overall_match": true/false, "confidence_score": "85%", "document_type": "Real Estate Deed", "mismatch_summary": ["value mismatch"], "is_hazy": false, "additional_notes": "optional" } Rules: - Only match fields that are provided in the input JSON. - Do not assume any field is mandatory. - Missing value is considered mismatch. - Partial similarity counts as mismatch. - Document classification must be truthful even if file_type differs. """ def _create_address_document_prompt(self) -> str: return """You are an expert KYC document analyzer. You will receive an image/PDF and a JSON input containing the exact fields that must be verified. Your responsibilities: 1. Extract and verify following fields provided in the JSON: - first_name - middle_name - last_name - to_date - from_date - full_address 2. For each provided field: - Extract the value from the document. 3. GENERIC ENHANCED EXTRACTION RULE (applies to ALL FIELDS): ------------------------------------------------------------ If the exact field value is NOT found: - Attempt to extract the closest semantically related text. 4. Detect document clarity: - is_hazy = true/false 5. STRICT JSON RESPONSE FORMAT (MANDATORY): { "analysis_results": [ { "field_name": "... or null", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "partial_match": true/false, "confidence": "high|medium|low" } ], "overall_match": true/false, "document_type": "...", "confidence_score": "0-100%", "mismatch_summary": ["..."], "is_hazy": false, "extracted_financials": { "extracted_claim_amount": "... or null", "extracted_collateral_value": "... or null" } } ENFORCEMENT RULES: - MANDATORY: Return analysis for EVERY field present in input JSON. Missing any field from analysis_results is strictly forbidden. - Return ONLY fields present in input JSON. - Never hallucinate missing values. - If document is unrelated → classify as "Unknown Document" and mark all fields mismatched. - Output STRICT JSON only. - Don't return null value in found_in_document, always return closest match text """ def _create_residence_document_prompt(self) -> str: """System prompt for residence document verification including haze + document type classification.""" return """You are an expert KYC document analyzer. Task: Given an image/PDF of a potential residence or identity document and provided structured personal & address data, you must: 1. Extract any visible personal name components and address components. 2. For each provided field, state if found, and whether it matches (exact, partial, no match, not found). 3. Detect if the document image is hazy/blurred/low-quality (return true/false as is_hazy based on readability of text, artifacts, lighting, focus). 4. Classify the document type with a concise label (e.g., 'US Driver License', 'US Utility Bill', 'Indian Aadhaar Card', 'Indian PAN Card', 'Residence Lease', 'Bank Statement', 'Unknown', ). 5. Provide an overall_match boolean: true only if all mandatory fields (first_name, last_name, city, state, zip_code, street_name, house_number) are matches (partial counts as mismatch). 6. Provide a confidence_score as a percentage string based on clarity + number of matches. 7. Populate mismatch_summary with short strings for each mismatch. JSON RESPONSE REQUIRED (no extra text) with structure: { "analysis_results": [ {"field_name": "first_name", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "confidence": "high|medium|low"}, ... ], "overall_match": true/false, "confidence_score": "85%", "document_type": "US Driver License", "mismatch_summary": ["city mismatch"], "is_hazy": false, "additional_notes": "optional notes" } Rules: - Partial spell differences or abbreviations count as mismatch (list them). - If a field isn't present, found_in_document = null and is_match = false. - is_hazy true only if significant portions of text are unreadable. - document_type must be one short label. """ def _create_generic_prompt(self, file_type: str) -> str: return ("You are a document analyzer. File type category: " f"{file_type}. Extract key data and classify document. Return a JSON object with keys: " "analysis_results, overall_match, confidence_score, document_type, mismatch_summary, is_hazy, additional_notes.") def _create_bank_statement_prompt(self) -> str: return """You are an expert financial document analyzer. You will receive an image/PDF and a JSON input containing the exact fields that must be verified. Your responsibilities: 1. Extract and verify following fields provided in the JSON: - first_name - middle_name - last_name - bank_name - account_type - balance 2. For each provided field: - Extract the value from the document. 3. GENERIC ENHANCED EXTRACTION RULE (applies to ALL FIELDS): ------------------------------------------------------------ If the exact field value is NOT found: - Attempt to extract the closest semantically related text. 4. Detect document clarity: - is_hazy = true/false 5. STRICT JSON RESPONSE FORMAT (MANDATORY): { "analysis_results": [ { "field_name": "... or null", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "partial_match": true/false, "confidence": "high|medium|low" } ], "overall_match": true/false, "document_type": "...", "confidence_score": "0-100%", "mismatch_summary": ["..."], "is_hazy": false, "extracted_financials": { "extracted_claim_amount": "... or null", "extracted_collateral_value": "... or null" } } ENFORCEMENT RULES: - Return ONLY fields present in input JSON. - Never hallucinate missing values. - For 'amount' field: confidence should be "low" if exact value/attribute name not found, "medium" if partial/full match. - If document is unrelated → classify as "Unknown Document" and mark all fields mismatched. - Output STRICT JSON only. - Don't return null value in found_in_document, always return closest match text """ ''' Goals: For each statement file determine months covered (YYYY-MM) and opening/closing balances per month. Extract transactions per month with fields: date (YYYY-MM-DD), description (short), debit (if any), credit (if any). Use numeric strings, null if not present. For each month compute total_debits (sum of debits), total_credits (sum of credits), and computed_closing_balance = opening_balance - total_debits + total_credits. Verify computed_closing_balance equals stated closing_balance (opening/closing arithmetic consistency). Across all files check last 6 consecutive months coverage ending with the most recent month present; list missing months if any. Determine statement_final_balance (closing balance of most recent month across all files) and compare with provided expected balance (input 'balance'). Match name components (first, middle if present, last) found on statements. Match bank_name and account_type. Detect manipulation by flagging arithmetic_inconsistencies if any month closing_balance differs from computed_closing_balance or if running balance progression across months does not align. JSON ONLY. Top-level key: bank_statement_results (array). Each element:\n{\n \"file_url\": \"...\",\n \"months_covered\": [\"2025-05\",\"2025-06\"],\n \"missing_months\": [\"2025-09\"],\n \"balances\": [\n {\"month\": \"2025-05\", \"opening_balance\": \"1234.56\", \"closing_balance\": \"1334.56\", \"total_debits\": \"100.00\", \"total_credits\": \"200.00\", \"computed_closing_balance\": \"1334.56\", \"arithmetic_consistent\": true}\n ],\n \"transactions\": [\n {\"date\": \"2025-05-02\", \"description\": \"ATM Withdrawal\", \"debit\": \"50.00\", \"credit\": null}\n ],\n \"provided_final_balance\": \"10000.00\",\n \"statement_final_balance\": \"10000.00\",\n \"final_balance_matches\": true,\n \"months_complete\": true,\n \"arithmetic_inconsistencies\": [],\n \"name_matches\": {\"first_name\": true, \"middle_name\": true, \"last_name\": true},\n \"bank_name_matches\": true,\n \"account_type_matches\": true,\n \"errors\": []\n}\nRules:\n- If data can't be extracted set value to null and add an error note.\n- Be conservative; do not hallucinate months or balances; if uncertain leave null and add error.\n- months_complete is true only if 6 consecutive months are present.\n- arithmetic_consistent true only if opening_balance - total_debits + total_credits == closing_balance for each month.\n- List any discrepancies in arithmetic_inconsistencies.\n""" ''' """ def _create_bank_statement_user_message(self, data: Dict, file_url: str) -> List[Dict]: urls = [u.strip() for u in file_url.split(',') if u.strip()] meta_lines = [ f"Expected First Name: {data.get('first_name','N/A')}", f"Expected Middle Name: {data.get('middle_name','')}", f"Expected Last Name: {data.get('last_name','N/A')}", f"Expected Bank Name: {data.get('bank_name','N/A')}", f"Expected Account Type: {data.get('account_type','N/A')}", f"Expected Final Balance: {data.get('balance','N/A')}", "Files Provided (index:url):" ] for idx, u in enumerate(urls): meta_lines.append(f"{idx}:{u}") meta_lines.append("Return JSON per spec with bank_statement_results only; one object per input file in order.") meta = "\n".join(meta_lines) # Return text block; images will be added by PDF handling logic in verify_document return [{"type": "text", "text": meta}] """ def _create_bank_statement_user_message(self, data: Dict, file_url: str, extracted_text: str = None) -> List[Dict]: """ Dynamically builds the user message for bank statement verification. Only includes fields provided by the client. """ lines = ["Bank Statement Verification Input:"] # Add LlamaParse extracted text if available if extracted_text: lines.append("") lines.append("--- LLAMAPARSE EXTRACTED TEXT ---") lines.append(extracted_text[:3000]) lines.append("--- END LLAMAPARSE TEXT ---") lines.append("") for key, value in data.items(): # Format key nicely for display pretty_key = key.replace("_", " ").title() lines.append(f"{pretty_key}: {value}") lines.append("Return JSON only per system instructions.") text = "\n".join(lines) return [ {"type": "text", "text": text}, {"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} ] def _create_residence_user_message(self, data: Dict, file_url: str, extracted_text: str = None) -> List[Dict]: lines = ["Residence Document Verification Input:"] # Add LlamaParse extracted text if available if extracted_text: lines.append("") lines.append("--- LLAMAPARSE EXTRACTED TEXT ---") lines.append(extracted_text[:3000]) lines.append("--- END LLAMAPARSE TEXT ---") lines.append("") lines.append(f"First Name: {data.get('first_name','N/A')}") lines.append(f"Middle Name: {data.get('middle_name','')}") lines.append(f"Last Name: {data.get('last_name','N/A')}") lines.append(f"City: {data.get('city','N/A')}") lines.append(f"State: {data.get('state','N/A')}") lines.append(f"County: {data.get('county','N/A')}") lines.append(f"ZIP Code: {data.get('zip_code','N/A')}") lines.append(f"Street Name: {data.get('street_name','N/A')}") lines.append(f"House Number: {data.get('house_number','N/A')}") lines.append("Return JSON only per system instructions.") text = "\n".join(lines) return [ {"type": "text", "text": text}, {"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} ] def _create_address_user_message(self, data: Dict, file_url: str, extracted_text: str = None) -> List[Dict]: """ Dynamically builds the user message for residence/identity verification. Only includes fields provided by the client. Prevents forcing unnecessary fields. """ lines = ["Document Verification Input:"] # Add LlamaParse extracted text if available if extracted_text: lines.append("") lines.append("--- LLAMAPARSE EXTRACTED TEXT ---") lines.append(extracted_text[:3000]) lines.append("--- END LLAMAPARSE TEXT ---") lines.append("") for key, value in data.items(): # Format key nicely for display pretty_key = key.replace("_", " ").title() lines.append(f"{pretty_key}: {value}") lines.append("Return JSON only per system instructions.") text = "\n".join(lines) return [ {"type": "text", "text": text}, {"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} ] def _create_vehicle_document_prompt(self) -> str: """System prompt for vehicle document verification.you are an expert vehicle document analyzer. You will receive an image/PDF of a vehicle-related document (e.g., registration card, title certificate, insurance card) along with a JSON input containing the exact fields that must be verified Focus on: - Extract and compare following fields (first_name, middle_name, last_name) if present. - if middle name is not provided, dont change provided_value to N/A, keep it as send by user. - Matching provided fields (year, maker, model, mileage, current_value, vehicle_number). - Extracting vehicle registration/plate number and confirming match. - Extracting model and confirming match. - Detecting document expiry date (e.g., registration card validity) and whether it's expired as of today. - Classifying the document type (e.g., 'Vehicle Registration Card', 'Insurance Card', 'Title Certificate'). - Haze/quality detection. JSON response must include analysis_results per provided fields and vehicle_details block. """ return vehicle_prompt def _create_vehicle_user_message(self, data: Dict, file_url: str, extracted_text: str = None) -> List[Dict]: lines = ["Vehicle Document Verification Input:"] # Add LlamaParse extracted text if available if extracted_text: lines.append("") lines.append("--- LLAMAPARSE EXTRACTED TEXT ---") lines.append(extracted_text[:3000]) lines.append("--- END LLAMAPARSE TEXT ---") lines.append("") lines.extend([ f"Owner First Name: {data.get('first_name','')}", f"Owner Middle Name: {data.get('middle_name','')}", f"Owner Last Name: {data.get('last_name','')}", f"Year: {data.get('year','')}", f"Maker: {data.get('maker','')}", f"Model: {data.get('model','')}", f"Mileage: {data.get('mileage','')}", f"Current Value: {data.get('current_value','')}", f"Vehicle Number: {data.get('number', data.get('vehicle_no',''))}", "Return JSON only per system instructions." ]) text = "\n".join(lines) return [ {"type": "text", "text": text}, {"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} ] def _create_business_document_prompt(self) -> str: return """You are an expert business document analyzer. You will receive an image/PDF and a JSON input containing the exact fields that must be verified. Your responsibilities: 1. Extract and verify following fields provided in the JSON: - first_name - middle_name - last_name - ein - business_type - location - business_name 2. For each provided field: - Extract the value from the document. 3. GENERIC ENHANCED EXTRACTION RULE (applies to ALL FIELDS): ------------------------------------------------------------ If the exact field value is NOT found: - Attempt to extract the closest semantically related text. 4. Detect document clarity: - is_hazy = true/false 5. STRICT JSON RESPONSE FORMAT (MANDATORY): { "analysis_results": [ { "field_name": "... or null", "provided_value": "...", "found_in_document": "... or null", "is_match": true/false, "partial_match": true/false, "confidence": "high|medium|low" } ], "overall_match": true/false, "document_type": "...", "confidence_score": "0-100%", "mismatch_summary": ["..."], "is_hazy": false, "extracted_financials": { "extracted_claim_amount": "... or null", "extracted_collateral_value": "... or null" } } ENFORCEMENT RULES: - MANDATORY: Return analysis for EVERY field present in input JSON. Missing any field from analysis_results is strictly forbidden. - Return ONLY fields present in input JSON. - Never hallucinate missing values. - If document is unrelated → classify as "Unknown Document" and mark all fields mismatched. - Output STRICT JSON only. - Don't return null value in found_in_document, always return closest match text """ def _create_business_user_message(self, data: Dict, file_url: str, extracted_text: str = None) -> List[Dict]: lines = ["Business Document Verification Input:"] # Add LlamaParse extracted text if available if extracted_text: lines.append("") lines.append("--- LLAMAPARSE EXTRACTED TEXT ---") lines.append(extracted_text[:]) lines.append("--- END LLAMAPARSE TEXT ---") lines.append("") lines.extend([ f"First Name: {data.get('first_name','')}", f"Middle Name: {data.get('middle_name','')}", f"Last Name: {data.get('last_name','')}", f"EIN: {data.get('ein','')}", f"Business Type: {data.get('business_type', data.get('type',''))}", f"Location: {data.get('location','')}", f"Business Name: {data.get('business_name','')}", "Return JSON only per system instructions." ]) text = "\n".join(lines) return [ {"type": "text", "text": text}, {"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} ] def _create_generic_user_message(self, data: Dict, file_url: str) -> List[Dict]: return [ {"type": "text", "text": f"Generic file_type input: {data}. Return structured JSON."}, {"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} ] async def _pdf_url_to_data_images(self, pdf_url: str, max_pages: int = 2) -> List[str]: """Download a PDF from URL and convert up to max_pages to PNG data URIs. Uses PyMuPDF; raises if conversion fails.""" async with aiohttp.ClientSession() as session: async with session.get(pdf_url) as resp: if not resp.ok: raise Exception(f"Failed to download PDF: {resp.status}") pdf_bytes = await resp.read() try: doc = fitz.open(stream=pdf_bytes, filetype="pdf") except Exception as e: raise Exception(f"Unable to open PDF: {e}") images: List[str] = [] for i, page in enumerate(doc): if i >= max_pages: break pix = page.get_pixmap(dpi=144) # higher dpi for readability png_bytes = pix.tobytes("png") # direct PNG bytes b64 = base64.b64encode(png_bytes).decode('utf-8') data_uri = f"data:image/png;base64,{b64}" images.append(data_uri) if not images: raise Exception("No pages converted from PDF.") return images def _create_response_message(self, data: Dict, file_url: str, extracted_text: str = None) -> str: """ Dynamically builds the user message for residence/identity verification. Only includes fields provided by the client. Prevents forcing unnecessary fields. Includes LlamaParse-extracted text if available for better accuracy. """ lines = ["Document Verification Input:"] # Add LlamaParse extracted text if available if extracted_text: lines.append("") lines.append("--- LLAMAPARSE EXTRACTED TEXT ---") lines.append(extracted_text) lines.append("--- END LLAMAPARSE TEXT ---") lines.append("") for key, value in data.items(): # Format key nicely for display pretty_key = key.replace("_", " ").title() lines.append(f"{pretty_key}: {value}") lines.append("Return JSON only per system instructions.") text = "\n".join(lines) print(text) return text # return [ # {"type": "text", "text": text} # ] #{"type": "image_url", "image_url": {"url": file_url, "detail": "high"}} # Global instance azure_openai_client = AzureOpenAIClient()