import hashlib import logging from datetime import datetime, timezone from decimal import Decimal from pathlib import Path from uuid import uuid4 import pdfplumber from sqlalchemy.orm import Session from app.core.config import get_settings from app.core.exceptions import NotFoundError, ValidationError from app.models.enums import ExtractionStatus, PaymentStatus, PolicyCoverageType, PolicyStatus, extraction_status_is, extraction_status_value from app.models.extraction_result import ExtractionResult from app.models.insurance_company import InsuranceCompany from app.models.policy import Policy from app.models.policy_document import PolicyDocument from app.models.policy_document import PolicyDocument from app.parsers.base import ExtractedField from app.parsers.registry import PARSER_REGISTRY, detect_company from app.parsers.text_normalize import normalize_pdf_text from app.repositories.customer_repository import CustomerRepository from app.repositories.insurance_company_repository import InsuranceCompanyRepository from app.repositories.policy_repository import PolicyRepository from app.repositories.vehicle_repository import VehicleRepository from app.schemas.upload import ConfirmExtractionRequest from app.services.commission_service import CommissionService from app.services.extraction_correction_service import ( ExtractionCorrectionService, SOURCE_UPLOAD_REVIEW, submitted_values_from_payload, suggested_values_from_defaults, ) from app.services.policy_service import PolicyService from app.utils.date_parser import parse_flexible_date from app.utils.helpers import update_payment_summary from app.utils.policy_status import compute_policy_status from app.utils.body_types import normalize_body_type from app.utils.file_storage import get_file_storage, read_storage_bytes from app.utils.pdf_text import pack_raw_text_fields from app.utils.vehicle_types import normalize_vehicle_type from app.utils.virus_scan import scan_file_bytes logger = logging.getLogger(__name__) class UploadService: def __init__(self, db: Session): self.db = db self.settings = get_settings() self.customers = CustomerRepository(db) self.vehicles = VehicleRepository(db) self.policies = PolicyRepository(db) self.companies = InsuranceCompanyRepository(db) def _ensure_upload_dir(self, agency_id: int | None = None) -> Path: path = Path(self.settings.upload_dir) if agency_id: path = path / str(agency_id) path.mkdir(parents=True, exist_ok=True) return path @staticmethod def _checksum(file_bytes: bytes) -> str: return hashlib.sha256(file_bytes).hexdigest() def _find_pending_by_checksum(self, agency_id: int, checksum: str) -> ExtractionResult | None: return ( self.db.query(ExtractionResult) .filter( ExtractionResult.agency_id == agency_id, ExtractionResult.file_checksum == checksum, ExtractionResult.policy_id.is_(None), ExtractionResult.status != ExtractionStatus.DISMISSED, ) .order_by(ExtractionResult.id.desc()) .first() ) def _find_dismissed_by_checksum(self, agency_id: int, checksum: str) -> ExtractionResult | None: return ( self.db.query(ExtractionResult) .filter( ExtractionResult.agency_id == agency_id, ExtractionResult.file_checksum == checksum, ExtractionResult.status == ExtractionStatus.DISMISSED, ) .order_by(ExtractionResult.id.desc()) .first() ) def _find_saved_policy_id_by_checksum(self, agency_id: int, checksum: str) -> int | None: row = ( self.db.query(PolicyDocument.policy_id) .join(Policy, Policy.id == PolicyDocument.policy_id) .filter(Policy.agency_id == agency_id, PolicyDocument.checksum == checksum) .order_by(PolicyDocument.id.desc()) .first() ) return row[0] if row else None def list_known_checksums(self, agency_id: int) -> dict[str, list[str]]: pending_rows = ( self.db.query(ExtractionResult.file_checksum) .filter( ExtractionResult.agency_id == agency_id, ExtractionResult.policy_id.is_(None), ExtractionResult.file_checksum.isnot(None), ExtractionResult.status != ExtractionStatus.DISMISSED, ) .distinct() .all() ) dismissed_rows = ( self.db.query(ExtractionResult.file_checksum) .filter( ExtractionResult.agency_id == agency_id, ExtractionResult.file_checksum.isnot(None), ExtractionResult.status == ExtractionStatus.DISMISSED, ) .distinct() .all() ) saved_rows = ( self.db.query(PolicyDocument.checksum) .join(Policy, Policy.id == PolicyDocument.policy_id) .filter(Policy.agency_id == agency_id) .distinct() .all() ) return { "pending_checksums": sorted({row[0] for row in pending_rows if row[0]}), "dismissed_checksums": sorted({row[0] for row in dismissed_rows if row[0]}), "saved_checksums": sorted({row[0] for row in saved_rows if row[0]}), } def _build_upload_response( self, extraction: ExtractionResult, *, duplicate: bool = False, skipped: bool = False, message: str | None = None, ) -> dict: raw = extraction.raw_data or {} company = raw.get("company_detection", {}) fields = raw.get("fields", []) return { "extraction_id": extraction.id, "status": extraction_status_value(extraction.status), "company": { "code": company.get("code"), "name": company.get("name"), "confidence": company.get("confidence"), "insurance_company_id": extraction.insurance_company_id, }, "fields": fields, "storage_path": extraction.storage_path, "checksum": extraction.file_checksum or raw.get("checksum"), "text_source": raw.get("text_source", "native"), "ocr_engine": raw.get("ocr_engine"), "is_scanned": raw.get("is_scanned", False), "duplicate": duplicate, "skipped": skipped, "message": message, "policy_id": extraction.policy_id, } @staticmethod def _build_skipped_saved_response(checksum: str, policy_id: int, filename: str) -> dict: return { "extraction_id": None, "status": "saved", "company": None, "fields": [], "storage_path": None, "checksum": checksum, "duplicate": True, "skipped": True, "message": f"'{filename}' was already saved as a policy.", "policy_id": policy_id, } @staticmethod def _build_skipped_dismissed_response(checksum: str, filename: str) -> dict: return { "extraction_id": None, "status": "dismissed", "company": None, "fields": [], "storage_path": None, "checksum": checksum, "duplicate": True, "skipped": True, "message": f"'{filename}' was removed from the review queue and will not be imported again.", "policy_id": None, } def extract_pdf( self, file_bytes: bytes, filename: str, user_id: int, agency_id: int, *, lab_mode: bool = False, ) -> dict: if not filename.lower().endswith(".pdf"): raise ValidationError("Only PDF files are allowed") if len(file_bytes) > self.settings.max_upload_size_bytes: raise ValidationError(f"File exceeds {self.settings.max_upload_size_mb}MB limit") scan_file_bytes(file_bytes, self.settings) checksum = self._checksum(file_bytes) if self._find_dismissed_by_checksum(agency_id, checksum): return self._build_skipped_dismissed_response(checksum, filename) existing_pending = self._find_pending_by_checksum(agency_id, checksum) if existing_pending: return self._build_upload_response( existing_pending, duplicate=True, skipped=True, message="This file is already in the review queue.", ) saved_policy_id = self._find_saved_policy_id_by_checksum(agency_id, checksum) if saved_policy_id: return self._build_skipped_saved_response(checksum, saved_policy_id, filename) self._ensure_upload_dir(agency_id) storage_name = f"{uuid4().hex}_{filename}" relative_key = f"{agency_id}/{storage_name}" storage = get_file_storage(self.settings) storage_path = storage.save(relative_key, file_bytes) text_pack = pack_raw_text_fields(file_bytes) text = text_pack["raw_text"] company_code, company_name, company_confidence = detect_company(text, filename) parser = PARSER_REGISTRY.get(company_code or "generic", PARSER_REGISTRY["generic"]) result = parser(text) from app.services.parser_training_service import ParserTrainingService try: result.fields = ParserTrainingService.apply_learned_patterns( self.db, agency_id, company_code or "", text, result.fields, raw_text_pages=text_pack["raw_text_pages"], ) except Exception: logger.exception( "Learned training merge failed: %s", filename, extra={ "agency_id": agency_id, "upload_filename": filename, "company_code": company_code, }, ) field_map = result.get_field_map() has_policy_or_vehicle = field_map.get("policy_number") or field_map.get("vehicle_registration_number") required_present = all(field_map.get(r) and field_map[r].value for r in ["customer_name", "mobile_number"]) has_end_date = bool(field_map.get("policy_end_date") and field_map["policy_end_date"].value) if required_present and has_end_date and has_policy_or_vehicle: status = ExtractionStatus.SUCCESS elif has_end_date and has_policy_or_vehicle: status = ExtractionStatus.PARTIAL else: status = ExtractionStatus.FAILED company_record = self.companies.get_by_code(company_code) if company_code else None if not company_record and company_name: company_record = self.companies.get_by_code("unknown") extraction = ExtractionResult( agency_id=agency_id, uploaded_by=user_id, original_filename=filename, file_checksum=checksum, status=status, insurance_company_id=company_record.id if company_record else None, raw_data={ **result.to_dict(), "company_detection": { "code": company_code, "name": company_name, "confidence": company_confidence, }, "checksum": checksum, "file_size": len(file_bytes), "raw_text": text_pack["raw_text"], "raw_text_pages": text_pack["raw_text_pages"], "page_count": text_pack["page_count"], "text_source": text_pack.get("text_source", "native"), "ocr_engine": text_pack.get("ocr_engine"), "is_scanned": text_pack.get("is_scanned", False), "lab_mode": lab_mode, }, storage_path=str(storage_path), ) self.db.add(extraction) self.db.commit() self.db.refresh(extraction) return self._build_upload_response(extraction) @staticmethod def _summary_from_raw_data(raw_data: dict | None) -> dict[str, str | None]: raw = raw_data or {} field_map = { f["field_name"]: f.get("value") for f in raw.get("fields", []) if isinstance(f, dict) and f.get("field_name") } company = raw.get("company_detection") or {} return { "customer_name": field_map.get("customer_name"), "policy_number": field_map.get("policy_number"), "vehicle_registration_number": field_map.get("vehicle_registration_number"), "policy_end_date": field_map.get("policy_end_date"), "insurance_company_name": company.get("name"), } @staticmethod def _policy_match_dict(policy: Policy, match_type: str) -> dict: return { "match_type": match_type, "existing_policy_id": policy.id, "existing_policy_number": policy.policy_number, "customer_name": policy.customer.name if policy.customer else None, "vehicle_registration_number": policy.vehicle.registration_number if policy.vehicle else None, "policy_end_date": policy.policy_end_date.isoformat() if policy.policy_end_date else None, } def resolve_policy_match( self, agency_id: int, summary: dict[str, str | None], *, by_number: dict[str, Policy] | None = None, by_vehicle: dict[str, Policy] | None = None, ) -> dict | None: policy_number = (summary.get("policy_number") or "").strip() if policy_number: policy = None if by_number is not None: policy = by_number.get(policy_number.lower()) else: policy = self.policies.find_by_policy_number(agency_id, policy_number) if policy: return self._policy_match_dict(policy, "policy_number") vehicle_reg = (summary.get("vehicle_registration_number") or "").strip() if vehicle_reg: policy = None if by_vehicle is not None: policy = by_vehicle.get(vehicle_reg.lower()) else: policy = self.policies.find_by_vehicle_registration(agency_id, vehicle_reg) if policy: return self._policy_match_dict(policy, "vehicle_registration") return None @staticmethod def _dedupe_extractions(rows: list[ExtractionResult]) -> list[ExtractionResult]: seen_checksums: set[str] = set() deduped_rows: list[ExtractionResult] = [] for row in rows: checksum = row.file_checksum or (row.raw_data or {}).get("checksum") if checksum: if checksum in seen_checksums: continue seen_checksums.add(checksum) deduped_rows.append(row) return deduped_rows def _build_extraction_list_items( self, agency_id: int, rows: list[ExtractionResult], *, queue_match: str | None = None, ) -> tuple[list[dict], int, int]: deduped_rows = self._dedupe_extractions(rows) summaries = [self._summary_from_raw_data(row.raw_data) for row in deduped_rows] by_number, by_vehicle = self.policies.find_active_matches( agency_id, policy_numbers=[s.get("policy_number") or "" for s in summaries], vehicle_registrations=[s.get("vehicle_registration_number") or "" for s in summaries], ) company_ids = {row.insurance_company_id for row in deduped_rows if row.insurance_company_id} company_names: dict[int, str] = {} if company_ids: for company in self.db.query(InsuranceCompany).filter(InsuranceCompany.id.in_(company_ids)).all(): company_names[company.id] = company.name all_items: list[dict] = [] new_count = 0 existing_count = 0 for row, summary in zip(deduped_rows, summaries): if extraction_status_is(row.status, ExtractionStatus.DISMISSED): continue policy_match = self.resolve_policy_match(agency_id, summary, by_number=by_number, by_vehicle=by_vehicle) is_new = policy_match is None if is_new: new_count += 1 else: existing_count += 1 company_name = company_names.get(row.insurance_company_id) if row.insurance_company_id else None if not company_name: company_name = summary.get("insurance_company_name") all_items.append( { "id": row.id, "status": extraction_status_value(row.status), "original_filename": row.original_filename, "insurance_company_id": row.insurance_company_id, "insurance_company_name": company_name, "customer_name": summary.get("customer_name"), "policy_number": summary.get("policy_number"), "vehicle_registration_number": summary.get("vehicle_registration_number"), "policy_end_date": summary.get("policy_end_date"), "created_at": row.created_at, "is_new": is_new, "policy_match": policy_match, } ) if queue_match == "new": all_items = [item for item in all_items if item["is_new"]] elif queue_match == "existing": all_items = [item for item in all_items if not item["is_new"]] return all_items, new_count, existing_count def list_extractions( self, agency_id: int, *, pending_only: bool = True, page: int = 1, page_size: int = 50, search: str | None = None, status: str | None = None, queue_match: str | None = None, ) -> tuple[list[dict], int, int, int]: query = self.db.query(ExtractionResult).filter(ExtractionResult.agency_id == agency_id) if pending_only: query = query.filter( ExtractionResult.policy_id.is_(None), ExtractionResult.status != ExtractionStatus.DISMISSED, ) if search: query = query.filter(ExtractionResult.original_filename.ilike(f"%{search.strip()}%")) if status: query = query.filter(ExtractionResult.status == ExtractionStatus(status)) if pending_only: rows = query.order_by(ExtractionResult.created_at.desc()).limit(500).all() all_items, new_count, existing_count = self._build_extraction_list_items( agency_id, rows, queue_match=queue_match ) total = len(all_items) start = (page - 1) * page_size items = all_items[start : start + page_size] return items, total, new_count, existing_count total = query.count() rows = ( query.order_by(ExtractionResult.created_at.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) items, _, _ = self._build_extraction_list_items(agency_id, rows) return items, total, 0, 0 def get_extraction(self, extraction_id: int, agency_id: int) -> ExtractionResult: extraction = self.db.query(ExtractionResult).filter(ExtractionResult.id == extraction_id).first() if not extraction: raise NotFoundError("Extraction result not found") if extraction.agency_id is not None and extraction.agency_id != agency_id: raise NotFoundError("Extraction result not found") if extraction_status_is(extraction.status, ExtractionStatus.DISMISSED): raise NotFoundError("This file was removed from the review queue") return extraction def dismiss_extraction(self, agency_id: int, user_id: int, extraction_id: int) -> dict: extraction = self.db.query(ExtractionResult).filter(ExtractionResult.id == extraction_id).first() if not extraction: raise NotFoundError("Extraction result not found") if extraction.agency_id is not None and extraction.agency_id != agency_id: raise NotFoundError("Extraction result not found") if extraction.policy_id: raise ValidationError("This extraction was already saved as a policy") if extraction_status_is(extraction.status, ExtractionStatus.DISMISSED): return { "id": extraction.id, "status": extraction_status_value(extraction.status), "message": "Already removed from the review queue", } extraction.status = ExtractionStatus.DISMISSED self.db.flush() self.db.commit() self.db.refresh(extraction) return { "id": extraction.id, "status": extraction_status_value(extraction.status), "original_filename": extraction.original_filename, "checksum": extraction.file_checksum, "message": "Removed from the review queue. This file will not appear again.", } def get_extraction_pdf_path(self, extraction_id: int, agency_id: int) -> tuple[str, str]: extraction = self.get_extraction(extraction_id, agency_id) if not extraction.storage_path: raise NotFoundError("PDF file not found for this extraction") storage = get_file_storage(self.settings) if not storage.exists(extraction.storage_path): raise NotFoundError("PDF file missing on disk") return extraction.storage_path, extraction.original_filename def confirm_extraction( self, agency_id: int, user_id: int, extraction_id: int, payload: ConfirmExtractionRequest, ): extraction = self.get_extraction(extraction_id, agency_id) if extraction.policy_id: raise ValidationError("This extraction has already been saved as a policy") form_defaults = self.fields_to_form_defaults(extraction, agency_id) suggested = suggested_values_from_defaults(form_defaults, extraction) submitted = submitted_values_from_payload(payload) ExtractionCorrectionService(self.db).record_correction( extraction=extraction, agency_id=agency_id, user_id=user_id, suggested=suggested, submitted=submitted, source=SOURCE_UPLOAD_REVIEW, notes="Auto-captured on upload review confirm", ) policy_end = parse_flexible_date(payload.policy_end_date) if not policy_end: raise ValidationError("Policy end date is required", details=[{"field": "policy_end_date", "error": "required"}]) if not payload.policy_number or not str(payload.policy_number).strip(): raise ValidationError( "Policy number is required", details=[{"field": "policy_number", "error": "required"}], ) PolicyService(self.db).ensure_unique_policy_number(agency_id, payload.policy_number) if payload.customer_id: customer = self.customers.get_by_id(payload.customer_id, agency_id) if not customer: raise NotFoundError("Customer not found") customer = self.customers.update( customer, { "name": payload.customer_name, "mobile": payload.mobile_number, "email": payload.email_address, "address": payload.address, "city": payload.city, "state": payload.state, "pincode": payload.pincode, "nominee_name": payload.nominee_name, }, ) else: existing = self.customers.get_by_mobile(agency_id, payload.mobile_number) if existing: raise ValidationError( "Customer with this mobile already exists. Link to existing customer or use dedup check.", details=[{"field": "mobile_number", "error": "duplicate"}], ) customer = self.customers.create( agency_id, { "name": payload.customer_name, "mobile": payload.mobile_number, "email": payload.email_address, "address": payload.address, "city": payload.city, "state": payload.state, "pincode": payload.pincode, "nominee_name": payload.nominee_name, }, ) vehicle = self.vehicles.create( customer.id, { "registration_number": payload.vehicle_registration_number, "engine_number": payload.engine_number, "chassis_number": payload.chassis_number, "make": payload.vehicle_make, "model": payload.vehicle_model, "fuel_type": payload.fuel_type, "vehicle_type": normalize_vehicle_type(payload.vehicle_type) or payload.vehicle_type, "body_type": normalize_body_type(payload.body_type) or payload.body_type, "manufacturing_year": payload.manufacturing_year, }, ) insurance_company_id = payload.insurance_company_id or extraction.insurance_company_id coverage_type = None if payload.coverage_type: normalized = payload.coverage_type.lower().replace(" ", "_") if "third" in normalized: coverage_type = PolicyCoverageType.THIRD_PARTY elif "comprehensive" in normalized: coverage_type = PolicyCoverageType.COMPREHENSIVE status = compute_policy_status(policy_end, PolicyStatus.DRAFT) total_commission = CommissionService(self.db).auto_commission( agency_id, payload.premium_amount, insurance_company_id, normalize_vehicle_type(payload.vehicle_type) or payload.vehicle_type, coverage_type.value if coverage_type else payload.coverage_type, payload.total_commission, ) pending, _, payment_status = update_payment_summary(total_commission, Decimal("0.00")) policy = self.policies.create( agency_id, { "customer_id": customer.id, "vehicle_id": vehicle.id, "insurance_company_id": insurance_company_id, "policy_number": payload.policy_number, "policy_type": payload.policy_type, "coverage_type": coverage_type, "policy_start_date": parse_flexible_date(payload.policy_start_date), "policy_end_date": policy_end, "premium_amount": payload.premium_amount, "idv": payload.idv, "ncb": payload.ncb, "status": status, "total_commission": total_commission, "total_paid": Decimal("0.00"), "pending_amount": pending, "payment_status": payment_status, "notes": payload.notes, "created_by": user_id, }, ) raw = extraction.raw_data or {} file_size = raw.get("file_size") or len(read_storage_bytes(extraction.storage_path, self.settings)) checksum = raw.get("checksum") or hashlib.sha256(read_storage_bytes(extraction.storage_path, self.settings)).hexdigest() document = PolicyDocument( policy_id=policy.id, storage_path=extraction.storage_path, original_filename=extraction.original_filename, file_size=file_size, checksum=checksum, uploaded_by=user_id, uploaded_at=datetime.now(timezone.utc), ) self.db.add(document) extraction.policy_id = policy.id if insurance_company_id: extraction.insurance_company_id = insurance_company_id self.db.commit() return PolicyService(self.db).get_policy(agency_id, policy.id) @staticmethod def _extract_text(file_bytes: bytes) -> str: from app.utils.pdf_text import extract_text_pages, pages_to_raw_text return pages_to_raw_text(extract_text_pages(file_bytes)) def fields_to_form_defaults(self, extraction: ExtractionResult, agency_id: int) -> dict[str, str | None]: raw = extraction.raw_data or {} company = raw.get("company_detection", {}) company_code = raw.get("company_code") or company.get("code") raw_text = raw.get("raw_text") or "" raw_text_pages = raw.get("raw_text_pages") existing_fields = [ ExtractedField( field_name=f["field_name"], value=f.get("value"), confidence=f.get("confidence") or 0.0, source_page=f.get("source_page"), source=f.get("source") or "extracted", ) for f in raw.get("fields", []) ] if raw_text: from app.services.parser_training_service import ParserTrainingService try: existing_fields = ParserTrainingService.apply_learned_patterns( self.db, agency_id, company_code or "", raw_text, existing_fields, raw_text_pages=raw_text_pages, ) except Exception: logger.exception( "Learned training merge failed while building form defaults", extra={ "agency_id": agency_id, "extraction_id": extraction.id, "company_code": company_code, }, ) field_map = {f.field_name: f.value for f in existing_fields} return { "customer_name": field_map.get("customer_name"), "mobile_number": field_map.get("mobile_number"), "email_address": field_map.get("email_address"), "address": field_map.get("address"), "city": field_map.get("city"), "state": field_map.get("state"), "pincode": field_map.get("pincode"), "nominee_name": field_map.get("nominee_name"), "vehicle_registration_number": field_map.get("vehicle_registration_number"), "engine_number": field_map.get("engine_number"), "chassis_number": field_map.get("chassis_number"), "vehicle_make": field_map.get("vehicle_make") or field_map.get("make"), "vehicle_model": field_map.get("vehicle_model") or field_map.get("model"), "vehicle_type": field_map.get("vehicle_type"), "body_type": field_map.get("body_type"), "fuel_type": field_map.get("fuel_type"), "manufacturing_year": field_map.get("manufacturing_year"), "policy_number": field_map.get("policy_number"), "policy_type": field_map.get("policy_type"), "coverage_type": field_map.get("coverage_type") or field_map.get("comprehensive_third_party"), "policy_start_date": field_map.get("policy_start_date"), "policy_end_date": field_map.get("policy_end_date"), "premium_amount": field_map.get("premium_amount"), "idv": field_map.get("idv"), "ncb": field_map.get("ncb"), "insurance_company_name": company.get("name"), }