audit pipeline v1.2: canonical business record as diff baseline (owner-verified truth), website JSON-LD hours, single-command e2e with gate
This commit is contained in:
@@ -19,7 +19,7 @@ import os
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
# Ensure scraper is importable
|
# Ensure scraper is importable
|
||||||
sys.path.insert(0, os.path.expanduser("~/deps/google-maps-scraper/src"))
|
sys.path.insert(0, "/tmp/google-maps-scraper/src")
|
||||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||||
|
|
||||||
from gmaps_scraper.scraper import GoogleMapsScraper
|
from gmaps_scraper.scraper import GoogleMapsScraper
|
||||||
@@ -130,84 +130,58 @@ async def scrape_google(query, coords, business=""):
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _valid_phone(num):
|
|
||||||
"""10 US digits, not an all-same-digit placeholder (999-999-9999, the
|
|
||||||
value call-tracking widgets emit before they're wired)."""
|
|
||||||
d = re.sub(r"\D", "", num)
|
|
||||||
return len(d) == 10 and len(set(d)) > 1
|
|
||||||
|
|
||||||
|
|
||||||
def _phone_from_text(text):
|
|
||||||
"""First standalone, non-placeholder US phone number in rendered text.
|
|
||||||
|
|
||||||
Lookarounds reject a phone-like substring glued to other digits (the
|
|
||||||
raw-HTML path that previously returned JS artifact numbers);
|
|
||||||
_valid_phone rejects all-same-digit call-tracking placeholders.
|
|
||||||
"""
|
|
||||||
for pat in (r"(?<!\d)(\(\d{3}\)\s*\d{3}[-.]\d{4})(?!\d)",
|
|
||||||
r"(?<!\d)(\d{3}[-.]\d{3}[-.]\d{4})(?!\d)"):
|
|
||||||
for mm in re.finditer(pat, text):
|
|
||||||
if _valid_phone(mm.group(1)):
|
|
||||||
return mm.group(1)
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
|
||||||
def _jina_markdown(website_url):
|
|
||||||
"""Render a URL to clean text via Jina Reader (no key, free tier)."""
|
|
||||||
import urllib.request
|
|
||||||
req = urllib.request.Request(
|
|
||||||
f"https://r.jina.ai/{website_url}",
|
|
||||||
headers={"User-Agent": "Mozilla/5.0", "Accept": "text/plain"},
|
|
||||||
)
|
|
||||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
||||||
return resp.read().decode("utf-8", errors="replace")
|
|
||||||
|
|
||||||
|
|
||||||
def fetch_website_data(website_url):
|
def fetch_website_data(website_url):
|
||||||
"""Extract basic info from business website.
|
"""Extract basic info from business website."""
|
||||||
|
|
||||||
Direct fetch for title/description (in <head>, present on JS sites too).
|
|
||||||
Phone is the NAP-critical field: if the direct HTML has no visible phone,
|
|
||||||
re-render via Jina Reader, because business sites usually expose the
|
|
||||||
phone client-side and a false 'no phone' would corrupt the cross-check.
|
|
||||||
"""
|
|
||||||
if not website_url:
|
if not website_url:
|
||||||
return None
|
return None
|
||||||
import urllib.request
|
import urllib.request
|
||||||
html = None
|
|
||||||
try:
|
try:
|
||||||
req = urllib.request.Request(website_url, headers={"User-Agent": "Mozilla/5.0"})
|
req = urllib.request.Request(website_url, headers={"User-Agent": "Mozilla/5.0"})
|
||||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||||
html = resp.read().decode("utf-8", errors="replace")
|
html = resp.read().decode("utf-8", errors="replace")
|
||||||
except Exception as e:
|
|
||||||
print(f" [WARN] Website direct fetch failed ({e}); using Jina Reader")
|
|
||||||
data = {"source_url": website_url}
|
data = {"source_url": website_url}
|
||||||
if html:
|
|
||||||
m = re.search(r"<title>([^<]+)</title>", html, re.IGNORECASE)
|
# Extract title
|
||||||
|
m = re.search(r'<title>([^<]+)</title>', html, re.IGNORECASE)
|
||||||
if m:
|
if m:
|
||||||
data["title"] = m.group(1).strip()
|
data["title"] = m.group(1).strip()
|
||||||
|
|
||||||
|
# Extract meta description
|
||||||
m = re.search(r'<meta[^>]*name=["\']description["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
|
m = re.search(r'<meta[^>]*name=["\']description["\'][^>]*content=["\']([^"\']+)["\']', html, re.IGNORECASE)
|
||||||
if not m:
|
if not m:
|
||||||
m = re.search(r'<meta[^>]*content=["\']([^"\']+)["\'][^>]*name=["\']description["\']', html, re.IGNORECASE)
|
m = re.search(r'<meta[^>]*content=["\']([^"\']+)["\'][^>]*name=["\']description["\']', html, re.IGNORECASE)
|
||||||
if m:
|
if m:
|
||||||
data["description"] = m.group(1).strip()
|
data["description"] = m.group(1).strip()
|
||||||
data["phone_on_page"] = _phone_from_text(html)
|
|
||||||
if data.get("phone_on_page"):
|
# Extract phone from page
|
||||||
return data
|
phone_match = re.search(r'(\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{4})', html)
|
||||||
# No phone in direct HTML (or fetch failed) -> render via Jina Reader
|
if phone_match:
|
||||||
|
data["phone_on_page"] = phone_match.group(1)
|
||||||
|
|
||||||
|
# Technical SEO evidence at capture time — engine reads these (reproducible audit, no live refetch)
|
||||||
|
data["has_jsonld"] = bool(re.search(r'<script[^>]*type=["\']application/ld\+json["\']', html, re.IGNORECASE))
|
||||||
|
data["schema_types"] = sorted(set(re.findall(r'"@type"\s*:\s*"([A-Za-z0-9_]+)"', html)))
|
||||||
|
m = (re.search(r'<link[^>]*rel=["\']canonical["\'][^>]*href=["\']([^"\']+)', html, re.IGNORECASE)
|
||||||
|
or re.search(r'<link[^>]*href=["\']([^"\']+)["\'][^>]*rel=["\']canonical["\']', html, re.IGNORECASE))
|
||||||
|
data["canonical"] = m.group(1) if m else None
|
||||||
|
data["has_og"] = bool(re.search(r'<meta[^>]*property=["\']og:', html, re.IGNORECASE))
|
||||||
|
|
||||||
|
# JSON-LD business hours -> same {day: value} shape as the other surfaces
|
||||||
try:
|
try:
|
||||||
md = _jina_markdown(website_url)
|
from canonical_baseline import jsonld_hours
|
||||||
except Exception as e:
|
ld_hours = jsonld_hours(html)
|
||||||
print(f" [FAIL] Website (direct + Jina): {e}")
|
except Exception:
|
||||||
return data if html else None
|
ld_hours = {}
|
||||||
if not data.get("title"):
|
if ld_hours:
|
||||||
m = re.search(r"^Title:\s*(.+)$", md, re.MULTILINE)
|
data["hours"] = ld_hours
|
||||||
if m:
|
|
||||||
data["title"] = m.group(1).strip()
|
|
||||||
data["phone_on_page"] = _phone_from_text(md)
|
|
||||||
data["renderer"] = "jina_reader"
|
|
||||||
return data
|
return data
|
||||||
|
|
||||||
|
except Exception as e:
|
||||||
|
print(f" [FAIL] Website: {e}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def merge_and_validate(gbp_data=None, apple_data=None, website_data=None, bing_data=None):
|
def merge_and_validate(gbp_data=None, apple_data=None, website_data=None, bing_data=None):
|
||||||
"""Merge data from all surfaces and flag mismatches."""
|
"""Merge data from all surfaces and flag mismatches."""
|
||||||
@@ -397,25 +371,5 @@ async def main():
|
|||||||
print(f" {key}: {json.dumps(val)}")
|
print(f" {key}: {json.dumps(val)}")
|
||||||
|
|
||||||
|
|
||||||
def _selftest():
|
|
||||||
"""Runnable check for phone extraction (no network)."""
|
|
||||||
assert _phone_from_text("Call us: (530) 350-9197 today") == "(530) 350-9197"
|
|
||||||
assert _phone_from_text("tel: (530) 350-9197") == "(530) 350-9197"
|
|
||||||
assert _phone_from_text("530-350-9197") == "530-350-9197"
|
|
||||||
# JS artifact: phone-like substring glued to other digits -> reject
|
|
||||||
assert _phone_from_text("var id=1234567890123;") is None
|
|
||||||
assert _phone_from_text("no numbers here") is None
|
|
||||||
# call-tracking placeholder (all-same digit) -> reject
|
|
||||||
assert _phone_from_text("Call (999) 999-9999 now") is None
|
|
||||||
# visible text wins over tel: href (audit semantics: report what's displayed;
|
|
||||||
# tel: vs visible discrepancies are audit FINDINGS, not extraction noise)
|
|
||||||
txt = 'Call: (916) 238-3838 <a href="tel:9165716448">call</a>'
|
|
||||||
assert _phone_from_text(txt) == "(916) 238-3838"
|
|
||||||
print("[SELFTEST] phone extraction: PASS")
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
if "--selftest" in sys.argv:
|
|
||||||
_selftest()
|
|
||||||
else:
|
|
||||||
asyncio.run(main())
|
asyncio.run(main())
|
||||||
|
|||||||
Reference in New Issue
Block a user