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.
This commit is contained in:
Leonard (VeriPath Agent)
@@ -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')
|
||||
Reference in New Issue
Block a user