Spaces:
Running
Running
File size: 2,726 Bytes
b87bd2d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | #!/usr/bin/env python3
"""Extract + cache full text of every source PDF referenced by policy_facts.
Writes plaintext to tools/.pdf_text_cache/<sha-ish>.txt keyed by pdf path.
Idempotent. This is internal-data mining (the insurer's own IRDAI-filed
wording), NOT web fabrication.
"""
from __future__ import annotations
import hashlib
import json
import os
import sys
from pathlib import Path
import pdfplumber
ROOT = Path(__file__).resolve().parents[1]
CACHE = ROOT / "tools" / ".pdf_text_cache"
CACHE.mkdir(exist_ok=True)
# Known path fixes (typo'd source_pdf_path in policy_facts).
PATH_FIX = {
"rag/corpus/bajaj-allianz/group-health-guard-gold__wordings.pdf":
"rag/corpus/bajaj-allianz/group-health-guard-silver__wordings.pdf",
}
def cache_path(pdf_rel: str) -> Path:
h = hashlib.sha1(pdf_rel.encode()).hexdigest()[:16]
return CACHE / f"{h}.txt"
def text_for(pdf_rel: str) -> str | None:
pdf_rel = PATH_FIX.get(pdf_rel, pdf_rel)
p = ROOT / pdf_rel
if not p.exists():
return None
cp = cache_path(pdf_rel)
if cp.exists() and cp.stat().st_size > 0:
return cp.read_text(errors="replace")
parts = []
try:
with pdfplumber.open(p) as pdf:
for pg in pdf.pages:
parts.append(pg.extract_text() or "")
for tbl in (pg.extract_tables() or []):
for row in tbl:
parts.append(
" | ".join(c or "" for c in row)
)
except Exception as e: # noqa: BLE001
sys.stderr.write(f"!! {pdf_rel}: {e}\n")
return None
txt = "\n".join(parts)
cp.write_text(txt, errors="replace")
return txt
def resolve_source_pdf(facts: dict) -> str | None:
sp = facts.get("source_pdf_path")
if isinstance(sp, str):
return sp
meta = facts.get("_meta", {}) or {}
if isinstance(meta.get("primary_source_pdf"), str):
return meta["primary_source_pdf"]
for v in facts.values():
if isinstance(v, dict) and isinstance(v.get("source_pdf_path"), str):
return v["source_pdf_path"]
return None
if __name__ == "__main__":
import glob
n_ok = n_fail = 0
for f in sorted(glob.glob(str(ROOT / "40-data/policy_facts/*.json"))):
if os.path.basename(f).startswith("_"):
continue
facts = json.loads(Path(f).read_text())
sp = resolve_source_pdf(facts)
if not sp:
n_fail += 1
continue
t = text_for(sp)
if t:
n_ok += 1
else:
n_fail += 1
print("FAIL", os.path.basename(f), sp)
print(f"cached text for {n_ok} policies, {n_fail} failed")
|