"""Deterministic checks and agreement-based routing. VLM engines give no calibrated per-word confidence, so routing relies on: 1. agreement between two engines (or two preprocessing candidates), 2. script consistency (output must be in the expected Unicode block), 3. [illegible] markers or low word confidence from the engine, 4. field format validators. Thresholds here are starting values; calibrate them on human-verified data. """ from __future__ import annotations import re import unicodedata from dataclasses import dataclass, field from enum import Enum SCRIPT_RANGES = { "hi-IN": (0x0900, 0x097F), # Devanagari "ta-IN": (0x0B80, 0x0BFF), # Tamil "te-IN": (0x0C00, 0x0C7F), # Telugu "kn-IN": (0x0C80, 0x0CFF), # Kannada "ml-IN": (0x0D00, 0x0D7F), # Malayalam } INDIC_BLOCKS = (0x0900, 0x0D80) # Devanagari .. Malayalam: nine 128-code-point blocks, same layout def nfc(s: str | None) -> str: return unicodedata.normalize("NFC", s or "") def to_script(text: str, language: str | None) -> str: """Move letters the model wrote in a look-alike Indic script into the expected one (e.g. Telugu ప్రారం -> Kannada ಪ್ರಾರಂ). The Indic blocks share one layout, so the letter keeps its offset; characters with no counterpart in the target script are left as they are.""" if not text or language not in SCRIPT_RANGES: return text lo = SCRIPT_RANGES[language][0] out = [] for c in text: cp = ord(c) if INDIC_BLOCKS[0] <= cp < INDIC_BLOCKS[1] and not lo <= cp <= lo + 0x7F: t = chr(lo + (cp & 0x7F)) if unicodedata.name(t, None): # assigned in the target block c = t out.append(c) return nfc("".join(out)) def levenshtein(a: str, b: str) -> int: if len(a) < len(b): a, b = b, a prev = list(range(len(b) + 1)) for i, ca in enumerate(a, 1): cur = [i] for j, cb in enumerate(b, 1): cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb))) prev = cur return prev[-1] def cer(hyp: str, ref: str) -> float: """Character error rate of hyp against ref (on NFC code points, whitespace collapsed).""" hyp, ref = (re.sub(r"\s+", " ", nfc(x)).strip() for x in (hyp, ref)) if not ref: return 0.0 if not hyp else 1.0 return levenshtein(hyp, ref) / len(ref) def wer(hyp: str, ref: str) -> float: h, r = nfc(hyp).split(), nfc(ref).split() if not r: return 0.0 if not h else 1.0 # word-level Levenshtein via token mapping vocab = {w: chr(0xE000 + i) for i, w in enumerate(set(h) | set(r))} return levenshtein("".join(vocab[w] for w in h), "".join(vocab[w] for w in r)) / len(r) def pairwise_disagreement(a: str, b: str) -> float: """Symmetric CER-like distance between two hypotheses, 0 = identical.""" a, b = (re.sub(r"\s+", " ", nfc(x)).strip() for x in (a, b)) if not a and not b: return 0.0 return levenshtein(a, b) / max(len(a), len(b)) def script_purity(text: str, language: str) -> float: """Share of letter/mark characters that fall in the expected script block. Digits, punctuation, whitespace and Latin are ignored (forms often mix English).""" lo, hi = SCRIPT_RANGES.get(language, (0, 0x10FFFF)) relevant = [c for c in nfc(text) if unicodedata.category(c)[0] in ("L", "M") and not ("A" <= c <= "z")] if not relevant: return 1.0 return sum(lo <= ord(c) <= hi for c in relevant) / len(relevant) # --------------------------------------------------------------- field validators def _digits(s: str) -> str: """Map any Unicode decimal digit (Tamil, Telugu, ...) to ASCII.""" return "".join(str(unicodedata.digit(c)) if unicodedata.category(c) == "Nd" else c for c in s) VALIDATORS = { "pincode": lambda v: bool(re.fullmatch(r"[1-9]\d{5}", _digits(v).replace(" ", ""))), "mobile": lambda v: bool(re.fullmatch(r"(\+?91)?[6-9]\d{9}", re.sub(r"[\s-]", "", _digits(v)))), "date": lambda v: bool(re.fullmatch(r"\d{1,2}[/.-]\d{1,2}[/.-]\d{2,4}", _digits(v).strip())), "amount": lambda v: bool(re.fullmatch(r"₹?\s?\d{1,3}(,\d{2,3})*(\.\d{1,2})?|₹?\s?\d+(\.\d{1,2})?", _digits(v).strip())), "text": lambda v: bool(v.strip()), } # --------------------------------------------------------------- routing class Decision(str, Enum): ACCEPT = "ACCEPT" RETRY_ALT_CANDIDATE = "RETRY_ALT_CANDIDATE" HUMAN_REVIEW = "HUMAN_REVIEW" RECAPTURE = "RECAPTURE" @dataclass class RoutingConfig: max_disagreement: float = 0.05 # full text, between engines max_field_disagreement: float = 0.0 # critical fields must match exactly after NFC min_script_purity: float = 0.90 max_uncertain: int = 0 # [illegible] marks / low-confidence words max_attempts: int = 2 # candidates tried before escalating recapture_flags: frozenset[str] = frozenset({"BLURRY", "LOW_RESOLUTION"}) @dataclass class RoutingOutcome: decision: Decision reasons: list[str] = field(default_factory=list) metrics: dict = field(default_factory=dict) def route(texts: dict[str, str], language: str, *, quality_flags: list[str], attempt: int, uncertain_counts: dict[str, int] | None = None, fields: dict[str, dict[str, str | None]] | None = None, field_types: dict[str, str] | None = None, critical_fields: set[str] | None = None, cfg: RoutingConfig | None = None) -> RoutingOutcome: """texts: engine name -> full text. fields: engine name -> {field: value}.""" cfg = cfg or RoutingConfig() reasons: list[str] = [] metrics: dict = {} usable = {k: v for k, v in texts.items() if v.strip()} if not usable: if cfg.recapture_flags & set(quality_flags): return RoutingOutcome(Decision.RECAPTURE, ["no text and poor image quality"], metrics) reasons.append("no engine returned text") names = sorted(usable) if len(names) >= 2: d = pairwise_disagreement(usable[names[0]], usable[names[1]]) metrics["text_disagreement"] = round(d, 4) if d > cfg.max_disagreement: reasons.append(f"engines disagree on full text ({d:.1%})") elif usable: reasons.append("single engine only - no agreement signal") for n, t in usable.items(): p = script_purity(t, language) metrics[f"script_purity_{n}"] = round(p, 3) if p < cfg.min_script_purity: reasons.append(f"{n}: unexpected script ({p:.0%} in {language})") for n, c in (uncertain_counts or {}).items(): if c > cfg.max_uncertain: reasons.append(f"{n}: {c} illegible/low-confidence tokens") critical_fields = critical_fields or set() field_types = field_types or {} for fname in sorted({f for fv in (fields or {}).values() for f in fv}): vals = {e: fv.get(fname) for e, fv in (fields or {}).items()} present = {e: v for e, v in vals.items() if v} is_crit = fname in critical_fields if not present: if is_crit: reasons.append(f"critical field '{fname}' missing") continue validator = VALIDATORS.get(field_types.get(fname, "text"), VALIDATORS["text"]) for e, v in present.items(): if not validator(v): reasons.append(f"{e}: field '{fname}' fails {field_types.get(fname, 'text')} format") if is_crit and len(present) >= 2: a, b = list(present.values())[:2] if pairwise_disagreement(a, b) > cfg.max_field_disagreement: reasons.append(f"critical field '{fname}' differs between engines") elif is_crit and len(present) < len(vals): reasons.append(f"critical field '{fname}' read by only one engine") if not reasons: return RoutingOutcome(Decision.ACCEPT, [], metrics) if cfg.recapture_flags & set(quality_flags) and attempt >= cfg.max_attempts: return RoutingOutcome(Decision.RECAPTURE, reasons, metrics) if attempt < cfg.max_attempts: return RoutingOutcome(Decision.RETRY_ALT_CANDIDATE, reasons, metrics) return RoutingOutcome(Decision.HUMAN_REVIEW, reasons, metrics)