fix(auditing): reject bad GBP matches + strip phone glyph
- Add _looks_like_match() gate in scrape_google(): rejects search-page titles and names sharing no >=4-char token with the query, so the scraper no longer ingests a wrong entity (e.g. name='Hours', or 'RK Mechanical' for an 'Aire Serv' search). - Strip BMP PUA glyphs (U+E000-U+F8FF, e.g. the \ue0b0 map-pin) from GBP address + phone. Old regex only covered the Supplementary PUA plane (\U000E0000-\U000EFFFF) so the glyph leaked into the contract and broke downstream phone normalization. - Also: Apple Maps fallback path when GBP is rejected, and makedirs for the optional 3rd-arg output dir.
This commit is contained in:
Leonard (VeriPath Agent)
@@ -5,7 +5,7 @@ Extracts GBP + Apple Maps + Website, merges into verified JSON contract.
|
||||
|
||||
Working surfaces:
|
||||
- Google Business Profile (via noworneverev/google-maps-scraper)
|
||||
- Apple Maps (headless Firefox + /data/search JSON)
|
||||
- Apple Maps (headless Firefox → /data/search JSON)
|
||||
- Website (direct HTTP fetch)
|
||||
|
||||
Bot-walled (headless detection, no free bypass):
|
||||
@@ -42,7 +42,7 @@ async def geocode(query):
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
m = re.search(r'"center":\{"lat":(-?\d+\.\d+),"lng":(-?\d+\.\d+)\}', html)
|
||||
m = re.search(r'"center":\{"lat":(-?\d+\.\d+),"lng":(-?\d+\.\d+)}', html)
|
||||
if m:
|
||||
return float(m.group(1)), float(m.group(2))
|
||||
|
||||
@@ -63,15 +63,39 @@ def build_gmaps_url(query, coords):
|
||||
return f"https://www.google.com/maps/search/{encoded}/data=!3m1!4b1"
|
||||
|
||||
|
||||
async def scrape_google(query, coords):
|
||||
"""Scrape Google Business Profile data."""
|
||||
# Known search-page titles (not real places) — same set the lib rejects
|
||||
_SEARCH_PAGE_TITLES = {"results", "Results", "検索結果", "搜尋結果", "搜索结果"}
|
||||
|
||||
|
||||
def _looks_like_match(place_name, business):
|
||||
"""Reject search-page titles and names that share no significant token with the query."""
|
||||
if not place_name:
|
||||
return False
|
||||
name = place_name.strip()
|
||||
if name.lower() in _SEARCH_PAGE_TITLES:
|
||||
return False
|
||||
# Significant tokens = words >=4 chars from the business query
|
||||
biz_tokens = {t.lower() for t in re.findall(r'[A-Za-z]{4,}', business)}
|
||||
name_tokens = {t.lower() for t in re.findall(r'[A-Za-z]{4,}', name)}
|
||||
if not biz_tokens:
|
||||
return True
|
||||
# Require at least one shared significant token (e.g. "Crystal", "Plumbing")
|
||||
return bool(biz_tokens & name_tokens)
|
||||
|
||||
|
||||
async def scrape_google(query, coords, business=""):
|
||||
"""Scrape Google Business Profile data, validating the matched result."""
|
||||
url = build_gmaps_url(query, coords)
|
||||
try:
|
||||
async with GoogleMapsScraper() as scraper:
|
||||
result = await scraper.scrape(url)
|
||||
if result and result.place:
|
||||
p = result.place
|
||||
# Hours: ["Thursday9 AM٥ PM", "SundayClosed", ...]
|
||||
# Validation gate: reject wrong-entity matches before trusting
|
||||
if not _looks_like_match(p.name, business or query):
|
||||
print(f" [REJECT] GBP matched wrong entity: '{p.name}' (expected ~'{business}')")
|
||||
return None
|
||||
# Hours: ["Thursday9 AM–6 PM", "SundayClosed", ...]
|
||||
hours_dict = {}
|
||||
if p.hours:
|
||||
for h in p.hours:
|
||||
@@ -83,9 +107,10 @@ async def scrape_google(query, coords):
|
||||
hours_dict[day] = time_str
|
||||
|
||||
return {
|
||||
# Strip PUA glyphs (map-pin \ue0b0 etc.) from address + phone — BMP PUA (U+E000–U+F8FF) breaks downstream normalize
|
||||
"name": p.name,
|
||||
"address": p.address,
|
||||
"phone": p.phone,
|
||||
"address": re.sub(r'[\ue000-\uf8ff]', '', p.address or '').strip(),
|
||||
"phone": re.sub(r'[\ue000-\uf8ff]', '', p.phone or '').strip() if p.phone else p.phone,
|
||||
"website": p.website,
|
||||
"rating": p.rating,
|
||||
"reviews": p.review_count,
|
||||
@@ -123,14 +148,14 @@ def fetch_website_data(website_url):
|
||||
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:
|
||||
m = re.search(r'<meta[^.]*content=["\']([^"\']+["\'][^>]*name=["\']description["\']', html, re.IGNORECASE)
|
||||
m = re.search(r'<meta[^>]*content=["\']([^"\']+)["\'][^>]*name=["\']description["\']', html, re.IGNORECASE)
|
||||
if m:
|
||||
data["description"] = m.group(1).strip()
|
||||
|
||||
# Extract phone from page
|
||||
phone_match = re.search(r'(d{3}d{3}d{4})', html)
|
||||
phone_match = re.search(r'(\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{4})', html)
|
||||
if phone_match:
|
||||
data["phone_on_page"] = phone_match.group(1)
|
||||
|
||||
@@ -209,11 +234,12 @@ def merge_and_validate(gbp_data=None, apple_data=None, website_data=None, bing_d
|
||||
if hour_mismatches:
|
||||
contract["audit"]["verification"]["hours_mismatch"] = hour_mismatches
|
||||
|
||||
# Use GBP as primary source
|
||||
if gbp_data:
|
||||
# Use GBP as primary source — but only if it has real data
|
||||
gbp_usable = gbp_data and gbp_data.get("name") and gbp_data.get("name") not in ("Hours", "Open", "Closed", "") and gbp_data.get("address")
|
||||
if gbp_usable:
|
||||
contract["name"] = gbp_data.get("name")
|
||||
contract["address"] = re.sub(r'[\U000E0000-\U000EFFFF]', '', gbp_data.get("address", "")).strip()
|
||||
contract["phone"] = re.sub(r'[\U000E0000-\U000EFFFF]', '', gbp_data.get("phone", "")).strip()
|
||||
contract["address"] = re.sub(r'[\ue000-\uf8ff\U000E0000-\U000EFFFF]', '', gbp_data.get("address") or "").strip()
|
||||
contract["phone"] = re.sub(r'[\ue000-\uf8ff\U000E0000-\U000EFFFF]', '', gbp_data.get("phone") or "").strip()
|
||||
contract["website"] = gbp_data.get("website")
|
||||
contract["rating"] = gbp_data.get("rating")
|
||||
contract["reviews"] = gbp_data.get("reviews")
|
||||
@@ -231,6 +257,16 @@ def merge_and_validate(gbp_data=None, apple_data=None, website_data=None, bing_d
|
||||
"permanently_closed": gbp_data.get("permanently_closed"),
|
||||
"temporarily_closed": gbp_data.get("temporarily_closed"),
|
||||
}
|
||||
elif apple_data:
|
||||
# ponytail: fallback to Apple if GBP returned None fields
|
||||
contract["name"] = apple_data.get("name") or "Unknown"
|
||||
contract["address"] = re.sub(r'[\ue000-\uf8ff\U000E0000-\U000EFFFF]', '', apple_data.get("address") or "").strip()
|
||||
contract["phone"] = re.sub(r'[\ue000-\uf8ff\U000E0000-\U000EFFFF]', '', apple_data.get("phone") or "").strip()
|
||||
contract["website"] = apple_data.get("website") or ""
|
||||
contract["rating"] = apple_data.get("rating")
|
||||
contract["reviews"] = apple_data.get("reviews")
|
||||
contract["hours"] = apple_data.get("hours")
|
||||
contract["primary_source"] = "apple_maps"
|
||||
|
||||
return contract
|
||||
|
||||
@@ -255,7 +291,7 @@ async def main():
|
||||
|
||||
# Step 2: Scrape GBP (primary)
|
||||
print("\nScraping Google Business Profile...")
|
||||
gbp_data = await scrape_google(query, coords)
|
||||
gbp_data = await scrape_google(query, coords, business=business)
|
||||
if gbp_data:
|
||||
print(f" [OK] Google: {gbp_data['name']} ({gbp_data['rating']} ★, {gbp_data['reviews']} reviews)")
|
||||
print(f" Hours: {gbp_data.get('hours', {})}")
|
||||
@@ -300,7 +336,10 @@ async def main():
|
||||
safe_name = re.sub(r"[^\w\s-]", "", business).strip().replace(" ", "_").lower()
|
||||
date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||||
filename = f"{safe_name}_multi_surface_{date_str}.json"
|
||||
filepath = os.path.join(BASE_DIR, filename)
|
||||
# Accept optional output path as 3rd argument; default to script dir
|
||||
output_dir = sys.argv[3] if len(sys.argv) > 3 else BASE_DIR
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
filepath = os.path.join(output_dir, filename)
|
||||
|
||||
with open(filepath, "w") as f:
|
||||
json.dump(contract, f, indent=2, ensure_ascii=False)
|
||||
|
||||
Reference in New Issue
Block a user