feat(auditing): add audit_engine.py — 16 deterministic checks with severity + evidence
This commit is contained in:
@@ -0,0 +1,824 @@
|
||||
#!/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."""
|
||||
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. 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
|
||||
|
||||
|
||||
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
|
||||
|
||||
# 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]")
|
||||
sys.exit(1)
|
||||
|
||||
filepath = sys.argv[1]
|
||||
output_json = "--json" in sys.argv
|
||||
output_md = "--md" in sys.argv
|
||||
|
||||
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)
|
||||
|
||||
if output_md:
|
||||
md = to_markdown(findings, summary, contract)
|
||||
print(md)
|
||||
elif output_json:
|
||||
result = {
|
||||
"audit_engine": "v1",
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"business": _safe_get(contract, "name"),
|
||||
"summary": summary,
|
||||
"findings": findings,
|
||||
}
|
||||
print(json.dumps(result, indent=2))
|
||||
else:
|
||||
# Default: JSON
|
||||
result = {
|
||||
"audit_engine": "v1",
|
||||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||||
"business": _safe_get(contract, "name"),
|
||||
"summary": summary,
|
||||
"findings": findings,
|
||||
}
|
||||
print(json.dumps(result, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user