993 lines
37 KiB
Python
993 lines
37 KiB
Python
#!/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 <multi_surface_json> [--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 "")
|
||
|
||
# <title> 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()
|