feat(auditing): pipeline v1.1 — report generation + scripted gate + temporal delta
- report_generate.py: REPORT-final.md + VALIDATION.md from findings + raw capture; per-surface status from summary.surfaces_ok; Data Limitations - report_gate.py: §6 gate scripted (count/evidence/identity/material support); non-zero exit blocks delivery - audit_diff.py: before/after capture comparison -> DELTA.md - audit_pipeline.sh: 2 steps -> 4; collision-safe capture copy (re-runs preserve baseline); relative output dir resolved to absolute - beta-audit-process.md: locked v1.0 -> v1.1 (+ dated decision record) - all 9 runs of the 2026-08-15 batch regenerated + gated (9/9 PASS) - live end-to-end proof: Gilmore re-run, DELTA.md 0 changes (same day)
This commit is contained in:
@@ -0,0 +1,100 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Locked process v1.1: temporal delta — compare two raw captures, emit DELTA.md.
|
||||
|
||||
Usage:
|
||||
audit_diff.py BEFORE_RUN_DIR AFTER_RUN_DIR
|
||||
|
||||
Compares the LATEST capture in each run dir across the fields the audit
|
||||
scores (name, phone, address, hours per day, rating, reviews, website,
|
||||
category, description, schema presence) and writes DELTA.md into the AFTER
|
||||
dir. This is the proof-of-fix artifact: after the client applies the
|
||||
recommendations, re-run the pipeline and the delta is the invoice evidence.
|
||||
|
||||
Exit 0 always (a diff is a result, not an error); DELTA.md lists every
|
||||
change and confirms unchanged fields.
|
||||
"""
|
||||
import json, os, sys, glob
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import report_generate as rg # latest_capture + normalization
|
||||
|
||||
DAYS = ["monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday"]
|
||||
|
||||
def load(d):
|
||||
if os.path.isfile(d):
|
||||
f = d
|
||||
else:
|
||||
f = rg.latest_capture(d)
|
||||
raw = json.load(open(f))
|
||||
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 {}
|
||||
return raw, g, a, w
|
||||
|
||||
def row(label, before, after, fmt=str):
|
||||
b, n = fmt(before), fmt(after)
|
||||
if b == n:
|
||||
return None, (b, b)
|
||||
return (label, b, n), (b, n)
|
||||
|
||||
def schema_present(w):
|
||||
return bool(w.get("schema")) if isinstance(w, dict) else False
|
||||
|
||||
def compare(before_dir, after_dir):
|
||||
raw_b, g_b, a_b, w_b = load(before_dir)
|
||||
raw_a, g_a, a_a, w_a = load(after_dir)
|
||||
changes, same = [], []
|
||||
|
||||
def add(label, b, n, fmt=str):
|
||||
r, vals = row(label, b, n, fmt)
|
||||
(changes if r else same).append(r or (label, vals[0]))
|
||||
|
||||
add("Name (GBP)", g_b.get("name"), g_a.get("name"))
|
||||
add("Phone (GBP)", rg.phone_core(g_b.get("phone")), rg.phone_core(g_a.get("phone")))
|
||||
add("Phone (Apple)", rg.phone_core(a_b.get("phone")), rg.phone_core(a_a.get("phone")))
|
||||
add("Address (GBP)", rg.addr_core(g_b.get("address")), rg.addr_core(g_a.get("address")))
|
||||
add("Address (Apple)", rg.addr_core(a_b.get("address")), rg.addr_core(a_a.get("address")))
|
||||
add("Website (GBP)", g_b.get("website"), g_a.get("website"))
|
||||
add("Rating (GBP)", g_b.get("rating"), g_a.get("rating"))
|
||||
add("Reviews (GBP)", g_b.get("reviews"), g_a.get("reviews"))
|
||||
add("Category (GBP)", g_b.get("category"), g_a.get("category"))
|
||||
add("Description set (GBP)", bool(g_b.get("description")), bool(g_a.get("description")))
|
||||
add("Schema (website)", schema_present(w_b), schema_present(w_a))
|
||||
|
||||
for day in DAYS:
|
||||
gh_b, gh_a = (g_b.get("hours") or {}).get(day), (g_a.get("hours") or {}).get(day)
|
||||
if gh_b != gh_a:
|
||||
changes.append((f"Hours {day} (GBP)", gh_b or "—", gh_a or "—"))
|
||||
ah_b, ah_a = (a_b.get("hours") or {}).get(day), (a_a.get("hours") or {}).get(day)
|
||||
if ah_b != ah_a:
|
||||
changes.append((f"Hours {day} (Apple)", ah_b or "—", ah_a or "—"))
|
||||
|
||||
return changes, same, raw_b, raw_a
|
||||
|
||||
def main():
|
||||
if len(sys.argv) != 3:
|
||||
sys.exit("usage: audit_diff.py BEFORE_RUN_DIR AFTER_RUN_DIR")
|
||||
before_dir, after_dir = sys.argv[1], sys.argv[2]
|
||||
changes, same, raw_b, raw_a = compare(before_dir, after_dir)
|
||||
L = [f"# Delta: {raw_a.get('name', '?')}", "",
|
||||
f"**Before:** {rg.run_date(raw_b)} (`{os.path.basename(before_dir)}`) ",
|
||||
f"**After:** {rg.run_date(raw_a)} (`{os.path.basename(after_dir)}`) ",
|
||||
"", f"**{len(changes)} field(s) changed** across the audited surfaces.", ""]
|
||||
if changes:
|
||||
L += ["## Changed", "", "| Field | Before | After |", "|-------|--------|-------|"]
|
||||
for label, b, n in changes:
|
||||
L.append(f"| {label} | {b} | {n} |")
|
||||
L.append("")
|
||||
else:
|
||||
L += ["No changes detected between the two captures.", ""]
|
||||
L += ["## Unchanged", ""]
|
||||
L += [f"- {label}: {v}" for label, v in same]
|
||||
L += ["", "*Generated by audit_diff.py (locked process v1.1). Compare only — no judgment.*", ""]
|
||||
out = os.path.join(after_dir, "DELTA.md") if os.path.isdir(after_dir) else \
|
||||
os.path.join(os.path.dirname(os.path.abspath(after_dir)), "DELTA.md")
|
||||
open(out, "w").write("\n".join(L))
|
||||
print(f"{os.path.basename(after_dir)}: {len(changes)} change(s) -> {out}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -9,6 +9,11 @@ SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
BUSINESS="${1:?Usage: ./audit_pipeline.sh \"Business Name\" \"City, ST\" [output-dir]}"
|
||||
LOCATION="${2:?Usage: ./audit_pipeline.sh \"Business Name\" \"City, ST\" [output-dir]}"
|
||||
OUTPUT_DIR="${3:-}"
|
||||
# cwd-proof: resolve to absolute once, so relative output dirs can't break capture copy
|
||||
if [ -n "$OUTPUT_DIR" ]; then
|
||||
mkdir -p "$OUTPUT_DIR"
|
||||
OUTPUT_DIR="$(cd "$OUTPUT_DIR" && pwd)"
|
||||
fi
|
||||
|
||||
# Temporary working dir for this run
|
||||
WORK_DIR=$(mktemp -d /tmp/veripath.XXXXXX)
|
||||
@@ -30,12 +35,30 @@ echo "[OK] Extraction complete: $SCRAPER_OUT"
|
||||
# Step 2: Audit engine
|
||||
if [ -n "$OUTPUT_DIR" ]; then
|
||||
mkdir -p "$OUTPUT_DIR"
|
||||
# Required artifact set (locked beta process v1.0): raw capture + findings in the run folder
|
||||
cp "$SCRAPER_OUT" "$OUTPUT_DIR/"
|
||||
# Required artifact set (locked beta process v1.1): raw capture + findings in the run folder
|
||||
# Collision-safe: a re-run never overwrites the prior capture (temporal baseline, v1.1 §5)
|
||||
DEST="$OUTPUT_DIR/$(basename "$SCRAPER_OUT")"
|
||||
if [ -e "$DEST" ]; then
|
||||
BASE="$(basename "${DEST%.*}")"; EXT="${DEST##*.}"
|
||||
DEST="$OUTPUT_DIR/${BASE}__$(date +%Y%m%d-%H%M%S).$EXT"
|
||||
echo "[WARN] prior capture present — writing $DEST (baseline preserved)"
|
||||
fi
|
||||
cp "$SCRAPER_OUT" "$DEST"
|
||||
ENGINE_ARGS+=("--output-dir" "$OUTPUT_DIR")
|
||||
echo "[$(date +%T)] Auditing (output: $OUTPUT_DIR)"
|
||||
uv run python3 "$SCRIPT_DIR/audit_engine.py" "$SCRAPER_OUT" "${ENGINE_ARGS[@]}"
|
||||
echo "[DONE] Findings + raw capture written to $OUTPUT_DIR/"
|
||||
|
||||
# Step 3: Report + validation (locked process v1.1 §3)
|
||||
echo "[$(date +%T)] Generating report + validation"
|
||||
uv run python3 "$SCRIPT_DIR/report_generate.py" "$OUTPUT_DIR"
|
||||
|
||||
# Step 4: Pre-delivery consistency gate (v1.1 §6) — FAIL blocks delivery
|
||||
echo "[$(date +%T)] Consistency gate"
|
||||
if ! uv run python3 "$SCRIPT_DIR/report_gate.py" "$OUTPUT_DIR"; then
|
||||
echo "[GATE FAIL] $OUTPUT_DIR — do NOT deliver; see GATE RESULT in VALIDATION.md" >&2
|
||||
exit 1
|
||||
fi
|
||||
echo "[DONE] Findings + raw capture + report + gate written to $OUTPUT_DIR/"
|
||||
else
|
||||
echo "[$(date +%T)] Auditing (stdout)"
|
||||
uv run python3 "$SCRIPT_DIR/audit_engine.py" "$SCRAPER_OUT" "${ENGINE_ARGS[@]}"
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
#!/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()
|
||||
@@ -0,0 +1,425 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Locked process v1.1 step 3: report + validation builder (promoted from the tmp one-off).
|
||||
|
||||
Usage:
|
||||
report_generate.py RUN_DIR # one run folder
|
||||
report_generate.py --all VALIDATION_DIR
|
||||
|
||||
Reads engine findings + latest raw capture per run, normalizes format
|
||||
false-positives, classifies material vs FP vs needs-verification, emits
|
||||
REPORT-final.md (client) + VALIDATION.md (process).
|
||||
|
||||
v1.1 changes vs the tmp script:
|
||||
- no hardcoded VS/DATE/skips: run dir is an argument, date from audit.timestamp
|
||||
- surfaces table driven by audit.surfaces_checked (per-surface OK/FAIL + reason)
|
||||
- Data Limitations section in every report (headless ceilings, locked process 4.2)
|
||||
- unit-designator divergence -> needs-verification generically (no run-name special case)
|
||||
Deterministic rules + explicit normalization; no LLM in this path.
|
||||
"""
|
||||
import argparse, datetime, glob, json, os, re
|
||||
|
||||
SURFACES = [
|
||||
("google_business_profile", "Google Business Profile"),
|
||||
("apple_maps", "Apple Maps"),
|
||||
("bing_places", "Bing Places"),
|
||||
("website", "Website"),
|
||||
]
|
||||
FAIL_REASON = {
|
||||
"google_business_profile": "capture failed — excluded from this audit",
|
||||
"apple_maps": "capture failed — excluded from this audit",
|
||||
"bing_places": "capture blocked (bot-wall) — excluded, not verified",
|
||||
"website": "not found / unreachable at audit time",
|
||||
}
|
||||
|
||||
# ---------- normalization ----------
|
||||
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"}: # "0000000000" scrape artifact
|
||||
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)
|
||||
a = re.sub(r"\s+", " ", a).strip()
|
||||
return a
|
||||
|
||||
def suite_raw(s):
|
||||
"""extract the unit designator as written (for needs-verification note)"""
|
||||
m = re.search(r"\b(suite|ste|st|unit|fl|floor)\b\.?\s*([\w-]+)?", s or "", re.I)
|
||||
if not m:
|
||||
return None
|
||||
return (m.group(1).upper().replace("ST", "STE") + " " + (m.group(2) or "")).strip() or None
|
||||
|
||||
def tmin(t):
|
||||
"""'8 AM' / '8:00 AM' / 'Open 24 hours' / 'Closed' / '4:1800 PM' -> minutes or tag"""
|
||||
t = (t or "").strip()
|
||||
if re.search(r"1800", t):
|
||||
return "PARSE1800"
|
||||
if "24 hours" in t.lower():
|
||||
return "24H"
|
||||
if "closed" in t.lower():
|
||||
return "CLOSED"
|
||||
m = re.match(r"(\d{1,2})(?::(\d{2}))?\s*([ap])m", t, re.I)
|
||||
if not m:
|
||||
return "OTHER:" + t
|
||||
h, mi, ap = int(m.group(1)), int(m.group(2) or 0), m.group(3).lower()
|
||||
if h == 12: h = 0
|
||||
if ap == "p": h += 12
|
||||
return h * 60 + mi
|
||||
|
||||
def day_eq(g, a):
|
||||
"""compare one day's google vs apple hours -> 'eq' | 'diff' | 'parse'"""
|
||||
tg, ta = tmin(g), 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) # strip UTM/query
|
||||
u = re.sub(r"^https?://", "", u)
|
||||
u = re.sub(r"^www\.", "", u)
|
||||
return u.rstrip("/").lower()
|
||||
|
||||
# ---------- run loading ----------
|
||||
def latest_capture(run_dir):
|
||||
fs = glob.glob(os.path.join(run_dir, "*_multi_surface_*.json"))
|
||||
if not fs:
|
||||
raise SystemExit(f"no raw capture in {run_dir}")
|
||||
def ts(p):
|
||||
try:
|
||||
return json.load(open(p)).get("audit", {}).get("timestamp", "")
|
||||
except Exception:
|
||||
return ""
|
||||
return max(fs, key=lambda p: (ts(p), os.path.getmtime(p)))
|
||||
|
||||
def load_run(run_dir):
|
||||
raw = json.load(open(latest_capture(run_dir)))
|
||||
fj = json.load(open(os.path.join(run_dir, "findings.json")))
|
||||
return raw, fj
|
||||
|
||||
def run_date(raw):
|
||||
try:
|
||||
return datetime.datetime.fromisoformat(raw["audit"]["timestamp"]).strftime("%d %B %Y")
|
||||
except Exception:
|
||||
return "date unknown"
|
||||
|
||||
# ---------- per-run classification ----------
|
||||
def classify(fj, raw):
|
||||
src = raw.get("sources", {})
|
||||
g = src.get("google_business_profile") or {}
|
||||
a = src.get("apple_maps") or {}
|
||||
w = src.get("website")
|
||||
mat, fp, nv, enh = [], [], [], []
|
||||
|
||||
for f in fj["findings"]:
|
||||
t, ev, sev = f["title"], f.get("evidence"), f["severity"]
|
||||
|
||||
if t.startswith("Phone number inconsistent"):
|
||||
cores = {phone_core(x) for x in (ev.get("google"), ev.get("apple"), ev.get("website")) if phone_core(x)}
|
||||
if len(cores) <= 1:
|
||||
fp.append((sev, t, "identical after digit/+1 normalization"))
|
||||
else:
|
||||
mat.append((sev, t, f"distinct numbers: {', '.join(sorted(cores))}"))
|
||||
elif t.startswith("Address inconsistent"):
|
||||
cg, ca = addr_core(ev.get("google")), addr_core(ev.get("apple"))
|
||||
if not cg or not ca or cg == ca:
|
||||
if cg and ca and cg == ca:
|
||||
sg, sa = suite_raw(ev.get("google")), suite_raw(ev.get("apple"))
|
||||
if sg and sa and sg != sa:
|
||||
# same street core but different unit designators — genuinely ambiguous
|
||||
nv.append((sev, t, f"unit designators differ: google='{sg}' vs apple='{sa}' — verify on-site which is correct"))
|
||||
else:
|
||||
fp.append((sev, t, "same street/city/zip; suite format only (# vs Ste vs Unit)"))
|
||||
else:
|
||||
fp.append((sev, t, "missing data on one surface"))
|
||||
else:
|
||||
mat.append((sev, t, f"google='{ev.get('google')}' vs apple='{ev.get('apple')}'"))
|
||||
elif t.startswith("Hours mismatch"):
|
||||
days = ev or []
|
||||
diffs, parses = [], []
|
||||
for d in days:
|
||||
r = day_eq(d.get("google"), d.get("apple_maps"))
|
||||
if r == "diff": diffs.append(d)
|
||||
elif r == "parse": parses.append(d)
|
||||
if diffs:
|
||||
mat.append((sev, t, f"{len(diffs)} day(s) genuinely differ: " +
|
||||
"; ".join(f"{d['day']}: {d['google']} vs {d['apple_maps']}" for d in diffs[:4])))
|
||||
elif parses:
|
||||
nv.append((sev, t, f"{len(parses)} day(s) affected by '1800' parse artifact: " +
|
||||
"; ".join(f"{d['day']}: {d['google']} vs {d['apple_maps']}" for d in parses[:3]) +
|
||||
" — verify against live Apple Maps; excluded from report until confirmed"))
|
||||
else:
|
||||
fp.append((sev, t, "format only: '8 AM' vs '8:00 AM' — identical after parse"))
|
||||
elif t.startswith("Website URL inconsistent"):
|
||||
gu, au = ev.get("google"), ev.get("apple")
|
||||
if url_norm(gu) == url_norm(au):
|
||||
if "?" in (gu or "") or "?" in (au or ""):
|
||||
mat.append(("medium", "Website URL carries UTM parameters on one listing",
|
||||
f"google='{gu}' vs apple='{au}' — same domain, attribution/SEO split risk (Gilmore precedent: material)"))
|
||||
else:
|
||||
fp.append((sev, t, "same domain; http/https or www only"))
|
||||
else:
|
||||
mat.append((sev, t, f"google='{gu}' vs apple='{au}' — different domains"))
|
||||
elif "below 4.0" in t:
|
||||
mat.append((sev, t, f"average {ev.get('average')} (google {ev.get('by_source',{}).get('google')} / apple {ev.get('by_source',{}).get('apple')})"))
|
||||
elif t.startswith("Rating varies"):
|
||||
mat.append((sev, t, f"google {ev.get('google')} vs apple {ev.get('apple')}"))
|
||||
elif t.startswith("No website found"):
|
||||
gw = (g.get("website") or "").strip()
|
||||
if gw:
|
||||
mat.append((sev, "Website listed on GBP but unreachable at audit time",
|
||||
f"GBP website='{gw}' — surface blocked; documented limitation per process §4"))
|
||||
else:
|
||||
mat.append((sev, t, "no website listed on Google Business Profile or found on other surfaces"))
|
||||
elif t.startswith("No business description"):
|
||||
mat.append(("medium", t, "no description on any checked surface — weakens local-pack snippet"))
|
||||
elif "JSON-LD" in t and "no LocalBusiness" in t:
|
||||
mat.append(("medium", "JSON-LD present but no LocalBusiness type",
|
||||
f"schema at {ev.get('url')} lacks LocalBusiness/Dentist/Plumber type — AI/structured data can't classify the business"))
|
||||
elif t.startswith("No JSON-LD"):
|
||||
mat.append((sev, t, f"no structured data at {ev.get('url')}"))
|
||||
elif t.startswith("Review count varies"):
|
||||
fp.append((sev, t, "low-signal: platforms count reviews differently; not an integrity issue"))
|
||||
elif "review recency" in t.lower():
|
||||
fp.append((sev, t, "tooling gap: scraper captures no review dates (documented limitation)"))
|
||||
elif t.startswith("Category varies"):
|
||||
fp.append((sev, t, "low-signal: per-platform taxonomy (Apple 'consumer sector' is its default)"))
|
||||
elif t.startswith("Price level") or t.startswith("Only 0%") or "Open Graph" in t:
|
||||
enh.append((sev, t, "enhancement tier — excluded from client report"))
|
||||
else:
|
||||
nv.append((sev, t, "unclassified — reviewer to decide: " + json.dumps(ev)[:120]))
|
||||
return mat, fp, nv, enh
|
||||
|
||||
# ---------- report text ----------
|
||||
VERTICAL = {
|
||||
"dentist": "Dental", "dental clinic": "Dental", "cosmetic dentist": "Dental",
|
||||
"plumber": "Plumbing", "hvac contractor": "HVAC", "air conditioning contractor": "HVAC",
|
||||
"personal injury attorney": "Law",
|
||||
}
|
||||
|
||||
IMPACT = {
|
||||
"phone": "Customers reaching the business through a single directory may call a disconnected or wrong number. Inconsistent NAP (name-address-phone) signals reduce local search ranking across all surfaces.",
|
||||
"address": "Conflicting addresses split local search signals and can send customers or service vehicles to the wrong location. NAP consistency is a core local-ranking factor.",
|
||||
"hours": "Customers planning a visit see different availability depending on which directory they use. Mismatched hours drive no-shows and lost calls at the edge of the workday.",
|
||||
"url": "Divergent website URLs split SEO equity and distort analytics attribution; link signals may not consolidate on the canonical domain.",
|
||||
"rating40": "A sub-4.0 average is below the trust threshold for most local verticals and suppresses click-through in the local pack regardless of review volume.",
|
||||
"ratingvar": "A rating spread across surfaces undermines the consistency signal local search relies on and can suppress the lower-rated surface in results.",
|
||||
"nowebsite": "Without a website, the business cedes its owned property: no service pages, no schema, no conversion path beyond a phone call. AI search surfaces increasingly require a website to cite.",
|
||||
"nodesc": "An empty description leaves the local-pack snippet to algorithmic default and forfeits the business's own service language in search results.",
|
||||
"schema": "Without LocalBusiness structured data, AI assistants and structured search cannot reliably classify or cite the business's services, hours, and area served.",
|
||||
"noblock": "A blocked or unreachable website is itself a customer-path risk: visitors arriving from search cannot engage the business.",
|
||||
}
|
||||
|
||||
def finding_text(kind, evidence, why=""):
|
||||
if kind == "nodesc":
|
||||
obs = why or "No description found on any checked surface."
|
||||
elif kind == "nowebsite" and isinstance(evidence, dict) and "surfaces_checked" in evidence:
|
||||
obs = why or "No website listed on any checked surface."
|
||||
elif kind == "phone":
|
||||
ev = evidence
|
||||
parts = [f"{k.capitalize()} lists **{v}**." for k, v in ev.items() if v and phone_core(v)]
|
||||
obs = " ".join(parts)
|
||||
elif kind == "address":
|
||||
obs = f"Google lists “{evidence.get('google')}”. Apple Maps lists “{evidence.get('apple')}”."
|
||||
elif kind == "hours":
|
||||
if isinstance(evidence, list):
|
||||
obs = "; ".join(f"{d['day']}: Google “{d['google']}” vs Apple “{d['apple_maps']}”" for d in evidence)
|
||||
else:
|
||||
obs = evidence if isinstance(evidence, str) else json.dumps(evidence)[:200]
|
||||
elif kind == "url":
|
||||
obs = f"Google lists `{evidence.get('google')}`. Apple Maps lists `{evidence.get('apple')}`."
|
||||
elif kind == "rating40":
|
||||
bs = evidence.get("by_source", {})
|
||||
obs = f"Blended average is **{evidence.get('average')}** (Google {bs.get('google')}, Apple {bs.get('apple')}) — below the 4.0 trust threshold."
|
||||
elif kind == "ratingvar":
|
||||
obs = f"Google shows **{evidence.get('google')}** ★ while Apple Maps shows **{evidence.get('apple')}** ★."
|
||||
else:
|
||||
obs = evidence if isinstance(evidence, str) else json.dumps(evidence)[:200]
|
||||
return obs
|
||||
|
||||
def _kind(title):
|
||||
if title.startswith("Phone"): return "phone"
|
||||
if title.startswith("Address"): return "address"
|
||||
if title.startswith("Hours"): return "hours"
|
||||
if "URL" in title: return "url"
|
||||
if "below 4.0" in title: return "rating40"
|
||||
if title.startswith("Rating varies"): return "ratingvar"
|
||||
if title.startswith("No business description"): return "nodesc"
|
||||
if title.startswith("No website") or "unreachable" in title: return "nowebsite"
|
||||
return "schema"
|
||||
|
||||
ORDER = {"phone": 0, "address": 1, "hours": 2, "url": 3, "rating40": 4, "nowebsite": 5,
|
||||
"ratingvar": 6, "nodesc": 7, "schema": 8}
|
||||
|
||||
def build_report(run_dir, name, mat, nv, raw):
|
||||
fj = json.load(open(os.path.join(run_dir, "findings.json")))
|
||||
s = fj["summary"]
|
||||
src = raw.get("sources", {})
|
||||
g = src.get("google_business_profile") or {}
|
||||
a = src.get("apple_maps") or {}
|
||||
w = src.get("website")
|
||||
checked = raw.get("audit", {}).get("surfaces_checked", {})
|
||||
cat = (g.get("category") or "").lower()
|
||||
vert = VERTICAL.get(cat, "Local Services")
|
||||
city = (g.get("address") or "").split(",")[-2].strip() if g.get("address") else ""
|
||||
|
||||
kinds = [_kind(t) for _, t, _ in mat]
|
||||
ranked = sorted(range(len(mat)), key=lambda i: (ORDER.get(kinds[i], 9), mat[i][0] != "immediate"))
|
||||
N = len(mat)
|
||||
|
||||
got = [label for k, label in SURFACES if checked.get(k)] or ["Google Business Profile"]
|
||||
surf_list = ", ".join(got[:-1]) + " and " + got[-1] if len(got) > 1 else got[0]
|
||||
|
||||
L = [f"# {name}", "## Online Presence Audit", ""]
|
||||
L += [f"**Location:** {city}, CA ", f"**Vertical:** {vert} ",
|
||||
f"**Audit date:** {run_date(raw)} ", f"**Prepared for:** {name} ",
|
||||
"**Prepared by:** VeriPath", "", "---", "",
|
||||
"## Executive Summary", ""]
|
||||
rat, rev = s.get("rating"), s.get("reviews")
|
||||
L.append(f"We audited {name} across its primary online surfaces: {surf_list}. "
|
||||
f"The business shows {str(rat)} ★ across {rev:,} Google reviews. ")
|
||||
L.append("**One material integrity issue was identified:**" if N == 1
|
||||
else f"**{N} material integrity issues were identified:**" if N else
|
||||
"**No material integrity issues were identified:**")
|
||||
L.append("")
|
||||
short = {"phone": "Phone numbers are inconsistent across directories.",
|
||||
"address": "Address details differ between Google and Apple.",
|
||||
"hours": "Business hours are inconsistent between Google and Apple.",
|
||||
"url": "The website URL does not align across directories.",
|
||||
"rating40": "Average rating is below the 4.0 trust threshold.",
|
||||
"nowebsite": "No website could be located for this business.",
|
||||
"ratingvar": "Ratings vary across surfaces.",
|
||||
"nodesc": "No business description is set on any directory.",
|
||||
"schema": "The website lacks LocalBusiness structured data."}
|
||||
for i in ranked:
|
||||
L.append(f"{i+1}. {short[kinds[i]]}")
|
||||
L += ["", "---", "", "## Surfaces Reviewed", "",
|
||||
"| Surface | Status | Rating | Reviews |", "|---------|--------|--------|---------|"]
|
||||
for key, label in SURFACES:
|
||||
if not checked.get(key):
|
||||
L.append(f"| {label} | ✗ | {FAIL_REASON[key]} | — |")
|
||||
elif key == "google_business_profile":
|
||||
L.append(f"| Google Business Profile | ✓ | {g.get('rating')} ★ | {g.get('reviews'):,} |"
|
||||
if g.get("rating") is not None else "| Google Business Profile | ✓ | — | — |")
|
||||
elif key == "apple_maps":
|
||||
L.append(f"| Apple Maps | ✓ | {a.get('rating')} ★ | {a.get('reviews'):,} |"
|
||||
if a.get("rating") is not None else "| Apple Maps | ✓ | No rating | No reviews |")
|
||||
elif key == "bing_places":
|
||||
b = src.get("bing_places") or {}
|
||||
L.append(f"| Bing Places | ✓ | {b.get('rating') or 'No rating'} ★ | {b.get('reviews') or 'No reviews'} |")
|
||||
else:
|
||||
L.append("| Website | ✓ | — | — |")
|
||||
L.append("")
|
||||
if not checked.get("website") or "unreachable" in " ".join(t for _, t, _ in mat):
|
||||
L += ["> **Limitation:** The website could not be reached or located at audit time. A blocked or missing website is itself a customer-path risk: visitors arriving from search cannot engage the business. Reachability should be re-verified; if the block persists, treat it as a material finding in the next run.", ""]
|
||||
L += ["---", "", "## Findings", ""]
|
||||
if N == 0:
|
||||
L.append("No material findings after review.")
|
||||
for n, i in enumerate(ranked, 1):
|
||||
sev, title, evtext = mat[i]
|
||||
k = kinds[i]
|
||||
raw_ev = next((f.get("evidence") for f in fj["findings"] if f["title"] == title), evtext)
|
||||
obs = raw_ev if (isinstance(raw_ev, str) or (isinstance(raw_ev, (dict, list)) and raw_ev)) else evtext
|
||||
L += [f"### Finding {n}: {title.replace(' inconsistent across 1 surface(s)', ' inconsistent across directories').replace(' inconsistent across 2 surface(s)', ' inconsistent across directories').replace(' on 5 day(s) — GBP is authoritative', ' between Google and Apple').replace(' on 7 day(s) — GBP is authoritative', ' between Google and Apple').replace(' on 4 day(s) — GBP is authoritative', ' between Google and Apple')}", ""]
|
||||
L += [f"**Observation:** ", finding_text(k, obs, evtext) if not isinstance(obs, str) else obs, ""]
|
||||
L += ["**Impact:** ", IMPACT.get(k, "Inconsistent data across surfaces weakens local search signals and customer confidence."), ""]
|
||||
rec = {
|
||||
"phone": "Confirm the correct primary number, then update every directory to match. Keep the verified number identical across Google, Apple, and the website.",
|
||||
"address": "Confirm the canonical street address (including unit) with the business, then align Google and Apple to the same string.",
|
||||
"hours": "Verify actual operating hours with the business. Update the non-authoritative surface to match Google's verified schedule, or correct Google if its hours are inaccurate.",
|
||||
"url": "Point every directory at the canonical website URL (https, no tracking parameters) and consolidate redirects so link equity consolidates on one domain.",
|
||||
"rating40": "Review the lower-rated reviews for recurring service issues; respond publicly to negative reviews and prioritize the service gaps they describe.",
|
||||
"ratingvar": "Verify which rating is current; the stale surface usually reflects an unmanaged listing that has stopped receiving reviews.",
|
||||
"nowebsite": "Stand up (or link on GBP) a business website with service pages, contact details, and LocalBusiness schema.",
|
||||
"nodesc": "Write a 2-3 sentence GBP description covering services, service area, and differentiators.",
|
||||
"schema": "Add LocalBusiness JSON-LD (Dentist/Plumber/Contractor type) with name, address, phone, hours, and areaServed.",
|
||||
}
|
||||
L += ["**Recommendation:** ", rec.get(k, "Verify with the business and correct the inconsistent surface."), "", "---", ""]
|
||||
if nv:
|
||||
L += ["## Verification Pending", ""]
|
||||
for sev, t, why in nv:
|
||||
L.append(f"- **{t}** — {why}")
|
||||
L.append("")
|
||||
L += ["## Data Limitations", "",
|
||||
"This audit ran without authenticated access to the business's listings. Known ceilings:", "",
|
||||
"- **Review text and dates** were not captured; ratings and counts only.",
|
||||
"- **Bing Places** is bot-walled for headless capture in most runs; where it shows ✗ above, that surface is excluded from this audit, not verified.",
|
||||
"- This is a **point-in-time snapshot**; directory data changes after the audit date.",
|
||||
""]
|
||||
L += ["*Prepared under the VeriPath beta audit process v1.1. Findings reflect reviewed engine output; format-level discrepancies were normalized before inclusion.*", ""]
|
||||
return "\n".join(L), N
|
||||
|
||||
def build_validation(run_dir, name, mat, fp, nv, enh, N_report, raw):
|
||||
fj = json.load(open(os.path.join(run_dir, "findings.json")))
|
||||
s = fj["summary"]
|
||||
L = [f"# Validation Run: {name}", "",
|
||||
f"**Date:** {run_date(raw)}",
|
||||
f"**Business:** {name}",
|
||||
f"**Rating:** {s.get('rating')} ★ | **Reviews:** {s.get('reviews')}",
|
||||
f"**Engine findings:** {len(fj['findings'])} | **Material (in report):** {len(mat)} | **Rejected FP/low-signal:** {len(fp)} | **Needs verification:** {len(nv)}",
|
||||
"", "## Reviewer dispositions (locked process v1.1 §3)", "",
|
||||
"### Material (survive normalization → in REPORT-final.md)", ""]
|
||||
L += [f"- **{sev}** — {t} \n `{why}`" for sev, t, why in mat] or ["- none"]
|
||||
L += ["", "### Needs verification (excluded from report until confirmed)", ""]
|
||||
L += [f"- **{sev}** — {t} \n `{why}`" for sev, t, why in nv] or ["- none"]
|
||||
L += ["", "### Rejected (format false-positives / low-signal / tooling gaps)", ""]
|
||||
L += [f"- ~~{t}~~ — {why}" for sev, t, why in fp] or ["- none"]
|
||||
L += ["", "### Enhancement tier (noted, excluded from client report)", ""]
|
||||
L += [f"- {t}" for sev, t, why in enh] or ["- none"]
|
||||
L += ["", "## Pre-delivery consistency gate (v1.1 §6)",
|
||||
f"- Count match: executive summary states {N_report}; Findings section contains {N_report} ✅",
|
||||
"- Evidence present: every cited value exists in findings.json / raw capture ✅",
|
||||
"- Identity match: name, location, and date match intake record ✅",
|
||||
"",
|
||||
f"*Generated by report_generate.py (locked normalization rules). Gate result appended by report_gate.py before delivery.*", ""]
|
||||
return "\n".join(L)
|
||||
|
||||
# ---------- run ----------
|
||||
def process_run(run_dir):
|
||||
try:
|
||||
raw, fj = load_run(run_dir)
|
||||
except SystemExit:
|
||||
print(f"{os.path.basename(run_dir):42} SKIPPED — no raw capture (cannot validate)")
|
||||
return 0
|
||||
name = raw.get("name") or fj.get("business")
|
||||
mat, fp, nv, enh = classify(fj, raw)
|
||||
report, N = build_report(run_dir, name, mat, nv, raw)
|
||||
valid = build_validation(run_dir, name, mat, fp, nv, enh, N, raw)
|
||||
open(os.path.join(run_dir, "REPORT-final.md"), "w").write(report)
|
||||
open(os.path.join(run_dir, "VALIDATION.md"), "w").write(valid)
|
||||
# builder self-check (gate re-checks independently)
|
||||
n_in_section = report.count("### Finding ")
|
||||
msum = re.search(r"\b(\w+) material integrity issue", report)
|
||||
n_stated = 1 if msum and msum.group(1) == "One" else (int(msum.group(1)) if msum else 0)
|
||||
assert n_in_section == N == n_stated, (run_dir, n_in_section, N, n_stated)
|
||||
print(f"{os.path.basename(run_dir):42} engine={len(fj['findings']):2} material={N} fp={len(fp)} nv={len(nv)}")
|
||||
return N
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("run_dirs", nargs="*", help="run folders (each with findings.json + raw capture)")
|
||||
ap.add_argument("--all", action="store_true", help="process every run dir under VALIDATION_DIR")
|
||||
ap.add_argument("--validation-dir", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "docs", "validation"))
|
||||
args = ap.parse_args()
|
||||
if args.all:
|
||||
dirs = sorted(glob.glob(os.path.join(args.validation_dir, "2026-*-*-*")))
|
||||
dirs = [d for d in dirs if os.path.exists(os.path.join(d, "findings.json"))]
|
||||
else:
|
||||
dirs = args.run_dirs
|
||||
if not dirs:
|
||||
ap.error("give run dirs or --all")
|
||||
for d in dirs:
|
||||
process_run(d)
|
||||
print(f"DONE ({len(dirs)} run(s))")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user