#!/usr/bin/env python3 """ VeriPath Audit Engine v1 — takes raw multi-scraper JSON, produces structured findings. Each finding has: - id: unique slug (e.g. "nap_phone_mismatch") - severity: "immediate" | "high" | "medium" | "enhancement" - title: one-line summary - sub_score: one of the 7 salon-audit-v2 sub-scores - evidence: dict of raw source data backing the finding - recommendation: concrete fix Runs standalone: python audit_engine.py [--json] [--md] Output feeds directly into salon-audit-v2 skill for report generation. """ import json import re import sys import os from datetime import datetime, timezone from difflib import SequenceMatcher # ── Sub-score categories (salon-audit-v2) ────────────────────────────── DIGITAL_IDENTITY = "Digital Identity" TECHNICAL_SEO = "Technical SEO" CONVERSION = "Conversion" GOOGLE_BUSINESS = "Google Business" REVIEWS = "Reviews" CONTENT = "Content" AUTHORITY = "Authority" SEVERITY_IMMEDIATE = "immediate" # 🟥 SEVERITY_HIGH = "high" # 🟧 SEVERITY_MEDIUM = "medium" # 🟨 SEVERITY_ENHANCEMENT = "enhancement" # 🟩 def _safe_get(data, *keys, default=None): """Nested dict access without KeyError. Last positional arg is default if not a key.""" if not keys: return default cur = data for k in keys: if isinstance(cur, dict): cur = cur.get(k) if cur is None: return default else: return default return cur def _normalize_phone(phone): """Strip to digits, normalize US country code for comparison.""" if not phone: return "" d = re.sub(r'[^\d]', '', phone) # Strip leading 1 (US country code) so +1-530... and 530... match if d.startswith('1') and len(d) == 11: d = d[1:] return d def _normalize_address(addr): """Lowercase, strip extra whitespace, remove suite abbrev variances.""" if not addr: return "" s = re.sub(r'[\U000E0000-\U000EFFFF]', '', addr or '') s = re.sub(r'\s+', ' ', s.strip().lower()) s = re.sub(r'\bst[e]*\.?\b', 'ste', s) return s def _normalize_url(url): """Strip trailing slashes, normalize www, lowercase.""" if not url: return "" u = url.strip().lower().rstrip('/') return u.replace('http://www.', 'http://').replace('https://www.', 'https://') def _hours_overlap(a, b): """Check if two hours strings describe the same hours (normalize format).""" if not a or not b: return True # can't compare a_norm = re.sub(r'\s+', '', a).lower().replace(':', '').replace(' ', '') b_norm = re.sub(r'\s+', '', b).lower().replace(':', '').replace(' ', '') return a_norm == b_norm def _review_dates_from_samples(samples): """Extract dates from review samples across all sources.""" dates = [] for src_samples in samples: if not src_samples: continue for r in src_samples: d = r.get('date', '') if d: dates.append(d) return dates def _parse_review_date(date_str): """Try to parse a review date string to a datetime. Returns None on fail.""" for fmt in ("%b %d, %Y", "%B %d, %Y", "%Y-%m-%d", "%m/%d/%Y"): try: return datetime.strptime(date_str.strip(), fmt) except (ValueError, AttributeError): continue return None # ── Check functions (each returns list of findings) ──────────────────── def check_nap_phone(contract): """NAP: phone consistency across surfaces.""" findings = [] phones = {} for src_name, src_data in [("google", "google_business_profile"), ("apple", "apple_maps"), ("bing", "bing_places"), ("website", "website")]: data = _safe_get(contract, "sources", src_data, default={}) if not data: continue raw = data.get('phone') or data.get('phone_on_page') if raw: phones[src_name] = raw if len(phones) < 2: return findings normalized = {k: _normalize_phone(v) for k, v in phones.items()} first_val = list(normalized.values())[0] mismatches = {k: v for k, v in normalized.items() if v != first_val} if mismatches: findings.append({ "id": "nap_phone_mismatch", "severity": SEVERITY_IMMEDIATE, "title": f"Phone number inconsistent across {len(mismatches)} surface(s)", "sub_score": DIGITAL_IDENTITY, "evidence": {src: phones.get(src, "N/A") for src in phones}, "recommendation": "Standardize phone format across all directories. GBP is authoritative." }) return findings def check_nap_address(contract): """NAP: address consistency across surfaces.""" findings = [] addresses = {} for src_name, src_data in [("google", "google_business_profile"), ("apple", "apple_maps"), ("bing", "bing_places")]: data = _safe_get(contract, "sources", src_data, default={}) raw = data.get('address') if raw: addresses[src_name] = raw if len(addresses) < 2: return findings normalized = {k: _normalize_address(v) for k, v in addresses.items()} first_val = list(normalized.values())[0] mismatches = {k: v for k, v in normalized.items() if v != first_val} if mismatches: findings.append({ "id": "nap_address_mismatch", "severity": SEVERITY_IMMEDIATE, "title": f"Address inconsistent across {len(mismatches)} surface(s)", "sub_score": DIGITAL_IDENTITY, "evidence": {src: addresses.get(src, "N/A") for src in addresses}, "recommendation": "Standardize exact address format. GBP address is authoritative." }) return findings def check_nap_website(contract): """NAP: website URL consistency.""" findings = [] websites = {} for src_name, src_data in [("google", "google_business_profile"), ("apple", "apple_maps")]: data = _safe_get(contract, "sources", src_data, default={}) raw = data.get('website') if raw: websites[src_name] = raw if len(websites) < 2: return findings normalized = {k: _normalize_url(v) for k, v in websites.items()} first_val = list(normalized.values())[0] mismatches = {k: v for k, v in normalized.items() if v != first_val} if mismatches: findings.append({ "id": "nap_website_mismatch", "severity": SEVERITY_HIGH, "title": f"Website URL inconsistent across {len(mismatches)} surface(s)", "sub_score": DIGITAL_IDENTITY, "evidence": {src: websites.get(src, "N/A") for src in websites}, "recommendation": "Use one canonical URL (prefer HTTPS, no www) across all directories." }) return findings def check_hours_mismatch(contract): """Hours consistency vs reference: canonical record (owner-verified truth) if the client has one, else GBP (legacy surface-vs-surface). Evidence shape: {reference, reference_hours, provenance, deviations[]} where each deviation day carries the RAW value of every captured surface. """ findings = [] from canonical_baseline import resolve_reference, normalize_hours_dict, parse_hours_value label, ref_hours, prov = resolve_reference(contract, contract.get("_canonical_record")) ref_norm = normalize_hours_dict(ref_hours or {}) if not ref_norm: return findings # nothing to compare against sources = contract.get("sources", {}) SURF_KEYS = ("google_business_profile", "apple_maps", "bing_places", "website") deviations = [] for day, ref_val in sorted(ref_norm.items()): raw_vals = {} for sname in SURF_KEYS: v = (sources.get(sname) or {}).get("hours") or {} if v.get(day) is not None: raw_vals[sname] = v[day] if not raw_vals: continue # ponytail: skip the reference surface itself in legacy mode (it can't deviate from itself anyway) bad = any( parse_hours_value(v) is not None and parse_hours_value(v) != ref_val for v in raw_vals.values() ) if bad: entry = {"day": day, "reference": ref_val} entry.update(raw_vals) deviations.append(entry) if deviations: suffix = "owner-verified hours" if prov else "GBP is authoritative" findings.append({ "id": "hours_mismatch", "severity": SEVERITY_IMMEDIATE, "title": f"Hours mismatch on {len(deviations)} day(s) — {suffix}", "sub_score": GOOGLE_BUSINESS, "evidence": { "reference": label, "reference_hours": ref_norm, "provenance": prov, "deviations": deviations, }, "recommendation": ("Correct hours on the deviating surfaces to match the owner-verified schedule (canonical record)." if prov else "Correct hours on non-GBP surfaces to match GBP. Mismatched hours cause customer complaints and trust loss."), }) return findings def check_rating_health(contract): """Rating delta across surfaces and overall health.""" findings = [] ratings = {} for src_name, src_data in [("google", "google_business_profile"), ("apple", "apple_maps"), ("bing", "bing_places")]: data = _safe_get(contract, "sources", src_data, default={}) r = data.get('rating') if r is not None: ratings[src_name] = float(r) if not ratings: return findings avg = sum(ratings.values()) / len(ratings) delta = max(ratings.values()) - min(ratings.values()) if delta > 0.3: findings.append({ "id": "rating_delta_high", "severity": SEVERITY_HIGH, "title": f"Rating varies by {delta:.1f} across surfaces (range: {min(ratings.values()):.1f}–{max(ratings.values()):.1f})", "sub_score": REVIEWS, "evidence": ratings, "recommendation": "Investigate why some surfaces show different ratings. May indicate stale data or mixed-up business listings." }) if avg < 4.0: findings.append({ "id": "rating_below_threshold", "severity": SEVERITY_IMMEDIATE, "title": f"Average rating {avg:.1f} — below 4.0 threshold", "sub_score": REVIEWS, "evidence": {"average": round(avg, 2), "by_source": ratings}, "recommendation": "Address negative reviews, request new reviews from satisfied customers. Below 4.0 is a strong conversion blocker." }) return findings def check_review_count_delta(contract): """Review count variation across surfaces.""" findings = [] counts = {} for src_name, src_data in [("google", "google_business_profile"), ("apple", "apple_maps"), ("bing", "bing_places")]: data = _safe_get(contract, "sources", src_data, default={}) c = data.get('reviews') if c is not None: counts[src_name] = int(c) if len(counts) < 2: return findings delta = max(counts.values()) - min(counts.values()) if delta > 5: findings.append({ "id": "review_count_delta", "severity": SEVERITY_MEDIUM, "title": f"Review count varies by {delta} across surfaces", "sub_score": REVIEWS, "evidence": counts, "recommendation": "Large review count gaps suggest some surfaces have stale or merged data. Verify listing ownership." }) return findings def check_onpage_title_meta(contract): """On-page title/meta audit (technical SEO): presence, length, name, phone. Reads the website fields captured at scrape time (title, description, phone_on_page) so the audit is reproducible from the capture. Skips cleanly when no website surface was captured.""" findings = [] ws = _safe_get(contract, "sources", "website", default={}) or {} title = ws.get("title") or "" desc = ws.get("description") or "" if not title and not desc: return findings # no website surface captured name = (contract.get("name") or "").lower() phone_norm = _normalize_phone(contract.get("phone") or "") page_phone_norm = _normalize_phone(ws.get("phone_on_page") or "") # presence & length (Google truncates ~60 chars) if not title: findings.append({ "id": "missing_title", "severity": SEVERITY_HIGH, "sub_score": TECHNICAL_SEO, "title": "Website has no <title> tag — search engines fall back to the URL.", "evidence": {"title": ""}, "recommendation": "Add a descriptive <title> tag including the business name and main service.", }) else: if len(title) > 65: findings.append({ "id": "title_too_long", "severity": SEVERITY_ENHANCEMENT, "sub_score": TECHNICAL_SEO, "title": f"Title tag is {len(title)} chars (>65) — truncated in Google results.", "evidence": {"title": title, "length": len(title)}, "recommendation": "Shorten the <title> to ~50-60 characters.", }) if name and name not in title.lower(): findings.append({ "id": "title_missing_name", "severity": SEVERITY_MEDIUM, "sub_score": TECHNICAL_SEO, "title": "Title tag does not contain the business name.", "evidence": {"title": title, "business_name": name}, "recommendation": "Include the business name in the <title> for brand recognition.", }) # Meta description presence & length (Google truncates ~160 chars) if not desc: findings.append({ "id": "missing_meta_description", "severity": SEVERITY_MEDIUM, "sub_score": TECHNICAL_SEO, "title": "No meta description — Google may use arbitrary page text as the snippet.", "evidence": {"description": ""}, "recommendation": "Write a 150-160 character meta description naming the business and service.", }) elif len(desc) > 170: findings.append({ "id": "meta_description_too_long", "severity": SEVERITY_ENHANCEMENT, "sub_score": TECHNICAL_SEO, "title": f"Meta description is {len(desc)} chars (>170) — truncated in results.", "evidence": {"length": len(desc)}, "recommendation": "Shorten the meta description to ~150-160 characters.", }) # On-page NAP: flag only when a phone WAS captured and it differs from the # business phone (evidence-based; no capture = can't conclude, don't flag). if phone_norm and page_phone_norm and phone_norm != page_phone_norm: findings.append({ "id": "phone_not_on_page", "severity": SEVERITY_MEDIUM, "sub_score": DIGITAL_IDENTITY, "title": "Phone number found on the website differs from the business phone (on-page NAP gap).", "evidence": {"business_phone": phone_norm, "phone_on_page": page_phone_norm}, "recommendation": "Make the website phone match the business phone in the header/footer.", }) return findings def check_review_velocity(contract): """Review recency — are recent reviews flowing?""" findings = [] review_samples = [ _safe_get(contract, "sources", "apple_maps", "reviews_sample", default=[]), _safe_get(contract, "sources", "bing_places", "reviews_sample", default=[]), ] dates = _review_dates_from_samples(review_samples) if not dates: findings.append({ "id": "review_dates_unavailable", "severity": SEVERITY_ENHANCEMENT, # tooling limitation, not a business finding (matches classify() FP in report_generate.py) "title": "Cannot determine review recency — no dates available in samples", "sub_score": REVIEWS, "evidence": {"sources_with_dates": []}, "recommendation": "Review date data is not available from current surfaces. Monitor via GBP dashboard directly." }) return findings parsed = [_parse_review_date(d) for d in dates] parsed = [p for p in parsed if p is not None] if not parsed: return findings newest = max(parsed) now = datetime.now(timezone.utc).replace(tzinfo=None) days_since_newest = (now - newest).days if days_since_newest > 90: findings.append({ "id": "reviews_stale", "severity": SEVERITY_HIGH, "title": f"Most recent review is {days_since_newest} days old — no fresh review signal", "sub_score": REVIEWS, "evidence": {"newest_review_date": dates[0], "days_ago": days_since_newest}, "recommendation": "No recent reviews = no social proof for new visitors. Implement review request workflow post-appointment." }) return findings def _service_keywords(contract): """Derive the business's own service/category vocabulary (vertical-agnostic). Grounded in what the business actually sells — its category, description, and name — so the specificity signal adapts to any vertical instead of assuming one (the old version hardcoded salon terms). """ words = set() # Top-level capture fields are the most reliable vocabulary source for field in ("name", "category", "description"): val = contract.get(field) or "" for w in re.findall(r"[a-z]{3,}", val.lower()): words.add(w) for src in ["google_business_profile", "apple_maps", "bing_places", "website"]: data = _safe_get(contract, "sources", src, default={}) or {} for field in ("category", "description", "name", "title"): val = data.get(field) or "" for w in re.findall(r"[a-z]{3,}", val.lower()): words.add(w) # Drop generic words that don't indicate a specific service/product stop = {"the", "and", "for", "with", "this", "that", "your", "you", "our", "are", "was", "have", "has", "been", "will", "would", "like", "get", "got", "good", "great", "best", "very", "really", "service", "services", "business", "place", "store", "shop", "location", "area", "city", "staff", "customer", "customers", "experience", "quality", "professional", "professionals", "helpful", "friendly", "recommend", "recommended", "price", "prices", "paid", "cost", "costs", "time", "times", "day", "days", "week", "weeks", "month", "months", "year", "years", "online", "presence", "audit", "veripath"} return words - stop def check_review_text_quality(contract): """Check if reviews mention the business's own services/products (E-E-A-T signal). Vertical-agnostic: the keyword set is derived from the business's own category/description/name, not a hardcoded vertical. """ findings = [] all_samples = [] for src in ["apple_maps", "bing_places"]: samples = _safe_get(contract, "sources", src, "reviews_sample", default=[]) all_samples.extend(samples or []) if len(all_samples) < 3: return findings keywords = _service_keywords(contract) if not keywords: return findings # no vocabulary to match against; can't judge specificity specific_signals = 0 for r in all_samples: text = (r.get('text') or '').lower() if any(w in text for w in keywords): specific_signals += 1 ratio = specific_signals / len(all_samples) if ratio < 0.5: findings.append({ "id": "reviews_low_specificity", "severity": SEVERITY_ENHANCEMENT, "title": f"Only {ratio:.0%} of sample reviews mention specific services or staff", "sub_score": REVIEWS, "evidence": {"sample_size": len(all_samples), "specific_count": specific_signals, "keyword_basis": "category+description+name"}, "recommendation": "Encourage reviews that name specific services, products, or staff. Specific reviews rank higher and convert better." }) return findings def check_category_alignment(contract): """Category consistency and fragmentation across surfaces.""" findings = [] categories = {} for src_name, src_data in [("google", "google_business_profile"), ("apple", "apple_maps"), ("bing", "bing_places")]: data = _safe_get(contract, "sources", src_data, default={}) cat = data.get('category') if cat: categories[src_name] = cat.lower().strip() if len(categories) < 2: return findings unique = set(categories.values()) if len(unique) > 1: findings.append({ "id": "category_fragmentation", "severity": SEVERITY_MEDIUM, "title": f"Category varies across surfaces: {', '.join(sorted(unique))}", "sub_score": DIGITAL_IDENTITY, "evidence": categories, "recommendation": "Align primary category across all directories. GBP primary category is the ranking driver." }) return findings def check_closed_status(contract): """Check if GBP shows any closed status.""" findings = [] closed = _safe_get(contract, "closed_status", default={}) if closed.get('permanently_closed'): findings.append({ "id": "gbp_permanently_closed", "severity": SEVERITY_IMMEDIATE, "title": "GBP marked as permanently closed", "sub_score": GOOGLE_BUSINESS, "evidence": closed, "recommendation": "This kills all local visibility. Reopen the GBP immediately via google.com/business." }) if closed.get('temporarily_closed'): findings.append({ "id": "gbp_temporarily_closed", "severity": SEVERITY_HIGH, "title": "GBP marked as temporarily closed", "sub_score": GOOGLE_BUSINESS, "evidence": closed, "recommendation": "Remove temporary closure status. Temp-closed listings lose ranking and impressions." }) return findings def check_description(contract): """Check if any surface has a business description.""" findings = [] has_desc = False for src in ["google_business_profile", "bing_places", "website"]: data = _safe_get(contract, "sources", src, default={}) if data: desc = data.get('description') or data.get('description') if desc and len(desc) > 20: has_desc = True break if not has_desc: findings.append({ "id": "no_description", "severity": SEVERITY_MEDIUM, "title": "No business description found on any surface", "sub_score": CONTENT, "evidence": {"surfaces_checked": ["google", "bing", "website"]}, "recommendation": "Add a keyword-rich description to GBP and website. Descriptions improve click-through and category relevance." }) return findings def check_price_level(contract): """Check if price level is available.""" findings = [] has_price = False for src in ["google_business_profile", "apple_maps"]: data = _safe_get(contract, "sources", src, default={}) if data and data.get('price_level'): has_price = True break if not has_price: findings.append({ "id": "no_price_level", "severity": SEVERITY_ENHANCEMENT, "title": "Price level not set on GBP or Apple Maps", "sub_score": CONTENT, "evidence": {}, "recommendation": "Set price level ($–$$$$) on GBP. Helps users self-select and improves relevance." }) return findings def check_photos(contract): """Check if photo count is available.""" findings = [] photos = _safe_get(contract, "sources", "google_business_profile", "photos_count") if photos is not None and photos == 0: findings.append({ "id": "no_photos", "severity": SEVERITY_HIGH, "title": "GBP has zero photos", "sub_score": GOOGLE_BUSINESS, "evidence": {"photos_count": 0}, "recommendation": "Upload 20+ photos: exterior, interior, team, services, before/after. Photos increase booking by 42%." }) elif photos is not None and 0 < photos < 10: findings.append({ "id": "low_photos", "severity": SEVERITY_MEDIUM, "title": f"GBP has only {photos} photo(s)", "sub_score": GOOGLE_BUSINESS, "evidence": {"photos_count": photos}, "recommendation": "Aim for 20+ photos. More photos = more engagement and trust." }) return findings def check_website_schema(contract): """Check if website has JSON-LD schema (LocalBusiness/BeautySalon/etc).""" findings = [] website_url = _safe_get(contract, "sources", "website", "source_url") if not website_url: findings.append({ "id": "no_website", "severity": SEVERITY_HIGH, "title": "No website found", "sub_score": TECHNICAL_SEO, "evidence": {}, "recommendation": "A website is essential for credibility, schema markup, and conversion. Launch one immediately." }) return findings # Prefer capture-time evidence (reproducible audit). Live refetch only for # pre-fix captures that lack the fields. ws = _safe_get(contract, "sources", "website", default={}) or {} if "has_jsonld" in ws: has_jsonld = bool(ws.get("has_jsonld")) types = [t.lower() for t in ws.get("schema_types") or []] has_local_business = any(t in ("localbusiness", "beautysalon", "dayspa", "healthandbeautybusiness") for t in types) has_canonical = bool(ws.get("canonical")) has_og = bool(ws.get("has_og")) else: # Fetch and check for JSON-LD (legacy captures) try: import urllib.request req = urllib.request.Request(website_url, headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=10) as resp: html = resp.read().decode("utf-8", errors="replace") except Exception as e: findings.append({ "id": "website_unreachable", "severity": SEVERITY_HIGH, "title": f"Website fetch failed: {e}", "sub_score": TECHNICAL_SEO, "evidence": {"url": website_url, "error": str(e)}, "recommendation": "Verify website is live and accessible. A broken site kills all local signals." }) return findings # Check for JSON-LD has_jsonld = bool(re.search(r'<script[^>]*type=["\']application/ld\+json["\']', html, re.IGNORECASE)) has_local_business = bool(re.search(r'"@type"\s*:\s*"(?:LocalBusiness|BeautySalon|DaySpa|HealthAndBeautyBusiness)', html, re.IGNORECASE)) # Check for meta canonical has_canonical = bool(re.search(r'<link[^>]*rel=["\']canonical["\']', html, re.IGNORECASE)) # Check for OG tags has_og = bool(re.search(r'<meta[^>]*property=["\']og:', html, re.IGNORECASE)) if not has_jsonld: findings.append({ "id": "missing_jsonld_schema", "severity": SEVERITY_HIGH, "title": "No JSON-LD schema found on website", "sub_score": TECHNICAL_SEO, "evidence": {"url": website_url, "has_jsonld": False}, "recommendation": "Add LocalBusiness/BeautySalon JSON-LD schema. Critical for rich results and local ranking." }) elif not has_local_business: findings.append({ "id": "schema_wrong_type", "severity": SEVERITY_MEDIUM, "title": "JSON-LD present but no LocalBusiness/BeautySalon type", "sub_score": TECHNICAL_SEO, "evidence": {"url": website_url, "has_jsonld": True, "has_local_business": False}, "recommendation": "Ensure schema includes LocalBusiness or BeautySalon @type for local search relevance." }) if not has_canonical: findings.append({ "id": "missing_canonical", "severity": SEVERITY_MEDIUM, "title": "No canonical URL tag on website", "sub_score": TECHNICAL_SEO, "evidence": {"url": website_url}, "recommendation": "Add <link rel='canonical' href='...'> to prevent duplicate content issues." }) if not has_og: findings.append({ "id": "missing_og_tags", "severity": SEVERITY_ENHANCEMENT, "title": "No Open Graph tags on website", "sub_score": TECHNICAL_SEO, "evidence": {"url": website_url}, "recommendation": "Add OG tags for proper link preview on social media and messaging apps." }) return findings def check_google_maps_url(contract): """Check if GBP has a proper Google Maps URL.""" findings = [] gmaps_url = _safe_get(contract, "sources", "google_business_profile", "url") if not gmaps_url: findings.append({ "id": "no_gmaps_url", "severity": SEVERITY_HIGH, "title": "No Google Maps URL found — listing may be unclaimed", "sub_score": GOOGLE_BUSINESS, "evidence": {}, "recommendation": "Claim and verify the GBP at google.com/business. Unclaimed listings cannot be managed." }) return findings def check_coordinates_precision(contract): """Check if coordinates are precise enough (pin dropped vs exact).""" findings = [] coords = _safe_get(contract, "coordinates", default={}) lat = coords.get('lat') lon = coords.get('lon') if lat and lon: # Pin-dropped pins usually have fewer decimal places lat_precision = len(str(lat).split('.')[-1]) if '.' in str(lat) else 0 lon_precision = len(str(lon).split('.')[-1]) if '.' in str(lon) else 0 if min(lat_precision, lon_precision) < 6: findings.append({ "id": "low_coordinate_precision", "severity": SEVERITY_MEDIUM, "title": f"Coordinates may be pin-dropped ({lat_precision}/{lon_precision} decimal places)", "sub_score": GOOGLE_BUSINESS, "evidence": {"lat": lat, "lon": lon, "lat_precision": lat_precision, "lon_precision": lon_precision}, "recommendation": "Set exact pin location on GBP. Pin-dropped locations lose precision ranking." }) return findings # ── Main engine ────────────────────────────────────────────────────────── ALL_CHECKS = [ check_nap_phone, check_nap_address, check_nap_website, check_hours_mismatch, check_rating_health, check_review_count_delta, check_review_velocity, check_review_text_quality, check_category_alignment, check_closed_status, check_description, check_price_level, check_photos, check_website_schema, check_onpage_title_meta, check_google_maps_url, check_coordinates_precision, ] SEVERITY_ORDER = { SEVERITY_IMMEDIATE: 0, SEVERITY_HIGH: 1, SEVERITY_MEDIUM: 2, SEVERITY_ENHANCEMENT: 3, } def analyze(contract): """Run all checks on a multi-scraper contract. Returns sorted findings list.""" all_findings = [] for check_fn in ALL_CHECKS: try: findings = check_fn(contract) all_findings.extend(findings) except Exception as e: print(f" [WARN] {check_fn.__name__} failed: {e}", file=sys.stderr) # Deduplicate by id seen = set() deduped = [] for f in all_findings: if f["id"] not in seen: seen.add(f["id"]) deduped.append(f) # Sort by severity deduped.sort(key=lambda x: SEVERITY_ORDER.get(x["severity"], 99)) return deduped def summarize(findings, contract): """Return a summary dict with counts and scores.""" counts = {"immediate": 0, "high": 0, "medium": 0, "enhancement": 0} for f in findings: counts[f["severity"]] = counts.get(f["severity"], 0) + 1 # Sub-score health (count findings per sub-score) sub_scores = {} for f in findings: ss = f["sub_score"] sub_scores[ss] = sub_scores.get(ss, 0) + 1 surfaces = _safe_get(contract, "audit", "surfaces_checked", default={}) rating = _safe_get(contract, "rating") reviews = _safe_get(contract, "reviews") return { "total_findings": len(findings), "severity_counts": counts, "sub_score_issues": sub_scores, "surfaces_ok": {k: v for k, v in surfaces.items() if v}, "surfaces_failed": {k: v for k, v in surfaces.items() if not v}, "rating": rating, "reviews": reviews, } def to_markdown(findings, summary, contract): """Render findings as markdown for salon-audit-v2 consumption.""" lines = [] name = contract.get("name", "Unknown") lines.append(f"# Audit Findings — {name}") lines.append("") lines.append(f"**Generated:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}") lines.append(f"**Sources checked:** {', '.join(summary['surfaces_ok'].keys()) or 'None'}") lines.append(f"**Rating:** {contract.get('rating', 'N/A')} | **Reviews:** {contract.get('reviews', 'N/A')}") lines.append("") # Severity summary icons = {"immediate": "🟥", "high": "🟧", "medium": "🟨", "enhancement": "🟩"} sc = summary["severity_counts"] lines.append("## Severity Summary") lines.append("") lines.append(f"| Severity | Count |") lines.append(f"|----------|-------|") for sev in ["immediate", "high", "medium", "enhancement"]: lines.append(f"| {icons.get(sev, '')} {sev.capitalize()} | {sc.get(sev, 0)} |") lines.append(f"| **Total** | **{summary['total_findings']}** |") lines.append("") # Findings lines.append("## Findings") lines.append("") for i, f in enumerate(findings, 1): icon = icons.get(f["severity"], "") lines.append(f"### Finding #{i}: {f['title']}") lines.append("") lines.append(f"- **Severity:** {icon} {f['severity'].capitalize()}") lines.append(f"- **Category:** {f['sub_score']}") lines.append(f"- **Recommendation:** {f['recommendation']}") if f.get("evidence"): lines.append(f"- **Evidence:** ```json\n{json.dumps(f['evidence'], indent=2)}\n```") lines.append("") # Surfaces that failed if summary["surfaces_failed"]: lines.append("## Failed Surfaces") lines.append("") for src in summary["surfaces_failed"]: lines.append(f"- ❌ `{src}` — no data returned") lines.append("") return "\n".join(lines) def main(): if len(sys.argv) < 2: print("Usage: audit_engine.py <multi_surface_json> [--json] [--md] [--output-dir DIR]") sys.exit(1) filepath = sys.argv[1] output_json = "--json" in sys.argv output_md = "--md" in sys.argv # --output-dir: write results to directory instead of stdout output_dir = None i = 2 while i < len(sys.argv): if sys.argv[i] == "--output-dir" and i + 1 < len(sys.argv): output_dir = sys.argv[i + 1] i += 2 continue i += 1 if not os.path.exists(filepath): print(f"Error: {filepath} not found") sys.exit(1) with open(filepath) as f: contract = json.load(f) # Canonical baseline: owner-verified record from Gitea (None -> legacy GBP reference) try: from canonical_baseline import fetch_canonical_record, canonical_hours record = fetch_canonical_record(contract.get("name") or "") if record and canonical_hours(record): contract["_canonical_record"] = record print(f" [OK] Canonical baseline: {record['_path']} @ {record['_commit']}", file=sys.stderr) else: print(" [INFO] No verified canonical record — legacy surface-vs-surface reference", file=sys.stderr) except Exception as e: print(f" [WARN] canonical fetch failed ({e}) — legacy reference", file=sys.stderr) findings = analyze(contract) summary = summarize(findings, contract) result = { "audit_engine": "v1", "timestamp": datetime.now(timezone.utc).isoformat(), "business": _safe_get(contract, "name"), "summary": summary, "findings": findings, } if output_dir: os.makedirs(output_dir, exist_ok=True) json_path = os.path.join(output_dir, "findings.json") with open(json_path, "w") as f: json.dump(result, f, indent=2) print(f"Findings written to {json_path}", file=sys.stderr) if output_md: md = to_markdown(findings, summary, contract) md_path = os.path.join(output_dir, "findings.md") with open(md_path, "w") as f: f.write(md) print(f"Report written to {md_path}", file=sys.stderr) else: if output_md: md = to_markdown(findings, summary, contract) print(md) else: print(json.dumps(result, indent=2)) if __name__ == "__main__": main()