219 lines
9.7 KiB
Python
219 lines
9.7 KiB
Python
#!/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"):
|
|
# re-derive against the SAME shared baseline the engine used (canonical if present)
|
|
from canonical_baseline import fetch_canonical_record, resolve_reference, normalize_hours_dict, parse_hours_value
|
|
try:
|
|
rec = fetch_canonical_record(name)
|
|
except Exception:
|
|
rec = None
|
|
_label, ref_hours, _prov = resolve_reference({"name": name, "sources": src}, rec)
|
|
ref_norm = normalize_hours_dict(ref_hours or {})
|
|
dev = []
|
|
for day in ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"):
|
|
rv = ref_norm.get(day)
|
|
if rv is None:
|
|
continue
|
|
if any(parse_hours_value((s.get("hours") or {}).get(day)) not in (None, rv)
|
|
for s in src.values() if isinstance(s, dict) and s.get("hours")):
|
|
dev.append(day)
|
|
if not dev:
|
|
fails.append(f"support: hours claim not re-derived from raw (no surface deviates from reference)")
|
|
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()
|