tools/backfill_core_from_wording.py

#!/usr/bin/env python3
"""Backfill ONLY the deterministically-safe core fields from the policy's
own already-curated source PDF (insurer's IRDAI-filed wording).

ZERO fabrication. Each field is filled only when an unambiguous,
IRDAI-standardised clause is present verbatim. The verbatim sentence is
stored as source_quote and the existing wrapped-object schema is preserved:
    {"value":…, "unit":…, "source_quote":…, "source_pdf_path":…,
     "extraction_method":"tools/backfill_core_from_wording.py", "_confidence":…}

Fields handled (and ONLY these — the rest need official-brochure web sourcing):
  - cashless_treatment_supported  (IRDAI std "Cashless facility means …")
  - ayush_coverage               (AYUSH treatment/hospital coverage clause)
  - maternity_coverage           (explicit cover clause vs Excl18 exclusion)
  - initial_waiting_period_days  (IRDAI Excl03 standardised 30 days)
  - pre_existing_disease_waiting_months (IRDAI Excl01 "expiry of N months")

Run:
  python tools/backfill_core_from_wording.py --dry     # report only
  python tools/backfill_core_from_wording.py --write    # apply edits
"""
from __future__ import annotations

import json
import os
import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from _pdf_text_cache import resolve_source_pdf, text_for  # noqa: E402

ROOT = Path(__file__).resolve().parents[1]
FACTS_DIR = ROOT / "40-data" / "policy_facts"


def _clip(t: str, m: re.Match, before=90, after=240) -> str:
    s = max(0, m.start() - before)
    e = min(len(t), m.end() + after)
    return re.sub(r"\s+", " ", t[s:e]).strip()


def _empty(v) -> bool:
    if isinstance(v, dict):
        v = v.get("value") if "value" in v else (v.get("covered") if "covered" in v else v)
    return v is None or v == "" or v == [] or (isinstance(v, dict) and not v)


# ---- per-field deterministic extractors -----------------------------------
def x_cashless(t: str):
    m = re.search(r"Cashless facility means[^.]{0,400}\.", t, re.I)
    if m:
        return True, _clip(t, m)
    return None, None


_AYUSH_RE = re.compile(
    r"(AYUSH (?:Hospital|Treatment|Day Care)[^.]{0,300}\.)"
    r"|(Ayurveda, Yoga (?:and|&) Naturopathy, Unani, Siddha (?:and|&) Hom"
    r"o?eopathy[^.]{0,260}\.)",
    re.I,
)


# Product types where an AYUSH *definition* in the boilerplate does NOT imply
# the product covers AYUSH in-patient treatment (CI/PA/benefit/top-up-only).
_AYUSH_SKIP_NAME = re.compile(
    r"criti|critical[ -]?illness|personal[ -]?accident|\bpa\b|cancer|"
    r"hospi[- ]?cash|benefit(?:\s|$)|surakshit|term|janata personal",
    re.I,
)
# A genuine AYUSH benefit clause — coverage verb attached to the AYUSH benefit,
# NOT merely the IRDAI standard definition ("AYUSH Treatment refers to ...").
_AYUSH_COVER_RE = re.compile(
    r"AYUSH (?:Treatment|Hospital|Day Care|Cover|Benefit|Hospitali[sz]ation)"
    r"[^.]{0,300}?(?:we will (?:cover|indemnif|pay|reimburse)|"
    r"is covered|shall be (?:covered|payable|indemnified)|covered up to|"
    r"will be (?:covered|payable)|expenses (?:incurred|towards)[^.]{0,90}AYUSH|"
    r"in[- ]?patient (?:care|treatment)[^.]{0,80}AYUSH|"
    r"sum insured[^.]{0,40}AYUSH|AYUSH[^.]{0,80}up to the sum insured)"
    r"[^.]{0,200}\.",
    re.I,
)
# Also accept a "Coverage/Benefits" section header that lists AYUSH as a covered
# item (e.g. "4. AYUSH Treatment Section B-1.4" inside a benefits index).
_AYUSH_BENEFIT_INDEX_RE = re.compile(
    r"(?:Cover|Benefit|In[- ]?Patient)[^\n]{0,40}\n?[^\n]{0,40}"
    r"AYUSH (?:Treatment|Hospitali[sz]ation)",
    re.I,
)


def x_ayush(t: str, policy_name: str = ""):
    if policy_name and _AYUSH_SKIP_NAME.search(policy_name):
        return None, None
    m = _AYUSH_COVER_RE.search(t)
    if m:
        return True, _clip(t, m)
    return None, None


def x_maternity(t: str):
    # Covered?  explicit benefit clause: "We will indemnify ... Maternity
    # Expenses" / "Maternity Expenses ... is opted/covered". This is a
    # BENEFIT clause, never the IRDAI Excl18 definition.
    mcov = re.search(
        r"(?:[Ww]e will (?:cover|indemnif\w+|pay)|[Cc]ompany will "
        r"(?:cover|indemnif\w+|pay)|shall (?:cover|indemnif\w+))"
        r"[^.]{0,60}Maternity Expenses[^.]{0,260}\.", t
    ) or re.search(
        r"Maternity Expenses[^.]{0,30}(?:is|are)\s+(?:opted|covered)"
        r"[^.]{0,220}\.", t, re.I
    )
    if mcov:
        # Guard: not the exclusion sentence itself.
        if not re.search(r"Excl18|shall be excluded", mcov.group(0), re.I):
            return True, _clip(t, mcov)
    # Excluded in base?  IRDAI standard exclusion code Excl18 against the
    # Maternity head. This is the same signal the existing curation used
    # (e.g. acko-health-ii: maternity_coverage=false w/ the Excl18 quote).
    mex = re.search(
        r"Maternity(?: Expenses)?\s*[^.\n]{0,30}\(?(?:Code[- ]?)?Excl18\)?"
        r"[^.]{0,260}\.", t, re.I
    )
    if mex:
        return False, _clip(t, mex)
    return None, None


def x_initial_wait(t: str):
    m = re.search(
        r"(?:30|thirty)[ -]days?\b[^.]{0,140}?(?:Code[- ]?Excl03|Excl03|"
        r"from the (?:date of |)first policy commencement|within 30 days from "
        r"the first policy)[^.]{0,200}\.", t, re.I,
    ) or re.search(
        r"Excl03[^.]{0,40}(?:30|thirty)[ -]days?[^.]{0,200}\.", t, re.I
    )
    if m:
        return 30, _clip(t, m)
    return None, None


def x_ped_months(t: str):
    pats = [
        r"[Pp]re-?[Ee]xisting [Dd]iseases?[^.]{0,300}?expiry of\s+(\d{1,2})\s*"
        r"(?:\([^)]{0,18}\)\s*)?months",
        r"expiry of\s+(\d{1,2})\s*(?:\([^)]{0,18}\)\s*)?months[^.]{0,160}?"
        r"pre-?existing",
        r"[Pp]re-?[Ee]xisting [Dd]iseases?[^.]{0,260}?(\d{1,2})\s+months of "
        r"continuous coverage",
    ]
    for p in pats:
        m = re.search(p, t, re.I | re.S)
        if m:
            v = int(m.group(1))
            if 12 <= v <= 60:  # sanity: IRDAI PED is 12–48mo in practice
                return v, _clip(t, m)
    return None, None


EXTRACTORS = {
    "cashless_treatment_supported": (x_cashless, None, "high"),
    "ayush_coverage": (x_ayush, None, "high"),
    "maternity_coverage": (x_maternity, None, "high"),
    "initial_waiting_period_days": (x_initial_wait, "days", "high"),
    "pre_existing_disease_waiting_months": (x_ped_months, "months", "high"),
}


def main() -> int:
    write = "--write" in sys.argv
    changed_files = 0
    changed_fields = 0
    per_field = {}
    samples = []

    for f in sorted(FACTS_DIR.glob("*.json")):
        if f.name.startswith("_"):
            continue
        data = json.loads(f.read_text())
        sp = resolve_source_pdf(data)
        if not sp:
            continue
        t = text_for(sp)
        if not t:
            continue
        pname = data.get("policy_name", "") or ""
        file_changed = False
        for field, (fn, unit, conf) in EXTRACTORS.items():
            cur = data.get(field)
            if cur is not None and not _empty(cur):
                continue
            if field == "ayush_coverage":
                val, quote = fn(t, pname)
            else:
                val, quote = fn(t)
            if val is None:
                continue
            entry = {
                "value": val,
                "source_pdf_path": sp,
                "source_quote": quote[:600],
                "extraction_method": "tools/backfill_core_from_wording.py "
                "(deterministic IRDAI-standard clause match over pdfplumber text)",
                "_confidence": conf,
            }
            if unit:
                entry["unit"] = unit
            data[field] = entry
            file_changed = True
            changed_fields += 1
            per_field[field] = per_field.get(field, 0) + 1
            if len(samples) < 12:
                samples.append((f.name, field, val, quote[:140]))
        if file_changed:
            changed_files += 1
            if write:
                f.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")

    print(("WROTE" if write else "DRY-RUN") +
          f": {changed_fields} fields across {changed_files} files")
    print("per field:", json.dumps(per_field, indent=1))
    print("\nsamples:")
    for fn, fld, v, q in samples:
        print(f"  [{fn}] {fld} = {v}")
        print(f"     “{q}”")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())