#!/usr/bin/env python3 """Locked process v1.1 §6: pre-delivery consistency gate (scripted). Usage: report_gate.py RUN_DIR [--strict] Checks (all must pass): 1. Count match — executive-summary count == "### Finding N" count in REPORT-final.md 2. Evidence present — every material finding's evidence exists in findings.json with a non-empty payload 3. Identity match — report business name == raw capture name; report date == capture date 4. Surface disclosure — every surface with audit.surfaces_checked == false renders as ✗ in the Surfaces Reviewed table; no surface with a false flag renders as ✓ 5. Material support — re-derivation: for each material phone/address/hours finding, the claim is re-computed from the RAW CAPTURE (not the engine's evidence) and must still hold. Catches report/engine drift, not judgment calls. Exit 0 = PASS, 1 = FAIL (failures printed). Appends a GATE RESULT block to VALIDATION.md (replaces the prior block if re-run on the same report). --strict additionally fails when needs-verification items exist (pre-ship). """ import datetime, glob, json, os, re, sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import report_generate as rg # reuse normalization + loaders — single source of truth def phone_core(s): d = re.sub(r"\D", "", s or "") if len(d) == 11 and d.startswith("1"): d = d[1:] if not d or set(d) == {"0"}: return "" return d def addr_core(s): if not s: return "" a = s.lower() a = re.sub(r"\b(suite|ste|st|unit|fl|floor)\b\.?\s*\w*", " ", a) a = re.sub(r"#\s*\w*", " ", a) a = re.sub(r"[^\w\s]", " ", a) return re.sub(r"\s+", " ", a).strip() def day_eq(g, a): tg, ta = rg.tmin(g), rg.tmin(a) if "PARSE1800" in (tg, ta): return "parse" return "eq" if tg == ta else "diff" def url_norm(u): if not u: return "" u = re.sub(r"[?#].*$", "", u) u = re.sub(r"^https?://", "", u) u = re.sub(r"^www\.", "", u) return u.rstrip("/").lower() def latest_capture(run_dir): return rg.latest_capture(run_dir) NUM_WORDS = {"One": 1, "Two": 2, "Three": 3, "Four": 4, "Five": 5, "Six": 6, "Seven": 7, "Eight": 8, "Nine": 9} # findings whose whole claim IS an absence — the engine legitimately emits no evidence for them; # gate re-derives them from raw capture instead (check 5) ABSENCE_TITLES = ( "No website found", "unreachable", "No business description", "No JSON-LD", "no LocalBusiness", "Price level", "Open Graph", "UTM parameters", ) def check(run_dir, strict=False): fails = [] raw = json.load(open(latest_capture(run_dir))) fj = json.load(open(os.path.join(run_dir, "findings.json"))) report = open(os.path.join(run_dir, "REPORT-final.md")).read() name = raw.get("name") or fj.get("business") # 1. count match n_section = len(re.findall(r"^### Finding \d+:", report, re.M)) m = re.search(r"\b(\w+) material integrity issue", report) if m: w = m.group(1) n_stated = NUM_WORDS.get(w, w) n_stated = int(n_stated) if str(n_stated).isdigit() else 0 else: n_stated = 0 if not (n_section == n_stated): fails.append(f"count mismatch: summary says {n_stated}, findings section has {n_section}") # 2. evidence present — presence findings must carry engine evidence; absence findings # (no-website / no-description / no-schema / price / OG) are verified from raw in check 5 mat, fp, nv, enh = rg.classify(fj, raw) ev_by_title = {f["title"]: f.get("evidence") for f in fj["findings"]} for sev, t, _why in mat: if any(t.startswith(a) or a in t for a in ABSENCE_TITLES): continue ev = ev_by_title.get(t) if ev in (None, "", {}, []): fails.append(f"material finding with empty evidence: {t}") if strict and nv: fails.append(f"strict mode: {len(nv)} needs-verification item(s) open") # 3. identity match if not report.startswith(f"# {name}\n"): fails.append(f"identity: report title != capture name ({name!r})") d = rg.run_date(raw) if d != "date unknown" and d not in report: fails.append(f"identity: capture date {d!r} not in report") # 4. surface disclosure checked = raw.get("audit", {}).get("surfaces_checked", {}) table = report.split("## Surfaces Reviewed") if len(table) < 2: fails.append("surfaces: no Surfaces Reviewed section") else: tbl = table[1].split("##")[0] for key, label in rg.SURFACES: rows = [l for l in tbl.splitlines() if l.startswith(f"| {label} |")] if not rows: if checked.get(key): fails.append(f"surfaces: row missing for captured surface {label}") continue ok = rows[0].split("|")[2].strip().startswith("✓") if checked.get(key) is False and ok: fails.append(f"surfaces: {label} failed capture but shows ✓") if checked.get(key) is True and not ok: fails.append(f"surfaces: {label} captured but shows ✗") # 5. material support — re-derive material claims from RAW CAPTURE src = raw.get("sources", {}) g = src.get("google_business_profile") or {} a = src.get("apple_maps") or {} w = src.get("website") if isinstance(src.get("website"), dict) else {} for sev, t, why in mat: if t.startswith("Phone"): cores = {phone_core(x) for x in (g.get("phone"), a.get("phone"), w.get("phone_on_page")) if phone_core(x)} if len(cores) <= 1: fails.append(f"support: phone claim not re-derived from raw (cores={cores})") elif t.startswith("Address") and "unit designators" not in why: if addr_core(g.get("address")) == addr_core(a.get("address")): fails.append(f"support: address claim not re-derived from raw (cores equal)") elif t.startswith("Hours"): days = [] for day in ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"): gh = (g.get("hours") or {}).get(day) ah = (a.get("hours") or {}).get(day) if gh and ah and day_eq(gh, ah) == "diff": days.append(day) if not days: fails.append(f"support: hours claim not re-derived from raw (no differing day)") elif "UTM" in t: gu, au = g.get("website"), a.get("website") if not (( "?" in (gu or "")) != ("?" in (au or ""))): fails.append(f"support: UTM claim not re-derived from raw") elif t.startswith("No website") or "unreachable" in t: if not (g.get("website") or "").strip(): fails.append("support: unreachable-website claim not re-derived (GBP has no website)") elif t.startswith("No business description"): if any((s.get("description") or "").strip() for s in (g, a) if isinstance(s, dict)): fails.append("support: no-description claim not re-derived (a surface has a description)") return fails, (n_stated, len(mat), len(nv), name, d) def append_gate(run_dir, ok, fails, stats): p = os.path.join(run_dir, "VALIDATION.md") v = open(p).read() if os.path.exists(p) else "" v = re.sub(r"\n## GATE RESULT.*", "", v, flags=re.S) n_stated, n_mat, n_nv, name, d = stats block = "\n## GATE RESULT\n\n" block += f"**{('PASS' if ok else 'FAIL')}** — report_gate.py, {datetime.datetime.now().isoformat(timespec='seconds')}\n\n" block += (f"- Count: {n_stated} stated / {n_mat} material after review — " + ("match" if n_stated == n_mat else f"MISMATCH ({n_stated} vs {n_mat})") + "\n" f"- Identity: report title + date match raw capture ({name!r}, {d}) — " + ("match" if not any(f.startswith("identity") for f in fails) else "FAIL") + "\n" f"- Surface disclosure: all captured/failed surfaces rendered correctly — " + ("match" if not any(f.startswith("surfaces") for f in fails) else "FAIL") + "\n" f"- Material support: every material finding re-derived from raw capture — " + ("hold" if not any(f.startswith("support") for f in fails) else "FAIL") + "\n" f"- Needs-verification open: {n_nv}\n") if fails: block += "\n**Failures:**\n" + "".join(f"- {f}\n" for f in fails) open(p, "w").write(v + block) def main(): args = [a for a in sys.argv[1:]] strict = "--strict" in args dirs = [a for a in args if a != "--strict"] if not dirs: sys.exit("usage: report_gate.py RUN_DIR [--strict]") rc = 0 for run_dir in dirs: ok, fails = True, [] try: fails, stats = check(run_dir, strict=strict) except Exception as e: fails = [f"gate crashed: {e}"] stats = (0, 0, 0, "?", "?") ok = not fails if not ok: rc = 1 append_gate(run_dir, ok, fails, stats) print(f"{os.path.basename(run_dir):42} {'PASS' if ok else 'FAIL'}") for f in fails: print(f" - {f}") sys.exit(rc) if __name__ == "__main__": main()