veripath/implementation/auditing/apple_scraper.py

#!/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')