Leonard (VeriPath Agent)
6583c85871
- _normalize_address: drop commas, map 'suite'->'ste' so '3460 Robin Ln, Ste 4' == '3460 Robin Ln Ste 4' (no false address mismatch) - check_category_alignment: add _CATEGORY_SYNONYMS map so 'beauty salon' / 'beauty and spa' / 'beauty & spa' collapse; real drift (e.g. 'medical spa', 'barber shop') still fires - Phoenix proof run drops from 4 findings to 2 (removes 2 false positives)
867 lines
31 KiB
Python
867 lines
31 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, drop commas."""
|
||
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)
|
||
s = re.sub(r'\bsuite\b', 'ste', s)
|
||
s = s.replace(',', '') # ponytail: comma is pure formatting, not a material diff
|
||
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. GBP is authoritative."""
|
||
findings = []
|
||
gbp_hours = _safe_get(contract, "sources", "google_business_profile", "hours", default={})
|
||
apple_hours = _safe_get(contract, "sources", "apple_maps", "hours", default={})
|
||
bing_hours = _safe_get(contract, "sources", "bing_places", "hours", default={})
|
||
|
||
if not gbp_hours:
|
||
return findings # nothing to compare against
|
||
|
||
mismatched_days = []
|
||
for other_name, other_hours in [("apple_maps", apple_hours), ("bing_places", bing_hours)]:
|
||
if not other_hours:
|
||
continue
|
||
for day, gbp_val in gbp_hours.items():
|
||
if day in other_hours:
|
||
other_val = other_hours[day]
|
||
if not _hours_overlap(gbp_val, other_val):
|
||
mismatched_days.append({
|
||
"day": day,
|
||
"google": gbp_val,
|
||
other_name: other_val
|
||
})
|
||
|
||
if mismatched_days:
|
||
findings.append({
|
||
"id": "hours_mismatch",
|
||
"severity": SEVERITY_IMMEDIATE,
|
||
"title": f"Hours mismatch on {len(mismatched_days)} day(s) — GBP is authoritative",
|
||
"sub_score": GOOGLE_BUSINESS,
|
||
"evidence": mismatched_days,
|
||
"recommendation": "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_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_MEDIUM,
|
||
"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 check_review_text_quality(contract):
|
||
"""Check if reviews mention specific services/practitioners (E-E-A-T signal)."""
|
||
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 not all_samples:
|
||
return findings
|
||
|
||
specific_signals = 0
|
||
for r in all_samples:
|
||
text = (r.get('text') or '').lower()
|
||
if any(word in text for word in ['ashley', 'grace', 'mallory', 'facial', 'hair', 'waxing', 'lashes', 'spray tan', 'massage']):
|
||
specific_signals += 1
|
||
|
||
ratio = specific_signals / len(all_samples) if all_samples else 0
|
||
if ratio < 0.5 and len(all_samples) >= 3:
|
||
findings.append({
|
||
"id": "reviews_low_specificity",
|
||
"severity": SEVERITY_ENHANCEMENT,
|
||
"title": f"Only {ratio:.0%} of sample reviews mention specific services or practitioners",
|
||
"sub_score": REVIEWS,
|
||
"evidence": {"sample_size": len(all_samples), "specific_count": specific_signals},
|
||
"recommendation": "Encourage detailed reviews mentioning services and staff. Specific reviews rank higher and convert better."
|
||
})
|
||
|
||
return findings
|
||
|
||
|
||
# ponytail: small explicit synonym map; extend when new variants appear in the wild
|
||
_CATEGORY_SYNONYMS = {
|
||
"beauty salon": "beauty salon",
|
||
"beauty and spa": "beauty salon",
|
||
"beauty & spa": "beauty salon",
|
||
"beauty & spa": "beauty salon",
|
||
"salon and spa": "beauty salon",
|
||
"hair salon": "hair salon",
|
||
"hair & beauty salon": "hair salon",
|
||
"hair and beauty salon": "hair salon",
|
||
"day spa": "day spa",
|
||
"spa": "spa",
|
||
"medical spa": "medical spa",
|
||
"barber shop": "barber shop",
|
||
"barbershop": "barber shop",
|
||
}
|
||
|
||
|
||
def _normalize_category(cat):
|
||
if not cat:
|
||
return ""
|
||
c = re.sub(r'\s+', ' ', cat.strip().lower()).replace('&', '&')
|
||
return _CATEGORY_SYNONYMS.get(c, c)
|
||
|
||
|
||
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] = _normalize_category(cat)
|
||
|
||
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
|
||
|
||
# Fetch and check for JSON-LD
|
||
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")
|
||
|
||
# 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."
|
||
})
|
||
|
||
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
|
||
|
||
|
||
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_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)
|
||
|
||
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()
|