| """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), |
| "ta-IN": (0x0B80, 0x0BFF), |
| "te-IN": (0x0C00, 0x0C7F), |
| "kn-IN": (0x0C80, 0x0CFF), |
| "ml-IN": (0x0D00, 0x0D7F), |
| } |
| INDIC_BLOCKS = (0x0900, 0x0D80) |
|
|
|
|
| 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): |
| 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 |
| |
| 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) |
|
|
|
|
| |
| 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()), |
| } |
|
|
|
|
| |
| 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 |
| max_field_disagreement: float = 0.0 |
| min_script_purity: float = 0.90 |
| max_uncertain: int = 0 |
| max_attempts: int = 2 |
| 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) |
|
|
