""" idp.py — an Intelligent Document Processing pipeline (Knowledge Module 04). raw doc --classify--> route --extract--> fields+confidence+regions --confidence gate--> {straight-through | human review} --validate (IBAN mod-97, balance math, expiry)--> persist + audit The EXTRACTOR is the only cloud-dependent part; here it's a deterministic stand-in (`FakeExtractor`) that returns the SAME result shape as Azure AI Document Intelligence (a list of fields, each with a value, a confidence, and a bounding region). Swap in `AzureDocumentExtractor` to go live — nothing else changes. The confidence gate + validation + audit are the bank-grade parts and run identically offline. """ from __future__ import annotations from dataclasses import dataclass, field from validation import (date_not_expired, iban_is_valid, normalize_digits, statement_balance_consistent) # --- result shape (mirrors Azure DI output) --------------------------------- @dataclass class Field: name: str value: str confidence: float # 0..1 bounding_region: tuple = (0, 0, 0, 0) # (page, x, y, w) — for highlight/audit @dataclass class ExtractionResult: doc_type: str fields: list[Field] def get(self, name: str) -> Field | None: return next((f for f in self.fields if f.name == name), None) # --- synthetic documents (stand-ins for scanned PDFs/images) ---------------- # Each "raw doc" carries the fields a real extractor would read, with the # confidence the model would assign. Two are clean; two have a deliberately # low-confidence or inconsistent field to exercise the review/validation paths. SAMPLE_DOCS: dict[str, dict] = { "statement_clean": { "type": "bank_statement", "fields": { "AccountIBAN": ("AE070331234567890123456", 0.99), "OpeningBalance": ("1000.00", 0.98), "ClosingBalance": ("1914.50", 0.98), "Transactions": ("9000.00,-85.50,-8000.00", 0.97), # +9000 -85.50 -8000 => 914.50 net }, }, "statement_bad_row": { "type": "bank_statement", "fields": { "AccountIBAN": ("AE070331234567890123456", 0.99), "OpeningBalance": ("1000.00", 0.98), "ClosingBalance": ("1914.50", 0.98), "Transactions": ("9000.00,-85.50,-9000.00", 0.55), # OCR misread -8000 as -9000 }, }, "emirates_id": { "type": "id_document", "fields": { "FullName": ("Amal Al Mansoori", 0.97), "IDNumber": ("784-1990-1234567-1", 0.96), "Expiry": ("2028-03-01", 0.92), }, }, "id_lowconf": { "type": "id_document", "fields": { "FullName": ("Omar ???", 0.41), # blurry scan "IDNumber": ("784-1985-7654321-2", 0.93), "Expiry": ("2024-01-01", 0.90), # expired! }, }, } # --- classifier ------------------------------------------------------------- def classify(raw: dict) -> str: """Route to the right extractor (K04 §6). Here the synthetic doc states its type; in production a custom classifier or an LLM picks among passport / ID / statement / invoice so you call the matching prebuilt/custom model.""" return raw["type"] # --- extractor (the swappable, cloud-dependent part) ------------------------ class FakeExtractor: """Deterministic stand-in for Azure AI Document Intelligence.""" def extract(self, raw: dict) -> ExtractionResult: fields = [Field(name=n, value=normalize_digits(v), confidence=c, bounding_region=(1, 0, i * 20, 100)) for i, (n, (v, c)) in enumerate(raw["fields"].items())] return ExtractionResult(doc_type=raw["type"], fields=fields) class AzureDocumentExtractor: """PRODUCTION extractor. Drop-in for FakeExtractor: from azure.ai.documentintelligence import DocumentIntelligenceClient from azure.identity import DefaultAzureCredential client = DocumentIntelligenceClient(endpoint, DefaultAzureCredential()) poller = client.begin_analyze_document("prebuilt-layout", body=doc_bytes) result = poller.result() # map result.documents[0].fields -> [Field(name, value, confidence, # bounding_region=field.bounding_regions[0])] Use 'prebuilt-idDocument' for IDs, 'prebuilt-invoice' for invoices, or a composed custom model behind the classifier.""" # --- the pipeline ----------------------------------------------------------- # Per-field confidence thresholds, weighted by RISK. Money/identity fields are # strict; a name can be a touch lower. Below threshold => human review. THRESHOLDS = { "AccountIBAN": 0.95, "ClosingBalance": 0.95, "OpeningBalance": 0.95, "Transactions": 0.90, "IDNumber": 0.95, "Expiry": 0.90, "FullName": 0.85, } DEFAULT_THRESHOLD = 0.90 @dataclass class ProcessedDoc: doc_type: str extracted: dict straight_through: bool review_fields: list[str] = field(default_factory=list) validation: dict = field(default_factory=dict) audit: list = field(default_factory=list) class IDPPipeline: def __init__(self, extractor=None) -> None: self.extractor = extractor or FakeExtractor() self.review_queue: list[ProcessedDoc] = [] self.store: list[ProcessedDoc] = [] def process(self, raw: dict) -> ProcessedDoc: audit = [] doc_type = classify(raw) audit.append({"step": "classify", "type": doc_type}) result = self.extractor.extract(raw) audit.append({"step": "extract", "n_fields": len(result.fields)}) # 1) CONFIDENCE GATE — low-confidence high-risk fields go to human review. review = [f.name for f in result.fields if f.confidence < THRESHOLDS.get(f.name, DEFAULT_THRESHOLD)] audit.append({"step": "confidence_gate", "review_fields": review}) extracted = {f.name: f.value for f in result.fields} # 2) VALIDATION — hard rules (K04 §5,§7). validation = self._validate(doc_type, result) audit.append({"step": "validate", "result": validation}) # any failed validation also forces review for k, ok in validation.items(): if ok is False and k not in review: review.append(k) straight_through = len(review) == 0 doc = ProcessedDoc(doc_type, extracted, straight_through, review, validation, audit) (self.store if straight_through else self.review_queue).append(doc) audit.append({"step": "route", "straight_through": straight_through}) return doc def _validate(self, doc_type: str, result: ExtractionResult) -> dict: out: dict[str, bool] = {} if doc_type == "bank_statement": iban = result.get("AccountIBAN") if iban: out["AccountIBAN"] = iban_is_valid(iban.value) o, c, t = result.get("OpeningBalance"), result.get("ClosingBalance"), result.get("Transactions") if o and c and t: txns = [float(x) for x in t.value.split(",") if x] out["Transactions"] = statement_balance_consistent( float(o.value), txns, float(c.value)) elif doc_type == "id_document": idn = result.get("IDNumber") exp = result.get("Expiry") if idn: out["IDNumber"] = bool(idn.value.replace("-", "").isdigit()) if exp: out["Expiry"] = date_not_expired(exp.value) return out def stp_rate(self) -> float: total = len(self.store) + len(self.review_queue) return len(self.store) / total if total else 0.0