scraper.py

"""
Core Scraper Engine for Google Maps
===================================
Playwright-based scraper optimized for Category 1 (list-view) data extraction.
Includes bandwidth optimization for 1GB free proxy tier.
"""

import re
import time
import logging
import hashlib
from datetime import datetime
from typing import Optional, List, Dict, Any, Callable
from urllib.parse import urlparse, parse_qs

from playwright.sync_api import (
    sync_playwright,
    Page,
    Browser,
    BrowserContext,
    Route,
    TimeoutError as PlaywrightTimeout
)

from pathlib import Path

from db import DatabaseManager

logger = logging.getLogger(__name__)

PROFILE_DIR = Path(__file__).parent / ".gmap_profile"


class GMapScraper:
    """
    Google Maps scraper for Category 1 (list-view metadata) extraction.

    Features:
    - Bandwidth optimization (blocks images/fonts/CSS)
    - Coordinate extraction from URLs
    - Google ID parsing (place_id, data_id, cid)
    - Robust error handling per-listing
    """

    def __init__(self, config: dict, db: DatabaseManager):
        self.config = config
        self.db = db
        self.browser: Optional[Browser] = None
        self.context: Optional[BrowserContext] = None
        self.page: Optional[Page] = None

        # Callbacks for progress reporting
        self.on_city_start: Optional[Callable] = None
        self.on_city_complete: Optional[Callable] = None
        self.on_city_error: Optional[Callable] = None
        self.on_listing_found: Optional[Callable] = None
        self.on_business_inserted: Optional[Callable] = None  # Called with (business_data, is_new)
        self.on_bandwidth_used: Optional[Callable] = None  # Called with (bytes_used)

        # Bandwidth tracking
        self._bandwidth_bytes = 0

        # Cross-scrape dedup: set of hashes to exclude
        self.excluded_hashes: set = set()

    def _block_media(self, route: Route):
        """
        Bandwidth optimization: abort non-essential resource requests.
        Essential for staying within 1GB free proxy limit.
        Also tracks bandwidth usage.
        """
        resource_type = route.request.resource_type
        blocked_types = []

        if self.config.get('optimization', {}).get('block_images', True):
            blocked_types.append('image')
        if self.config.get('optimization', {}).get('block_fonts', True):
            blocked_types.append('font')
        if self.config.get('optimization', {}).get('block_stylesheets', True):
            blocked_types.append('stylesheet')
        if self.config.get('optimization', {}).get('block_media', True):
            blocked_types.append('media')

        if resource_type in blocked_types:
            route.abort()
        else:
            route.continue_()

    def _track_response(self, response):
        """Track bandwidth from responses."""
        try:
            # Get response body size (if available)
            headers = response.headers
            content_length = headers.get('content-length', '0')
            bytes_used = int(content_length) if content_length.isdigit() else 0

            # Estimate from URL if no content-length
            if bytes_used == 0:
                # Rough estimate: 5KB average per request
                bytes_used = 5000

            self._bandwidth_bytes += bytes_used

            if self.on_bandwidth_used:
                self.on_bandwidth_used(bytes_used)
        except:
            pass

    def get_bandwidth_used(self) -> int:
        """Get total bandwidth used in bytes."""
        return self._bandwidth_bytes

    def reset_bandwidth(self):
        """Reset bandwidth counter."""
        self._bandwidth_bytes = 0

    def _parse_coordinates(self, url: str) -> tuple:
        """
        Extract latitude/longitude from Google Maps URL.

        Primary pattern (data URLs): !3d{lat}!4d{lng}
        Example: !3d32.9545907!4d-96.8375646 -> (32.9545907, -96.8375646)

        Fallback pattern: @{lat},{lng},{zoom}z
        """
        if not url:
            return None, None

        # Primary: data URL format used in place URLs with rclk=1
        match = re.search(r'!3d(-?\d+\.\d+)!4d(-?\d+\.\d+)', url)
        if match:
            return float(match.group(1)), float(match.group(2))

        # Fallback: classic @lat,lng format
        match = re.search(r'@(-?\d+\.\d+),(-?\d+\.\d+)', url)
        if match:
            return float(match.group(1)), float(match.group(2))
        return None, None

    def _parse_google_ids(self, url: str) -> dict:
        """
        Extract Google identifiers from Maps URL.

        Returns:
            {
                'place_id': 'ChIJ...',         # Permanent Google Place ID (!19s)
                'data_id': '0x...:0x...',       # Hex compound identifier (!1s)
                'cid': '123...',                # Customer ID (decimal, from cid= param)
                'google_feature_id': '/g/...'   # Feature ID (!16s)
            }
        """
        from urllib.parse import unquote
        ids = {'place_id': None, 'data_id': None, 'cid': None, 'google_feature_id': None}

        if not url:
            return ids

        # Place ID: !19sChIJ... format in data URLs
        place_match = re.search(r'!19s(ChIJ[A-Za-z0-9_\-]+)', url)
        if place_match:
            ids['place_id'] = place_match.group(1)

        # Data ID: Compound hex format 0x...:0x...
        data_match = re.search(r'(0x[a-f0-9]+:0x[a-f0-9]+)', url)
        if data_match:
            ids['data_id'] = data_match.group(1)

        # CID: Usually in query params or embedded in URL
        cid_match = re.search(r'cid[=:](\d+)', url)
        if cid_match:
            ids['cid'] = cid_match.group(1)

        # Feature ID: !16s%2Fg%2F... (URL-encoded /g/XXXXXXXXX)
        feature_match = re.search(r'!16s(%2[Ff]g%2[Ff][^!?&]+)', url)
        if feature_match:
            ids['google_feature_id'] = unquote(feature_match.group(1))

        return ids

    def _parse_rating_text(self, text: str) -> tuple:
        """
        Parse rating and review count from text.

        Formats handled:
        - "4.8(154)" or "4.9(3,986) · $20–30" (visible browser mode)
        - "4.8" or "4.9" (headless browser mode - just rating, no review count)

        Returns: (rating: float, review_count: int or None, raw_text: str)
        """
        if not text:
            return None, None, None

        text = text.strip()

        # Pattern 1: "X.X(NUMBER)" - handles commas in review count like "4.9(3,986)"
        match = re.search(r'(\d+\.?\d*)\s*\(([\d,]+)', text)
        if match:
            rating = float(match.group(1))
            # Remove commas from review count before converting to int
            count_str = match.group(2).replace(',', '')
            count = int(count_str)
            return rating, count, text

        # Pattern 2: Standalone rating "4.8" or "4.9" (headless mode)
        # Must be a valid rating (1.0-5.0 range) and standalone or at start of line
        rating_match = re.match(r'^(\d+\.\d)$', text)
        if rating_match:
            rating = float(rating_match.group(1))
            # Valid Google ratings are 1.0-5.0
            if 1.0 <= rating <= 5.0:
                return rating, None, text

        return None, None, None

    def _parse_phone(self, text_lines: List[str]) -> Optional[str]:
        """
        Extract phone number from listing text content.
        Looks for patterns like: +1 XXX-XXX-XXXX, (XXX) XXX-XXXX, etc.
        """
        phone_patterns = [
            r'\+1[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{4}',  # +1 format
            r'\(\d{3}\)\s?\d{3}[\s\-]?\d{4}',                   # (XXX) format
            r'\d{3}[\s\-]\d{3}[\s\-]\d{4}',                     # XXX-XXX-XXXX
        ]

        for line in text_lines:
            for pattern in phone_patterns:
                match = re.search(pattern, line)
                if match:
                    return match.group(0)

        return None

    def extract_category_one(self, page: Page, search_query: str, search_city: str) -> List[Dict[str, Any]]:
        """
        Extract Category 1 metadata from all visible listings.

        Category 1 = Data visible in the list/grid view without clicking.
        This is the fastest extraction method and uses minimal bandwidth.
        """
        results = []

        try:
            # Get all listing articles in the feed
            listings = page.locator('div[role="article"]').all()
            logger.debug(f"Found {len(listings)} potential listings")

            for rank, listing in enumerate(listings, start=1):
                try:
                    data = self._extract_single_listing(listing, rank, search_query, search_city)
                    if data and data.get('name'):
                        results.append(data)
                        if self.on_listing_found:
                            self.on_listing_found(data)
                except Exception as e:
                    logger.debug(f"Failed to extract listing {rank}: {e}")
                    continue

        except Exception as e:
            logger.error(f"Error extracting listings: {e}")

        return results

    def _extract_single_listing(
        self,
        listing,
        rank: int,
        search_query: str,
        search_city: str
    ) -> Optional[Dict[str, Any]]:
        """Extract data from a single listing element."""
        data = {
            'rank_on_page': rank,
            'search_query': search_query,
            'search_city': search_city,
            'scraped_at': datetime.now().isoformat()
        }

        # 1. Get the main link (contains name and URL)
        link_locator = listing.locator('a[href*="google.com/maps"]')
        if link_locator.count() == 0:
            # Try alternate selector
            link_locator = listing.locator('a[href*="maps.google"]')

        if link_locator.count() > 0:
            url = link_locator.first.get_attribute('href')
            name = link_locator.first.get_attribute('aria-label')

            data['google_url'] = url
            data['name'] = name

            # Parse coordinates from URL
            lat, lng = self._parse_coordinates(url)
            data['latitude'] = lat
            data['longitude'] = lng

            # Parse Google IDs from URL
            ids = self._parse_google_ids(url)
            data.update(ids)
        else:
            return None  # No link = not a valid listing

        # Derive state and city_name from search_city ("Dallas, TX" -> "TX", "Dallas")
        if search_city and ',' in search_city:
            parts = [p.strip() for p in search_city.split(',')]
            data['city_name'] = parts[0]
            data['state'] = parts[-1]
        else:
            data['city_name'] = search_city
            data['state'] = None

        # 2. Parse text content for phone, rating, address, price
        try:
            text_content = listing.inner_text()
            text_lines = [line.strip() for line in text_content.split('\n') if line.strip()]

            # Phone number
            data['phone_number'] = self._parse_phone(text_lines)

            # Rating and review count
            # Format varies:
            #   "4.8(255)" — combined (no CSS blocking)
            #   "4.8" then "(55)" on next line — split (CSS blocked)
            #   "4.8" alone — headless/limited view
            for idx, line in enumerate(text_lines):
                rating, count, raw = self._parse_rating_text(line)
                if rating is not None:
                    data['rating'] = rating
                    data['review_count'] = count
                    data['rating_text_raw'] = raw
                    # If count is missing, check next line for "(N)" pattern
                    if count is None and idx + 1 < len(text_lines):
                        next_line = text_lines[idx + 1].strip()
                        count_match = re.match(r'^\(([\d,]+)\)$', next_line)
                        if count_match:
                            data['review_count'] = int(count_match.group(1).replace(',', ''))
                            data['rating_text_raw'] = f"{line}{next_line}"
                    break

            # Review count fallback: aria-label on the stars/rating element
            # Google renders: aria-label="Rated 4.9 out of 5, 255 reviews"
            if not data.get('review_count'):
                try:
                    stars_el = listing.locator('[aria-label*="out of 5"], [aria-label*="stars"]')
                    if stars_el.count() > 0:
                        aria = stars_el.first.get_attribute('aria-label') or ''
                        review_match = re.search(r'([\d,]+)\s*review', aria, re.IGNORECASE)
                        if review_match:
                            data['review_count'] = int(review_match.group(1).replace(',', ''))
                        if not data.get('rating'):
                            rating_match = re.search(r'(\d+\.?\d*)\s*(?:out of|stars)', aria, re.IGNORECASE)
                            if rating_match:
                                data['rating'] = float(rating_match.group(1))
                except Exception:
                    pass

            # Category, Address, and Price parsing
            # Google Maps format: "Category · [icon] · Address" on one line
            # May also include price level: "Smoke shop · $ · 123 Main St"
            name_lower = data.get('name', '').lower() if data.get('name') else ''

            for line in text_lines:
                # Skip lines that are clearly not category/address
                if '(' in line and ')' in line:  # Rating line
                    continue
                if line.lower() == name_lower:  # Skip the name itself
                    continue
                if any(skip in line.lower() for skip in ['open', 'closed', '+1', 'www.', 'http']):
                    continue

                # Look for lines with · separator (category · address pattern)
                if '·' in line:
                    parts = [p.strip() for p in line.split('·')]
                    # Filter out empty parts and icon characters (like \ue934)
                    clean_parts = []
                    price_parts = []
                    for p in parts:
                        # Skip empty or single Unicode icon chars
                        if not p or (len(p) == 1 and ord(p[0]) > 0xe000):
                            continue
                        # Capture price indicators separately ("$", "$$", "$$$", "$$$$")
                        if re.match(r'^\$+$', p.strip()):
                            price_parts.append(p.strip())
                            continue
                        # Skip price ranges like "$20–30"
                        if p.startswith('$') and len(p) > 2:
                            continue
                        clean_parts.append(p)

                    if price_parts and not data.get('price_level'):
                        data['price_level'] = price_parts[0]

                    if len(clean_parts) >= 1:
                        # First clean part is usually the category
                        if not data.get('category'):
                            data['category'] = clean_parts[0]
                        # Look for address in remaining parts
                        address_patterns = ['st', 'ave', 'blvd', 'rd', 'dr', 'ln', 'way', 'hwy', 'pkwy', 'suite', 'ste']
                        for part in clean_parts[1:]:
                            part_lower = part.lower()
                            if any(p in part_lower for p in address_patterns) or re.search(r'\d+', part):
                                data['address_text'] = part
                                break
                    continue

            # Fallback: look for address in any line if not found
            if not data.get('address_text'):
                address_patterns = ['st', 'ave', 'blvd', 'rd', 'dr', 'ln', 'way', 'hwy', 'pkwy']
                for line in text_lines:
                    line_lower = line.lower()
                    if any(pattern in line_lower for pattern in address_patterns):
                        if line.lower() != name_lower:
                            data['address_text'] = line
                            break

            # Extract zip code from address text
            if data.get('address_text'):
                zip_match = re.search(r'\b(\d{5})(?:-\d{4})?\b', data['address_text'])
                if zip_match:
                    data['zip_code'] = zip_match.group(1)

            # Open/closed status
            for line in text_lines:
                line_lower = line.lower()
                if 'open' in line_lower or 'closed' in line_lower:
                    data['open_status'] = line
                    # Check for hours snippet (e.g., "Closes 5 PM")
                    if 'closes' in line_lower or 'opens' in line_lower:
                        data['hours_snippet'] = line
                    break

            # Service options (e.g. "Delivery · In-store pickup · No-contact delivery")
            service_keywords = ['delivery', 'dine-in', 'takeout', 'takeaway', 'pickup', 'no-contact', 'curbside']
            for line in text_lines:
                line_lower = line.lower()
                if any(kw in line_lower for kw in service_keywords):
                    data['service_options'] = line
                    break

            # Description snippet: any remaining line that doesn't fit other categories
            # e.g. "Hookahs, pipes, vaporizers & accessories" or "Women-led"
            skip_patterns = ['open', 'closed', 'closes', 'opens', '+1', '(']
            cat_lower = (data.get('category', '') or '').lower()
            addr_lower = (data.get('address_text', '') or '').lower()
            name_lower_set = {
                data.get('name', '').lower(),
                cat_lower,
                addr_lower,
                (data.get('open_status', '') or '').lower(),
                (data.get('hours_snippet', '') or '').lower(),
                (data.get('service_options', '') or '').lower(),
                (data.get('phone_number', '') or '').lower(),
            }
            for line in text_lines:
                line_lower = line.lower()
                if line_lower in name_lower_set:
                    continue
                if any(p in line_lower for p in skip_patterns):
                    continue
                if re.match(r'^\d+\.?\d*$', line):  # pure number (rating)
                    continue
                # Skip the combined "Category · icon · Address" line
                if cat_lower and cat_lower in line_lower:
                    continue
                if addr_lower and addr_lower in line_lower:
                    continue
                # It's an unaccounted-for text line — likely a description/highlight
                data['description_snippet'] = line
                break

        except Exception as e:
            logger.debug(f"Error parsing text content: {e}")

        # 3. Accessibility tags (e.g. "Wheelchair accessible entrance")
        try:
            accessibility_els = listing.locator('[aria-label*="accessible" i], [aria-label*="wheelchair" i]').all()
            if accessibility_els:
                tags = []
                for el in accessibility_els:
                    lbl = el.get_attribute('aria-label') or ''
                    if lbl and lbl not in tags:
                        tags.append(lbl)
                if tags:
                    data['accessibility'] = '; '.join(tags)
        except Exception:
            pass

        # 4. Website URL — list view rarely shows a website button, but try common selectors
        try:
            for selector in [
                'a[data-value="Website"]',
                'a[jsaction*="website"]',
                'a[aria-label*="Website"]',
                'a[aria-label*="website"]',
            ]:
                btn = listing.locator(selector)
                if btn.count() > 0:
                    href = btn.first.get_attribute('href')
                    if href and 'google.com' not in href:
                        data['website_url'] = href
                        break
        except Exception:
            pass

        # 4. Thumbnail URL
        try:
            img = listing.locator('img').first
            if img.count() > 0:
                data['thumbnail_url'] = img.get_attribute('src')
        except:
            pass

        return data

    def extract_deep_details(self, listing, timeout_ms: int = 5000) -> Dict[str, Any]:
        """
        Extract detailed info by clicking into a listing's detail panel.

        This extracts Category 2 data (phone, website, full address, hours)
        that is only visible in the detail view.

        Args:
            listing: The listing element to click
            timeout_ms: Timeout for waiting for detail panel

        Returns:
            Dict with phone_number, website_url, full_address, hours, and other details
        """
        details = {}

        try:
            # Click the listing to open detail panel
            listing.scroll_into_view_if_needed()
            time.sleep(0.2)
            listing.click()

            # Wait for detail panel to load (look for action buttons)
            try:
                self.page.wait_for_selector(
                    'button[data-item-id^="phone"], a[data-item-id="authority"], div[role="main"]',
                    timeout=timeout_ms
                )
                time.sleep(0.5)  # Let content settle
            except:
                return details

            # Extract phone number from detail panel
            # Note: data-item-id format changed from "phone" to "phone:tel:+1234567890"
            try:
                phone_btn = self.page.locator('button[data-item-id^="phone"]')
                if phone_btn.count() > 0:
                    # Phone is in aria-label like "Phone: (214) 555-1234"
                    aria = phone_btn.first.get_attribute('aria-label')
                    if aria and 'phone' in aria.lower():
                        # Extract phone from "Phone: (XXX) XXX-XXXX"
                        phone_match = re.search(
                            r'[\+]?[\d\s\-\(\)]{10,}',
                            aria
                        )
                        if phone_match:
                            details['phone_number'] = phone_match.group(0).strip()
            except Exception as e:
                logger.debug(f"Error extracting phone: {e}")

            # Extract website URL
            try:
                website_link = self.page.locator('a[data-item-id="authority"]')
                if website_link.count() > 0:
                    href = website_link.first.get_attribute('href')
                    if href:
                        details['website_url'] = href
            except Exception as e:
                logger.debug(f"Error extracting website: {e}")

            # Extract full address
            try:
                address_btn = self.page.locator('button[data-item-id="address"]')
                if address_btn.count() > 0:
                    aria = address_btn.first.get_attribute('aria-label')
                    if aria and 'address' in aria.lower():
                        # Remove "Address: " prefix
                        details['full_address'] = aria.replace('Address:', '').strip()
            except Exception as e:
                logger.debug(f"Error extracting address: {e}")

            # Extract hours
            try:
                hours_btn = self.page.locator('button[data-item-id*="oh"]')
                if hours_btn.count() > 0:
                    aria = hours_btn.first.get_attribute('aria-label')
                    if aria:
                        details['hours_text'] = aria
            except Exception as e:
                logger.debug(f"Error extracting hours: {e}")

            # Extract review count if not already captured
            try:
                review_btn = self.page.locator('button[jsaction*="review"]')
                if review_btn.count() > 0:
                    text = review_btn.first.inner_text()
                    count_match = re.search(r'([\d,]+)\s*review', text, re.IGNORECASE)
                    if count_match:
                        details['review_count'] = int(count_match.group(1).replace(',', ''))
            except:
                pass

            # Go back to results list
            try:
                back_btn = self.page.locator('button[jsaction*="back"], button[aria-label="Back"]')
                if back_btn.count() > 0:
                    back_btn.first.click()
                    time.sleep(0.3)
                else:
                    # Press escape to close panel
                    self.page.keyboard.press('Escape')
                    time.sleep(0.3)
            except:
                self.page.keyboard.press('Escape')

        except Exception as e:
            logger.debug(f"Error in deep extraction: {e}")
            try:
                self.page.keyboard.press('Escape')
            except:
                pass

        return details

    def _enrich_listings(self, table_name: str, query: str, city: str):
        """
        Enrich listings with website URL and full address by navigating directly
        to each business's Google Maps page (avoids stale-element issues with
        the click-in-list-view approach).

        Uses confirmed selectors from live DOM inspection:
          - website:  a[data-item-id="authority"]
          - phone:    button[data-item-id^="phone"]  (data-item-id is "phone:tel:+1...")
          - address:  button[data-item-id="address"]
          - plus code: button[aria-label*="Plus code"]
        """
        try:
            cursor = self.db.conn.execute(
                f'SELECT dedup_hash, name, google_url, phone_number, zip_code, plus_code, review_count '
                f'FROM "{table_name}" '
                f'WHERE (website_url IS NULL OR website_url = \'\') '
                f'AND search_city = ? AND google_url IS NOT NULL',
                (city,)
            )
            businesses = cursor.fetchall()
        except Exception as e:
            logger.debug(f"Could not query for website enrichment: {e}")
            return

        if not businesses:
            return

        logger.info(f"Fetching websites for {len(businesses)} businesses in {city}")

        for dedup_hash, name, google_url, existing_phone, existing_zip, existing_plus, existing_review_count in businesses:
            try:
                self.page.goto(google_url, timeout=12000, wait_until='domcontentloaded')
                time.sleep(1.2)

                updates = {}

                # Website: a[data-item-id="authority"]
                website_el = self.page.locator('a[data-item-id="authority"]')
                if website_el.count() > 0:
                    href = website_el.first.get_attribute('href') or ''
                    if href and 'google.com' not in href:
                        updates['website_url'] = href

                # Full address: button[data-item-id="address"]
                # aria-label = "Address: 4620 Washington Ave Suite A, Houston, TX 77007"
                addr_el = self.page.locator('button[data-item-id="address"]')
                if addr_el.count() > 0:
                    aria = addr_el.first.get_attribute('aria-label') or ''
                    if aria:
                        full_addr = re.sub(r'^Address:\s*', '', aria).strip()
                        updates['address_text'] = full_addr
                        # Extract zip code from full address if not already present
                        if not existing_zip:
                            zip_match = re.search(r'\b(\d{5})(?:-\d{4})?\b', full_addr)
                            if zip_match:
                                updates['zip_code'] = zip_match.group(1)

                # Plus code (only if missing)
                if not existing_plus:
                    plus_el = self.page.locator('button[aria-label*="Plus code"]')
                    if plus_el.count() > 0:
                        aria = plus_el.first.get_attribute('aria-label') or ''
                        if 'Plus code' in aria:
                            updates['plus_code'] = re.sub(r'^Plus code:\s*', '', aria).strip()

                # Phone fallback (only if missing from list-view)
                if not existing_phone:
                    phone_el = self.page.locator('button[data-item-id^="phone"]')
                    if phone_el.count() > 0:
                        aria = phone_el.first.get_attribute('aria-label') or ''
                        phone_match = re.search(r'[\+]?[\d\s\-\(\)]{10,}', aria)
                        if phone_match:
                            updates['phone_number'] = phone_match.group(0).strip()

                # Review count fallback from detail page
                if not existing_review_count:
                    try:
                        review_btn = self.page.locator('button[jsaction*="reviewChart"], button[aria-label*="review"]')
                        if review_btn.count() > 0:
                            rev_text = review_btn.first.inner_text()
                            rev_match = re.search(r'([\d,]+)\s*review', rev_text, re.IGNORECASE)
                            if rev_match:
                                updates['review_count'] = int(rev_match.group(1).replace(',', ''))
                        if 'review_count' not in updates:
                            # Try aria-label on rating element: "Rated 4.9 out of 5, 255 reviews"
                            stars_el = self.page.locator('[aria-label*="out of 5"]')
                            if stars_el.count() > 0:
                                aria = stars_el.first.get_attribute('aria-label') or ''
                                rev_match = re.search(r'([\d,]+)\s*review', aria, re.IGNORECASE)
                                if rev_match:
                                    updates['review_count'] = int(rev_match.group(1).replace(',', ''))
                    except Exception:
                        pass

                if updates:
                    set_clause = ', '.join(f'{k} = ?' for k in updates)
                    values = list(updates.values()) + [dedup_hash]
                    self.db.conn.execute(
                        f'UPDATE "{table_name}" SET {set_clause} WHERE dedup_hash = ?',
                        values
                    )
                    self.db.conn.commit()

            except Exception as e:
                logger.debug(f"Website enrich error for {name}: {e}")
                continue

    def scrape_city_deep(
        self,
        query: str,
        city: str,
        table_name: str,
        max_scrolls: int = None,
        preset: str = None,
        on_deep_progress: Callable = None
    ) -> tuple:
        """
        Scrape a city with deep extraction (clicks into each listing for phone/website).

        This is slower but extracts complete data including phone numbers and websites.

        Returns: (results_count, error_message)
        """
        if not self.page:
            raise RuntimeError("Browser not started. Call start_browser() first.")

        full_query = f"{query} in {city}"
        scrape_config = self.config.get('scrape', {})
        max_scrolls = max_scrolls or scrape_config.get('scroll_limit', 10)
        scroll_delay = scrape_config.get('scroll_delay_ms', 1500) / 1000
        timeout = scrape_config.get('page_load_timeout_ms', 15000)

        if self.on_city_start:
            self.on_city_start(city, query)

        try:
            # Navigate directly to search URL (more reliable than filling search box)
            self._navigate_search(query, city, timeout)
            time.sleep(2)

            # Scroll to load all results first
            feed = self.page.locator('div[role="feed"]')
            prev_count = 0
            stale_scrolls = 0
            for scroll_num in range(max_scrolls):
                try:
                    feed.evaluate("el => el.scrollTo(0, el.scrollHeight)")
                    time.sleep(scroll_delay)

                    # Check for end of results
                    end_indicator = self.page.locator("text=You've reached the end")
                    if end_indicator.count() > 0:
                        break

                    # Detect stale scrolling
                    cur_count = self.page.locator('div[role="article"]').count()
                    if cur_count == prev_count:
                        stale_scrolls += 1
                        if stale_scrolls >= 3:
                            break
                    else:
                        stale_scrolls = 0
                    prev_count = cur_count
                except:
                    break

            # Now extract all listings with deep details
            listings = self.page.locator('div[role="article"]').all()
            total_listings = len(listings)
            logger.info(f"Found {total_listings} listings for deep extraction")

            seen_hashes = set()
            inserted = 0

            for i, listing in enumerate(listings):
                try:
                    # First get basic info
                    basic_data = self._extract_single_listing(listing, i + 1, query, city)
                    if not basic_data or not basic_data.get('name'):
                        continue

                    # Check for duplicates
                    unique_str = f"{basic_data.get('name', '')}{basic_data.get('google_url', '')}"
                    dedup_hash = hashlib.md5(unique_str.encode()).hexdigest()

                    if dedup_hash in seen_hashes:
                        continue
                    seen_hashes.add(dedup_hash)

                    # Check cross-scrape exclusion
                    if dedup_hash in self.excluded_hashes:
                        is_new = False
                    else:
                        # Now get deep details (this is the slow part)
                        deep_data = self.extract_deep_details(listing)

                        # Merge deep data into basic data (deep data takes priority)
                        if deep_data.get('phone_number'):
                            basic_data['phone_number'] = deep_data['phone_number']
                        if deep_data.get('website_url'):
                            basic_data['website_url'] = deep_data['website_url']
                        if deep_data.get('full_address'):
                            basic_data['full_address'] = deep_data['full_address']
                        if deep_data.get('hours_text'):
                            basic_data['hours_text'] = deep_data['hours_text']
                        if deep_data.get('review_count') and not basic_data.get('review_count'):
                            basic_data['review_count'] = deep_data['review_count']

                        basic_data['dedup_hash'] = dedup_hash

                        # Insert immediately and report progress
                        is_new = self.db.insert_result(table_name, basic_data)
                        if is_new:
                            inserted += 1

                    # Real-time progress callback for live monitor
                    if self.on_business_inserted:
                        self.on_business_inserted(basic_data, is_new)

                    # Report progress
                    if on_deep_progress:
                        on_deep_progress(i + 1, total_listings, basic_data)

                except Exception as e:
                    logger.debug(f"Error processing listing {i}: {e}")
                    continue

            if self.on_city_complete:
                self.on_city_complete(city, inserted, len(seen_hashes))

            return inserted, None

        except PlaywrightTimeout:
            error_msg = f"Timeout loading results for {city}"
            logger.warning(error_msg)
            return 0, error_msg

        except Exception as e:
            error_msg = str(e)
            logger.error(f"Error scraping {city}: {error_msg}")
            return 0, error_msg

    def enrich_table_with_deep_data(
        self,
        table_name: str,
        limit: int = None,
        on_progress: Callable = None
    ) -> Dict[str, int]:
        """
        Enrich an existing table with phone numbers and websites using deep extraction.

        Finds businesses missing phone/website and looks them up on Google Maps.

        Args:
            table_name: Table to enrich
            limit: Max businesses to enrich (None = all)
            on_progress: Callback(current, total, business_name)

        Returns:
            Dict with enrichment statistics
        """
        stats = {
            'total_processed': 0,
            'phones_found': 0,
            'websites_found': 0,
            'errors': 0
        }

        # Get businesses missing phone or website
        query = f"""
            SELECT dedup_hash, name, google_url, search_city
            FROM "{table_name}"
            WHERE (phone_number IS NULL OR phone_number = '')
               OR (website_url IS NULL OR website_url = '')
        """
        if limit:
            query += f" LIMIT {limit}"

        cursor = self.db.conn.execute(query)
        businesses = cursor.fetchall()

        if not businesses:
            logger.info("No businesses need enrichment")
            return stats

        logger.info(f"Enriching {len(businesses)} businesses in {table_name}")

        if not self.page:
            self.start_browser()

        for i, (dedup_hash, name, google_url, city) in enumerate(businesses):
            try:
                if on_progress:
                    on_progress(i + 1, len(businesses), name)

                # Navigate directly to the business
                if google_url:
                    self.page.goto(google_url, timeout=15000)
                    time.sleep(1.5)

                    # Extract details from detail view
                    details = {}

                    # Phone
                    try:
                        phone_btn = self.page.locator('button[data-item-id="phone"]')
                        if phone_btn.count() > 0:
                            aria = phone_btn.first.get_attribute('aria-label')
                            if aria:
                                phone_match = re.search(r'[\+]?[\d\s\-\(\)]{10,}', aria)
                                if phone_match:
                                    details['phone_number'] = phone_match.group(0).strip()
                                    stats['phones_found'] += 1
                    except:
                        pass

                    # Website
                    try:
                        website_link = self.page.locator('a[data-item-id="authority"]')
                        if website_link.count() > 0:
                            href = website_link.first.get_attribute('href')
                            if href:
                                details['website_url'] = href
                                stats['websites_found'] += 1
                    except:
                        pass

                    # Update database
                    if details:
                        updates = []
                        values = []
                        if details.get('phone_number'):
                            updates.append("phone_number = ?")
                            values.append(details['phone_number'])
                        if details.get('website_url'):
                            updates.append("website_url = ?")
                            values.append(details['website_url'])

                        if updates:
                            values.append(dedup_hash)
                            self.db.conn.execute(
                                f"UPDATE \"{table_name}\" SET {', '.join(updates)} WHERE dedup_hash = ?",
                                values
                            )
                            self.db.conn.commit()

                    stats['total_processed'] += 1

            except Exception as e:
                logger.debug(f"Error enriching {name}: {e}")
                stats['errors'] += 1
                continue

        return stats

    # Path for exported cookies (shared across workers)
    COOKIES_FILE = PROFILE_DIR / "cookies.json"

    @staticmethod
    def has_google_login() -> bool:
        """Check if a Google-authenticated browser profile exists."""
        return PROFILE_DIR.exists() and (PROFILE_DIR / "Default").exists()

    @classmethod
    def export_cookies(cls):
        """Export cookies from persistent profile to a JSON file for parallel workers."""
        if not cls.has_google_login():
            return
        import json
        pw = sync_playwright().start()
        ctx = pw.chromium.launch_persistent_context(
            str(PROFILE_DIR),
            headless=True,
            args=["--disable-blink-features=AutomationControlled"],
        )
        page = ctx.pages[0] if ctx.pages else ctx.new_page()
        page.goto("https://www.google.com/maps", timeout=30000)
        import time; time.sleep(2)
        cookies = ctx.cookies()
        with open(cls.COOKIES_FILE, "w") as f:
            json.dump(cookies, f)
        ctx.close()
        pw.stop()
        logger.info(f"Exported {len(cookies)} cookies for parallel workers")

    @classmethod
    def load_cookies(cls):
        """Load previously exported cookies."""
        import json
        if cls.COOKIES_FILE.exists():
            with open(cls.COOKIES_FILE) as f:
                return json.load(f)
        return None

    @staticmethod
    def google_login():
        """
        Open a browser window for the user to sign into Google manually.
        The session is saved to a persistent profile for future scrapes.
        """
        PROFILE_DIR.mkdir(parents=True, exist_ok=True)
        pw = sync_playwright().start()
        ctx = pw.chromium.launch_persistent_context(
            str(PROFILE_DIR),
            headless=False,
            viewport={"width": 1280, "height": 900},
            args=["--disable-blink-features=AutomationControlled"],
        )
        page = ctx.pages[0] if ctx.pages else ctx.new_page()
        page.add_init_script(
            'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'
        )
        page.goto("https://accounts.google.com/signin")
        # Wait until the user reaches a Google Maps page (signed in)
        print("Sign into your Google account in the browser window.")
        print("After signing in, navigate to Google Maps and close the browser.")
        try:
            # Block until the browser is closed by the user
            page.wait_for_event("close", timeout=300_000)
        except Exception:
            pass
        try:
            ctx.close()
        except Exception:
            pass
        pw.stop()

    def start_browser(self):
        """Initialize browser with proxy and optimizations."""
        # Skip if browser already running (idempotent)
        if self.browser is not None:
            logger.debug("Browser already running, skipping start")
            return

        self._playwright = sync_playwright().start()

        browser_config = self.config.get('browser', {})
        headless = browser_config.get('headless', False)

        # Configure proxy
        proxy_config = None
        if self.config.get('proxy', {}).get('enabled', False):
            proxy_server = self.config['proxy'].get('server', '')
            if proxy_server and 'USERNAME' not in proxy_server:
                proxy_config = {"server": proxy_server}
            else:
                logger.warning("Proxy enabled but credentials not configured. Running without proxy.")

        # Try to use Google authentication if available
        cookies = self.load_cookies()
        use_persistent = self.has_google_login() and proxy_config is None and not cookies

        if use_persistent:
            # Single-worker mode: use persistent profile directly
            try:
                self.context = self._playwright.chromium.launch_persistent_context(
                    str(PROFILE_DIR),
                    headless=headless,
                    viewport={"width": 1920, "height": 1080},
                    args=["--disable-blink-features=AutomationControlled"],
                )
                self.browser = self.context  # persistent context acts as both
                self.page = self.context.pages[0] if self.context.pages else self.context.new_page()
                self.page.add_init_script(
                    'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'
                )
                logger.info("Using authenticated Google profile")
            except Exception as e:
                # Profile locked by another process — fall back to cookies or plain
                logger.warning(f"Persistent profile locked ({e}), falling back")
                use_persistent = False

        if not use_persistent:
            # Launch browser without persistent profile
            self.browser = self._playwright.chromium.launch(
                headless=headless,
                proxy=proxy_config,
                args=["--disable-blink-features=AutomationControlled"],
            )
            self.context = self.browser.new_context(
                user_agent=browser_config.get('user_agent'),
                viewport={"width": 1920, "height": 1080},
            )
            self.page = self.context.new_page()
            self.page.add_init_script(
                'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'
            )

            # Inject cookies if available (parallel worker mode)
            if cookies:
                self.context.add_cookies(cookies)
                logger.info("Injected Google auth cookies")

        # Enable bandwidth blocking
        self.page.route("**/*", self._block_media)

        # Enable bandwidth tracking via response handler
        self.page.on("response", self._track_response)

        # Reset bandwidth counter for new session
        self.reset_bandwidth()

        # Dismiss any Google consent dialog on first navigation
        self._dismiss_consent()

        logger.info("Browser started successfully")

    def _dismiss_consent(self):
        """Navigate to Google Maps once and dismiss cookie/consent dialogs if present."""
        try:
            self.page.goto("https://www.google.com/maps", timeout=30000)
            # Check for consent form first — skip selector loop if none present
            consent_form = self.page.locator('form[action*="consent"]')
            if consent_form.count() == 0:
                return
            for selector in [
                'button:has-text("Accept all")',
                'button:has-text("Reject all")',
                'button:has-text("I agree")',
            ]:
                try:
                    btn = self.page.locator(selector).first
                    if btn.is_visible(timeout=1500):
                        btn.click()
                        time.sleep(1)
                        break
                except:
                    continue
        except Exception as e:
            logger.debug(f"Consent dismissal: {e}")

    def _navigate_search(self, query: str, city: str, timeout: int) -> None:
        """Navigate directly to a Google Maps search URL for reliability."""
        from urllib.parse import quote_plus
        search_url = f"https://www.google.com/maps/search/{quote_plus(f'{query} in {city}')}"
        browser_timeout = self.config.get('browser', {}).get('timeout_ms', 60000)
        self.page.goto(search_url, timeout=browser_timeout)
        self.page.wait_for_selector('div[role="feed"]', timeout=timeout)

    def stop_browser(self):
        """Clean up browser resources."""
        if self.context and self.context is not self.browser:
            self.context.close()
        self.context = None
        if self.browser:
            self.browser.close()
        self.browser = None
        if hasattr(self, '_playwright') and self._playwright:
            self._playwright.stop()
            self._playwright = None
        self.page = None
        logger.info("Browser stopped")

    def scrape_city(
        self,
        query: str,
        city: str,
        table_name: str,
        max_scrolls: int = None,
        preset: str = None,
        fetch_websites: bool = False
    ) -> tuple:
        """
        Scrape a single city for the given query.

        Returns: (results_count, error_message)
        """
        if not self.page:
            raise RuntimeError("Browser not started. Call start_browser() first.")

        full_query = f"{query} in {city}"
        scrape_config = self.config.get('scrape', {})
        max_scrolls = max_scrolls or scrape_config.get('scroll_limit', 10)
        scroll_delay = scrape_config.get('scroll_delay_ms', 1500) / 1000
        timeout = scrape_config.get('page_load_timeout_ms', 15000)

        # Create job entry for telemetry tracking
        job_id = self.db.create_job(table_name, query, city, preset)
        self.db.update_job_status(job_id, 'RUNNING')
        start_time = time.time()
        bandwidth_start = self._bandwidth_bytes

        if self.on_city_start:
            self.on_city_start(city, query)

        try:
            # Navigate directly to search URL (more reliable than filling search box)
            self._navigate_search(query, city, timeout)

            # Let initial results load
            time.sleep(2)

            # Scroll to load more results
            feed = self.page.locator('div[role="feed"]')

            # Track seen listings for incremental extraction
            seen_hashes = set()
            inserted = 0
            cross_dupes = 0
            prev_seen_count = 0
            stale_scrolls = 0

            # Extract and insert incrementally during scrolling
            for scroll_num in range(max_scrolls + 1):  # +1 to extract before first scroll too
                # Extract current visible listings
                current_results = self.extract_category_one(self.page, query, city)

                # Process only new listings (not seen in this city yet)
                for result in current_results:
                    unique_str = f"{result.get('name', '')}{result.get('phone_number', '')}"
                    dedup_hash = hashlib.md5(unique_str.encode()).hexdigest()

                    # Skip if already processed in this city
                    if dedup_hash in seen_hashes:
                        continue
                    seen_hashes.add(dedup_hash)

                    # Check cross-scrape exclusion
                    if dedup_hash in self.excluded_hashes:
                        cross_dupes += 1
                        is_new = False
                    else:
                        is_new = self.db.insert_result(table_name, result)
                        if is_new:
                            inserted += 1

                    # Notify callback for live monitoring (real-time feed update)
                    if self.on_business_inserted:
                        self.on_business_inserted(result, is_new)

                # Detect stale scrolling (no new results found)
                if len(seen_hashes) == prev_seen_count:
                    stale_scrolls += 1
                    if stale_scrolls >= 3:
                        logger.debug(f"No new results after {stale_scrolls} scrolls, stopping at {len(seen_hashes)} listings")
                        break
                else:
                    stale_scrolls = 0
                prev_seen_count = len(seen_hashes)

                # Don't scroll on last iteration
                if scroll_num >= max_scrolls:
                    break

                # Scroll the feed
                feed.evaluate("node => node.scrollTop += 5000")
                time.sleep(scroll_delay)

                # Check for "end of list" indicator
                try:
                    end_indicator = self.page.locator("text=You've reached the end of the list")
                    if end_indicator.count() > 0:
                        logger.debug(f"Reached end of list at scroll {scroll_num + 1}")
                        break
                except:
                    pass

            if cross_dupes > 0:
                logger.debug(f"Skipped {cross_dupes} cross-scrape duplicates in {city}")

            # Optionally enrich with website URLs via direct page navigation
            # (only when --websites flag is set; adds ~1-2s per business)
            if fetch_websites:
                self._enrich_listings(table_name, query, city)

            # Update job with success telemetry
            elapsed = time.time() - start_time
            bandwidth_used = (self._bandwidth_bytes - bandwidth_start) / 1024  # KB
            self.db.update_job_status(
                job_id, 'COMPLETED',
                results_found=inserted,
                execution_time_sec=elapsed,
                bandwidth_used_kb=bandwidth_used
            )

            if self.on_city_complete:
                self.on_city_complete(city, inserted, len(seen_hashes))

            return inserted, None

        except PlaywrightTimeout as e:
            error_msg = f"Timeout: {str(e)[:50]}"
            logger.warning(f"Timeout scraping {city}: {e}")

            # Update job with failure telemetry
            elapsed = time.time() - start_time
            bandwidth_used = (self._bandwidth_bytes - bandwidth_start) / 1024
            self.db.update_job_status(
                job_id, 'FAILED',
                execution_time_sec=elapsed,
                bandwidth_used_kb=bandwidth_used,
                error_message=error_msg
            )
            self.db.increment_job_retry(job_id)

            if self.on_city_error:
                self.on_city_error(city, error_msg)
            return 0, error_msg

        except Exception as e:
            error_msg = f"Error: {str(e)[:50]}"
            logger.error(f"Error scraping {city}: {e}")

            # Update job with failure telemetry
            elapsed = time.time() - start_time
            bandwidth_used = (self._bandwidth_bytes - bandwidth_start) / 1024
            self.db.update_job_status(
                job_id, 'FAILED',
                execution_time_sec=elapsed,
                bandwidth_used_kb=bandwidth_used,
                error_message=error_msg
            )
            self.db.increment_job_retry(job_id)

            if self.on_city_error:
                self.on_city_error(city, error_msg)
            return 0, error_msg

    def run_scrape(
        self,
        query: str,
        locations: List[str],
        table_name: str,
        mode: str = "full",
        max_scrolls: int = None,
        preset: str = None,
        deep: bool = False,
        fetch_websites: bool = False
    ) -> dict:
        """
        Run a full scrape across all locations.

        Args:
            query: Search term
            locations: List of "City, State" strings
            table_name: Database table to store results
            mode: "test" (1 scroll per city) or "full" (all scrolls)
            max_scrolls: Override scroll limit
            preset: Preset name for job tracking
            deep: If True, use deep extraction (clicks into each listing for phone/website)

        Returns:
            {
                'total_inserted': int,
                'total_found': int,
                'cities_completed': int,
                'cities_failed': int,
                'errors': List[dict]
            }
        """
        if mode == "test":
            max_scrolls = 1

        stats = {
            'total_inserted': 0,
            'total_found': 0,
            'cities_completed': 0,
            'cities_failed': 0,
            'errors': []
        }

        # Track if we started the browser (so we know if we should stop it)
        browser_was_running = self.browser is not None
        self.start_browser()

        try:
            for city in locations:
                if deep:
                    inserted, error = self.scrape_city_deep(query, city, table_name, max_scrolls, preset)
                else:
                    inserted, error = self.scrape_city(query, city, table_name, max_scrolls, preset, fetch_websites=fetch_websites)

                stats['total_inserted'] += inserted
                stats['total_found'] += inserted  # For now, same as inserted due to dedup

                if error:
                    stats['cities_failed'] += 1
                    stats['errors'].append({'city': city, 'error': error})
                else:
                    stats['cities_completed'] += 1

        finally:
            # Only stop browser if we started it (not if it was already running)
            if not browser_was_running:
                self.stop_browser()

        return stats