From cd40e652a0b818bd258328ce51e983b213d8e492 Mon Sep 17 00:00:00 2001 From: "Leonard (VeriPath Agent)" Date: Fri, 14 Aug 2026 21:02:37 +0000 Subject: [PATCH] build(auditing): add sibling scrapers for self-contained repo multi_scraper.py imports apple_scraper and bing_provider but they were not committed to implementation/auditing/. Add them so the repo runs without external file copies from the workdir. --- implementation/auditing/apple_scraper.py | 160 +++++++++++++++++++++++ implementation/auditing/bing_provider.py | 154 ++++++++++++++++++++++ 2 files changed, 314 insertions(+) create mode 100644 implementation/auditing/apple_scraper.py create mode 100644 implementation/auditing/bing_provider.py diff --git a/implementation/auditing/apple_scraper.py b/implementation/auditing/apple_scraper.py new file mode 100644 index 0000000..5157ed5 --- /dev/null +++ b/implementation/auditing/apple_scraper.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +"""Apple Maps scraper — captures /data/search JSON via headless Firefox.""" +import asyncio, json, re, urllib.parse +from playwright.async_api import async_playwright + +DAY_MAP = {'SUNDAY': 'sunday', 'MONDAY': 'monday', 'TUESDAY': 'tuesday', + 'WEDNESDAY': 'wednesday', 'THURSDAY': 'thursday', + 'FRIDAY': 'friday', 'SATURDAY': 'saturday'} + +def _hours_str(sec): + h, m = divmod(sec, 3600) + ampm = "AM" if h < 12 else "PM" + h12 = h % 12 or 12 + return f"{h12}:{m:02d} {ampm}" + + +async def scrape_apple(query: str) -> dict | None: + q = urllib.parse.quote(query) + + async with async_playwright() as pw: + browser = await pw.firefox.launch(headless=True) + context = await browser.new_context( + viewport={'width': 1920, 'height': 1080}, + user_agent='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) ' + 'AppleWebKit/605.1.15 ' + '(KHTML, like Gecko) Version/18.6 Safari/605.1.15', + locale='en-US', + timezone_id='America/Los_Angeles') + await context.add_init_script( + "Object.defineProperty(navigator,'webdriver',{get:()=>false});") + + page = await context.new_page() + search_json = None + + async def on_response(resp): + nonlocal search_json + url = resp.url + if '/data/search' in url and url.rstrip('/').endswith('/data/search'): + body = await resp.text() + if len(body) > 100: + search_json = body + + page.on('response', on_response) + await page.goto(f"https://maps.apple.com/?q={q}", + wait_until='domcontentloaded', timeout=30000) + await page.wait_for_timeout(15000) + await browser.close() + + if not search_json: + return None + + data = json.loads(search_json) + if data.get('status') != 'STATUS_SUCCESS': + return None + + # mapsResult[0].place.component[] holds typed blocks + place = data.get('mapsResult', [{}])[0].get('place', {}) + components = place.get('component', []) + + result = {} + for comp in components: + ctype = comp.get('type', '') + value = comp.get('value', [{}])[0] + + # NAME + PHONE + WEBSITE + CATEGORIES + if ctype == 'COMPONENT_TYPE_ENTITY': + ent = value.get('entity', {}) + name_list = ent.get('name', []) + if name_list: + result['name'] = name_list[0].get('stringValue', '') + cats = ent.get('localizedCategory', []) + if cats: + primary = cats[0].get('localizedName', [{}])[0] + result['category'] = primary.get('stringValue', '') + phone = ent.get('phoneNumberFormatted', '') + if phone: + result['phone'] = phone + url = ent.get('url', '') + if url: + result['website'] = url + + # ADDRESS + if ctype == 'COMPONENT_TYPE_ADDRESS_OBJECT': + addr = value.get('addressObject', {}) + lines = addr.get('formattedAddressLines', []) + # skip "United States" + lines = [l for l in lines if l != 'United States'] + result['address'] = ', '.join(lines) if lines else addr.get('shortAddress', '') + + # HOURS + if ctype == 'COMPONENT_TYPE_BUSINESS_HOURS': + hours_data = value.get('businessHours', {}) + weekly = hours_data.get('weeklyHours', []) + hours_dict = {} + for entry in weekly: + days = entry.get('day', []) + for tr in entry.get('timeRange', []): + for day in days: + d = DAY_MAP.get(day, day.lower()) + hours_dict[d] = f"{_hours_str(tr['from'])}-{_hours_str(tr['to'])}" + if hours_dict: + result['hours'] = hours_dict + + # RATING + if ctype == 'COMPONENT_TYPE_RATING': + for rv in comp.get('value', []): + r = rv.get('rating', {}) + if r.get('ratingType') == 'USER_RATING': + result['rating'] = r.get('score') + raw = r.get('numRatingsUsedForScore') or r.get('ratingsFormatted') + if raw: + result['reviews'] = int(re.sub(r'\D', '', str(raw))) + + # COORDINATES + if ctype == 'COMPONENT_TYPE_PLACE_INFO': + center = value.get('placeInfo', {}).get('center', {}) + if center: + result['coordinates'] = { + 'lat': center.get('lat'), + 'lon': center.get('lng'), + } + + # PRICE + if ctype == 'COMPONENT_TYPE_RESULT_SNIPPET': + snippet = value.get('resultSnippet', {}) + price = snippet.get('priceRange', {}) + if price.get('ratingType') == 'PRICE_RANGE': + score = price.get('score', 0) + result['price_level'] = '$' * score + + # REVIEWS (COMPONENT_TYPE_REVIEW) + for comp in components: + if comp.get('type') == 'COMPONENT_TYPE_REVIEW': + reviews = [] + for rv in comp.get('value', [])[:5]: # ponytail: top 5, paginate if more needed + review = rv.get('review', {}) + snippet_parts = review.get('snippet', []) + text = ''.join( + t.get('stringValue', '') + for t in snippet_parts if isinstance(t, dict) + ) + if text: + rating_info = review.get('rating', {}) + reviews.append({ + 'text': text.strip(), + 'rating': rating_info.get('score'), + }) + if reviews: + result['reviews_sample'] = reviews + + # Clean None/empty + result = {k: v for k, v in result.items() if v not in (None, {}, [])} + return result if result.get('name') else None + + +if __name__ == '__main__': + import sys + q = ' '.join(sys.argv[1:]) or 'Phoenix Salon + Spa Cameron Park, CA' + res = asyncio.run(scrape_apple(q)) + print(json.dumps(res, indent=2) if res else 'No data') diff --git a/implementation/auditing/bing_provider.py b/implementation/auditing/bing_provider.py new file mode 100644 index 0000000..de9b6f3 --- /dev/null +++ b/implementation/auditing/bing_provider.py @@ -0,0 +1,154 @@ +#!/usr/bin/env python3 +""" +Bing Places scraper via Bing Web Search (NOT Bing Maps). + +Bing Maps blocks headless (WebGL wall), but Bing *web search* renders +full local business data in the DOM: name, address, phone, hours, +rating, reviews, categories, description. + +Data is Yelp-sourced (Bing attributes to Yelp). Same as Apple Maps. +""" +import asyncio +import json +import re +import sys +from playwright.async_api import async_playwright + + +async def scrape_bing(query: str) -> dict | None: + """Scrape Bing Places data from Bing web search results. + + Args: + query: "Business Name City, ST" + + Returns: + dict with name, address, phone, website, rating, reviews, + hours, category, description, reviews_sample, or None on failure. + """ + q = re.sub(r'\s+', ' ', query).strip() + url = f"https://www.bing.com/search?q={q.replace(' ', '+')}" + + async with async_playwright() as pw: + browser = await pw.firefox.launch(headless=True) + context = await browser.new_context( + viewport={"width": 1920, "height": 1080}, + user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.6 Safari/605.1.15", + locale="en-US", + timezone_id="America/Los_Angeles", + ) + await context.add_init_script("Object.defineProperty(navigator,'webdriver',{get:()=>false});") + page = await context.new_page() + try: + await page.goto(url, wait_until="domcontentloaded", timeout=30000) + await page.wait_for_timeout(8000) + body_text = await page.locator("body").text_content() + except Exception: + await browser.close() + return None + await browser.close() + + # Find business section by first word of query + first_word = q.split()[0] + biz_idx = body_text.find(first_word) + if biz_idx < 0: + return None + + # Grab 3000 chars from business mention — this is one long blob + section = body_text[biz_idx : biz_idx + 3000] + result = {} + + # NAME: from first word up to the first URL + name_m = re.match(r"([A-Za-z\s&+,.\'-]{3,60}?)\s*https?://", section) + if not name_m: + name_m = re.match(r"([A-Za-z\s&+,.\'-]{3,60}?)\s*Share", section) + if not name_m: + name_m = re.match(r"([A-Za-z\s&+,.\'-]{3,60}?)\s*\d", section) # before rating + if name_m: + result["name"] = name_m.group(1).strip() + else: + return None + + # WEBSITE + url_m = re.search(r"(https?://[a-zA-Z0-9._/-]+?)(?:\s|$)", section) + if url_m: + result["website"] = url_m.group(1).rstrip("/") + + # RATING + REVIEWS: "5/5 (65 reviews)" + rating_m = re.search(r"(\d+(?:\.\d+)?)\s*/\s*5.*?\(\s*(\d+)\s*reviews?\)", section) + if rating_m: + result["rating"] = float(rating_m.group(1)) + result["reviews"] = int(rating_m.group(2)) + + # CATEGORY: "reviews) · Beauty & spa in Cameron Park" + cat_m = re.search(r"reviews?\)\s*·\s*([A-Za-z\s&]+?)\s*in\s+", section) + if cat_m: + result["category"] = cat_m.group(1).strip() + + # ADDRESS: "3460 Robin Ln Ste 4, Cameron Park, CA 95682" + addr_m = re.search(r"(\d+\s+[A-Za-z\s.,#SteSuite]+?\d{5})", section) + if addr_m: + result["address"] = addr_m.group(1).strip().rstrip(",.") + + # PHONE: "(530) 350-9197" + phone_m = re.search(r"(\(\d{3}\)\s*\d{3}-\d{4})", section) + if not phone_m: + phone_m = re.search(r"(\d{3}-\d{3}-\d{4})", section) + if phone_m: + result["phone"] = phone_m.group(1) + + # HOURS: "Thursday9 AM - 7 PMFriday9 AM - 6 PM..." + # Pattern: DayName + hours (with no space between day and time) + hours = {} + days_order = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday"] + for day in days_order: + day_m = re.search( + rf"{day}(\d+(?::\d+)?\s*[AP]M?\s*-\s*\d+(?::\d+)?\s*[AP]M?|Closed|opens\s+\w+\s+\d+[AP]M?)", + section, + re.I, + ) + if day_m: + hours[day.lower()] = day_m.group(1).strip() + if hours: + result["hours"] = hours + + # DESCRIPTION: "AboutPhoenix Salon + Spa offers..." + about_idx = section.find("About") + if about_idx >= 0: + about_snip = section[about_idx:about_idx + 600] + # Skip "About" then business name + name = result.get("name", "") + name_idx = about_snip.find(name) + if name_idx >= 0: + desc_text = about_snip[name_idx + len(name):].strip() + # Grab first sentence + end = desc_text.find(".") + if end > 0: + result["description"] = desc_text[:end + 1].strip() + + # REVIEWS: "Apr 21, 2026I have been a customer...Learn more ·" + reviews = [] + for rm in re.finditer( + r"((?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)\s+\d{1,2},\s+\d{4})(.*?)Learn more", + section, + re.DOTALL, + ): + text = rm.group(2).strip().replace("✕", "") + text = re.sub(r'\s+', ' ', text)[:200] + if len(text) < 20: + continue + reviews.append({"text": text, "date": rm.group(1)}) + if len(reviews) >= 3: + break + if reviews: + result["reviews_sample"] = reviews + + return result if result.get("name") else None + + +if __name__ == "__main__": + query = sys.argv[1] if len(sys.argv) > 1 else "Phoenix Salon + Spa Cameron Park, CA" + result = asyncio.run(scrape_bing(query)) + if result: + print(json.dumps(result, indent=2, ensure_ascii=False)) + else: + print("No data extracted")