audit pipeline v1.2: canonical business record as diff baseline (owner-verified truth), website JSON-LD hours, single-command e2e with gate

implementation/auditing/report_generate.py

@@ -144,6 +144,14 @@ def classify(fj, raw):
             else:
                 mat.append((sev, t, f"google='{ev.get('google')}' vs apple='{ev.get('apple')}'"))
         elif t.startswith("Hours mismatch"):
+            if isinstance(ev, dict) and "deviations" in ev:
+                # reference mode (canonical or GBP): engine parse-normalized, deviations are real
+                devs = ev.get("deviations") or []
+                lbl = dict(SURFACES)
+                mat.append((sev, t, f"{len(devs)} day(s) deviate from reference: " +
+                            "; ".join(f"{d['day']}: " + ", ".join(f"{lbl[k]} “{v}”" for k, v in d.items() if k in lbl)
+                                      for d in devs[:4])))
+            else:
                 days = ev or []
                 diffs, parses = [], []
                 for d in days:
@@ -189,8 +197,8 @@ def classify(fj, raw):
             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 "review recency" in t.lower() or t.lower().startswith("most recent review"):
+            fp.append((sev, t, "low-signal: review velocity not measurable from ≤3-review samples (tooling ceiling); activity judged from GBP rating/count"))
         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:
@@ -231,7 +239,13 @@ def finding_text(kind, evidence, why=""):
     elif kind == "address":
         obs = f"Google lists “{evidence.get('google')}”. Apple Maps lists “{evidence.get('apple')}”."
     elif kind == "hours":
-        if isinstance(evidence, list):
+        if isinstance(evidence, dict) and "deviations" in evidence:
+            lbl = dict(SURFACES)
+            ref_txt = ", ".join(f"{d.capitalize()} “{v}”" for d, v in sorted((evidence.get("reference_hours") or {}).items()))
+            obs = (f"Reference ({evidence.get('reference')}): {ref_txt}. Deviations: " +
+                   "; ".join(f"{d['day'].capitalize()} — " + ", ".join(f"{lbl[k]} “{v}”" for k, v in d.items() if k in lbl)
+                             for d in (evidence.get("deviations") or [])))
+        elif 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]
@@ -329,13 +343,13 @@ def build_report(run_dir, name, mat, nv, raw):
         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"### Finding {n}: {title.replace(' — owner-verified hours', ' against owner-verified schedule').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.",
+            "hours": "Verify actual operating hours with the business, then update every surface whose hours deviate from the verified schedule above (directories and website schema).",
             "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.",