Blame for validate.py

"""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)