scraper_google.py

"""
Google Maps scraper — same interface as scraper_2gis / scraper_yandex.

Search URL: https://www.google.com/maps/search/{category}+{city}
Card URL:   https://www.google.com/maps/place/...

Anti-bot: random delays, stealth context, captcha detection, retry on fail.
"""

import asyncio
import hashlib
import logging
import random
import re
from typing import Optional

from playwright.async_api import Page, async_playwright

import storage
from models import (
    BOOKING_DOMAINS, BOOKING_TEXT_PATTERNS,
    BusinessCard, Phone, Review,
    _normalize_tg, _normalize_wa,
)

logger = logging.getLogger(__name__)

_PAIN_KEYWORDS_DEFAULT = [
    "не дозвониться", "не дозвонилась", "не дозвонился",
    "не могу записаться", "не можем записаться",
    "запись", "записаться", "не берут трубку",
    "долго ждать", "долго ждала", "долго ждал",
    "неудобно", "неудобная запись",
    "через директ", "в директ", "написала в вотсап", "написал в вотсап",
    "нет онлайн", "нет записи",
]


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _make_id(url: str) -> str:
    return hashlib.md5(url.encode()).hexdigest()[:12]


def _clean_url(url: str) -> str:
    """Strip query params that vary between visits."""
    return url.split("?")[0].rstrip("/")


def _search_url(city: str, category: str) -> str:
    q = f"{category} {city}".replace(" ", "+")
    return f"https://www.google.com/maps/search/{q}/"


def _is_blocked(content: str) -> bool:
    signals = ["detected unusual traffic", "captcha", "Sorry, we could not process"]
    return any(s.lower() in content.lower() for s in signals) and len(content) < 20000


def _is_booking(href: str, text: str) -> bool:
    if not href:
        return False
    for d in BOOKING_DOMAINS:
        if d in href:
            return True
    tl = text.lower()
    for p in BOOKING_TEXT_PATTERNS:
        if p in tl:
            return True
    return False


async def _new_context(pw, headless: bool):
    browser = await pw.chromium.launch(
        headless=headless,
        args=["--disable-blink-features=AutomationControlled", "--no-sandbox"],
    )
    ctx = await browser.new_context(
        user_agent=(
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/124.0.0.0 Safari/537.36"
        ),
        locale="ru-RU",
        timezone_id="Europe/Moscow",
        viewport={"width": 1440, "height": 900},
        java_script_enabled=True,
    )
    await ctx.add_init_script(
        "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"
    )
    return browser, ctx


# ---------------------------------------------------------------------------
# Search — collect card URLs from the results panel
# ---------------------------------------------------------------------------

async def search_cards(
    page: Page,
    city: str,
    category: str,
    min_cards: int,
    delay_min: float,
    delay_max: float,
) -> list[str]:
    url = _search_url(city, category)
    logger.info(f"  [Google] Поиск: {url}")

    try:
        await page.goto(url, wait_until="load", timeout=30000)
        # Accept consent if shown (EU/RU cookie dialog)
        try:
            btn = page.locator("button:has-text('Принять'), button:has-text('Accept all'), form[action*='consent'] button").first
            if await btn.count() > 0:
                await btn.click(timeout=3000)
                await asyncio.sleep(1)
        except Exception:
            pass
        await page.wait_for_selector(
            "[data-result-index], .hfpxzc, a[href*='/maps/place/']",
            timeout=12000,
        )
    except Exception as e:
        logger.warning(f"  [Google] Не удалось загрузить поиск: {e}")
        return []

    content = await page.content()
    if _is_blocked(content):
        logger.warning("  [Google] Обнаружена блокировка/капча")
        return []

    urls: list[str] = []
    seen: set[str] = set()

    for _ in range(20):  # max 20 scroll attempts
        links = await page.locator("a[href*='/maps/place/']").all()
        for a in links:
            try:
                href = await a.get_attribute("href", timeout=300)
                if not href:
                    continue
                m = re.match(r"(https://www\.google\.com/maps/place/[^?]+)", href)
                if not m:
                    continue
                clean = m.group(1).rstrip("/")
                if clean not in seen:
                    seen.add(clean)
                    urls.append(clean)
            except Exception:
                pass

        if len(urls) >= min_cards:
            break

        # Scroll the results panel
        await page.evaluate("""() => {
            const panel = document.querySelector('[role="feed"], .m6QErb');
            if (panel) panel.scrollTop += 1500;
            else window.scrollBy(0, 1500);
        }""")
        await asyncio.sleep(random.uniform(delay_min, delay_max))

    logger.info(f"  [Google] Найдено URL: {len(urls)}")
    return urls[:min_cards * 2]  # cap — parse will filter by rating etc.


# ---------------------------------------------------------------------------
# Card parser
# ---------------------------------------------------------------------------

async def parse_card(
    page: Page,
    url: str,
    city: str,
    category: str,
    max_reviews: int = 15,
    pain_keywords: list[str] | None = None,
) -> tuple[Optional[BusinessCard], list[Review]]:
    if pain_keywords is None:
        pain_keywords = _PAIN_KEYWORDS_DEFAULT

    try:
        await page.goto(url, wait_until="load", timeout=30000)
        await page.wait_for_selector(
            "h1, [data-attrid='title'], .DUwDvf, .fontHeadlineLarge",
            timeout=10000,
        )
    except Exception as e:
        logger.warning(f"  [Google] Не удалось загрузить карточку {url}: {e}")
        return None, []

    content = await page.content()
    if _is_blocked(content):
        logger.warning("  [Google] Капча на карточке")
        return None, []

    await asyncio.sleep(random.uniform(0.5, 1.5))

    card_id = _make_id(url)

    # name
    name = ""
    for sel in ["h1.DUwDvf", "h1.fontHeadlineLarge", "h1", "[data-attrid='title']"]:
        try:
            el = page.locator(sel).first
            if await el.count() > 0:
                name = (await el.inner_text(timeout=2000)).strip()
                if name:
                    break
        except Exception:
            pass

    if not name:
        return None, []

    # address
    address = ""
    for sel in [
        "button[data-item-id='address']",
        "[data-item-id='address']",
        "[aria-label*='Адрес'], [aria-label*='Address']",
        "button[data-tooltip='Copy address']",
    ]:
        try:
            el = page.locator(sel).first
            if await el.count() > 0:
                address = (await el.inner_text(timeout=1500)).strip()
                if address:
                    break
        except Exception:
            pass

    # rating
    rating: Optional[float] = None
    for sel in [
        "span.ceNzKf[aria-label]",
        "[aria-label*='звёзд'], [aria-label*='stars']",
        "span.MW4etd",
    ]:
        try:
            el = page.locator(sel).first
            if await el.count() > 0:
                aria = await el.get_attribute("aria-label", timeout=1000)
                if aria:
                    m = re.search(r"([\d][,.][\d])", aria)
                    if m:
                        rating = float(m.group(1).replace(",", "."))
                        break
                t = (await el.inner_text(timeout=1000)).strip()
                m = re.search(r"([\d][,.][\d])", t)
                if m:
                    rating = float(m.group(1).replace(",", "."))
                    break
        except Exception:
            pass

    # review count
    review_count: Optional[int] = None
    for sel in ["span.UY7F9", "[aria-label*='отзыв'], [aria-label*='review']"]:
        try:
            el = page.locator(sel).first
            if await el.count() > 0:
                t = (await el.inner_text(timeout=1000)).strip()
                m = re.search(r"(\d[\d\s]*)", t.replace("\xa0", ""))
                if m:
                    review_count = int(re.sub(r"\D", "", m.group(1)))
                    break
        except Exception:
            pass

    # phones
    phones: list[Phone] = []
    _phone_re = re.compile(r"(\+?[78][\s\-\(\d]{9,20})")
    for sel in [
        "button[data-item-id='phone:tel']",
        "a[href^='tel:']",
        "[data-item-id*='phone']",
        "[aria-label*='Телефон'], [aria-label*='Phone']",
    ]:
        try:
            for el in await page.locator(sel).all():
                href = await el.get_attribute("href", timeout=300) or ""
                if href.startswith("tel:"):
                    raw = href[4:].strip()
                else:
                    t = (await el.inner_text(timeout=800)).strip()
                    m = _phone_re.search(t)
                    raw = m.group(1) if m else t.split("\n")[0]
                digits = re.sub(r"\D", "", raw)
                if len(digits) >= 10 and raw not in [p.raw for p in phones]:
                    phones.append(Phone(raw=raw))
        except Exception:
            pass
        if phones:
            break

    # website
    website_url: Optional[str] = None
    has_website = False
    for sel in [
        "a[data-item-id='authority']",
        "a[aria-label*='сайт'], a[aria-label*='website'], a[aria-label*='Website']",
    ]:
        try:
            el = page.locator(sel).first
            if await el.count() > 0:
                website_url = await el.get_attribute("href", timeout=1000)
                if website_url and "google.com" not in website_url:
                    has_website = True
                    break
        except Exception:
            pass

    # working hours
    working_hours = ""
    try:
        el = page.locator("[aria-label*='часы'], [aria-label*='Hours'], .t39EBf").first
        if await el.count() > 0:
            working_hours = (await el.inner_text(timeout=1500)).strip().split("\n")[0]
    except Exception:
        pass

    # messengers — scan all links
    whatsapp: Optional[str] = None
    telegram: Optional[str] = None
    try:
        for a in await page.locator("a").all():
            href = await a.get_attribute("href", timeout=200) or ""
            if not whatsapp:
                whatsapp = _normalize_wa(href)
            if not telegram:
                telegram = _normalize_tg(href)
            if whatsapp and telegram:
                break
    except Exception:
        pass

    # booking detection
    has_booking = False
    booking_url: Optional[str] = None
    try:
        for a in await page.locator("a").all():
            href = await a.get_attribute("href", timeout=200) or ""
            text = await a.inner_text(timeout=200) or ""
            if _is_booking(href, text):
                has_booking = True
                booking_url = href
                break
    except Exception:
        pass

    # coordinates from URL or page
    lat: Optional[float] = None
    lon: Optional[float] = None
    try:
        current_url = page.url
        m = re.search(r"@(-?\d+\.\d+),(-?\d+\.\d+)", current_url)
        if m:
            lat, lon = float(m.group(1)), float(m.group(2))
    except Exception:
        pass
    if not lat:
        try:
            coords = await page.evaluate(r"""() => {
                const m = document.documentElement.innerHTML.match(/"coordinates"\s*:\s*\[(-?\d+\.\d+),\s*(-?\d+\.\d+)\]/);
                if (m) return {lat: parseFloat(m[2]), lon: parseFloat(m[1])};
                return null;
            }""")
            if coords:
                lat, lon = coords["lat"], coords["lon"]
        except Exception:
            pass

    # reviews
    reviews: list[Review] = []
    try:
        reviews = await _parse_reviews(page, card_id, max_reviews)
    except Exception as e:
        logger.warning(f"  [Google] Ошибка сбора отзывов: {e}")

    # pain point detection
    pain_point = False
    pain_point_quote = ""
    for r in reviews:
        tl = r.text.lower()
        for kw in pain_keywords:
            if kw in tl:
                pain_point = True
                idx = tl.index(kw)
                pain_point_quote = r.text[max(0, idx - 30): idx + 80].strip()
                break
        if pain_point:
            break

    card = BusinessCard(
        card_id=card_id,
        name=name,
        address=address,
        city=city,
        category=category,
        phones=phones,
        rating=rating,
        review_count=review_count,
        working_hours=working_hours,
        has_website=has_website,
        website_url=website_url,
        card_url=_clean_url(url),
        lat=lat,
        lon=lon,
        source="google",
        whatsapp=whatsapp,
        telegram=telegram,
        has_booking=has_booking,
        booking_url=booking_url,
        pain_point=pain_point,
        pain_point_quote=pain_point_quote,
    )
    return card, reviews


async def _parse_reviews(page: Page, card_id: str, max_reviews: int) -> list[Review]:
    reviews: list[Review] = []

    # Click "Reviews" tab
    try:
        tab = page.locator(
            "button[aria-label*='Отзывы'], button[aria-label*='Reviews'], "
            "[data-tab-index='1'], button:has-text('Отзывы')"
        ).first
        if await tab.count() > 0:
            await tab.click(timeout=3000)
            await asyncio.sleep(1.5)
    except Exception:
        pass

    # Sort by newest (optional, skip if fails)
    try:
        sort_btn = page.locator("button[aria-label*='Сортировать'], button[aria-label*='Sort']").first
        if await sort_btn.count() > 0:
            await sort_btn.click(timeout=2000)
            await asyncio.sleep(0.5)
            newest = page.locator("[data-index='1'], li[aria-label*='Новым'], li[aria-label*='Newest']").first
            if await newest.count() > 0:
                await newest.click(timeout=1500)
                await asyncio.sleep(1)
    except Exception:
        pass

    collected_keys: set[str] = set()

    for _ in range(max_reviews // 3 + 3):
        items = await page.locator(".jftiEf, [data-review-id]").all()
        for item in items:
            try:
                key = await item.get_attribute("data-review-id", timeout=300)
                if not key:
                    html = await item.inner_html(timeout=1000)
                    key = hashlib.md5(html[100:500].encode()).hexdigest()
                if key in collected_keys:
                    continue
                collected_keys.add(key)

                author = ""
                try:
                    author = (await item.locator(".d4r55, [class*='reviewer-name']").first.inner_text(timeout=500)).strip()
                except Exception:
                    pass

                date = ""
                try:
                    date = (await item.locator(".rsqaWe, [class*='review-date']").first.inner_text(timeout=500)).strip()
                except Exception:
                    pass

                text = ""
                try:
                    # Expand "Ещё" / "More" button first
                    more = item.locator("button.w8nwRe, button[aria-label*='Ещё'], button[aria-label*='More']").first
                    if await more.count() > 0:
                        await more.click(timeout=1000)
                except Exception:
                    pass
                try:
                    text = (await item.locator(".wiI7pd, [class*='review-text']").first.inner_text(timeout=800)).strip()
                except Exception:
                    pass

                r_val: Optional[float] = None
                try:
                    aria = await item.locator("[aria-label*='звёзд'], [aria-label*='star']").first.get_attribute("aria-label", timeout=500)
                    m = re.search(r"([\d][,.][\d]|\d)", aria or "")
                    if m:
                        r_val = float(m.group(1).replace(",", "."))
                except Exception:
                    pass

                if author or text:
                    reviews.append(Review(card_id=card_id, author=author, date=date, rating=r_val, text=text))

                if len(reviews) >= max_reviews:
                    break
            except Exception:
                pass

        if len(reviews) >= max_reviews:
            break

        # Scroll reviews panel
        try:
            await page.evaluate("""() => {
                const feed = document.querySelector('.m6QErb[aria-label], .DxyBCb');
                if (feed) feed.scrollTop += 1500;
            }""")
            await asyncio.sleep(1.5)
        except Exception:
            break

    return reviews


# ---------------------------------------------------------------------------
# ScraperGoogle — same interface as Scraper2GIS / ScraperYandex
# ---------------------------------------------------------------------------

class ScraperGoogle:
    def __init__(self, cfg: dict):
        self.cfg = cfg
        self.headless = cfg.get("browser", {}).get("headless", True)
        self.delay_min = cfg.get("delays", {}).get("between_cards_min_sec", 3)
        self.delay_max = cfg.get("delays", {}).get("between_cards_max_sec", 8)
        self.max_reviews = cfg.get("reviews_per_card", 15)
        self.rating_min = cfg.get("rating_filter", {}).get("min", 0.0)
        self.rating_max = cfg.get("rating_filter", {}).get("max", 5.0)
        self.pain_keywords = cfg.get("pain_point_keywords", _PAIN_KEYWORDS_DEFAULT)
        self.filter_no_booking = cfg.get("filters", {}).get("exclude_has_booking", True)
        self.filter_require_messenger = cfg.get("filters", {}).get("require_messenger", True)

    async def run(self, tasks: list[dict], saved_urls: set[str], on_card, on_reviews):
        async with async_playwright() as pw:
            browser, ctx = await _new_context(pw, self.headless)
            page = await ctx.new_page()

            for idx, task in enumerate(tasks):
                city = task["city"]
                category = task["category"]
                min_cards = task.get("min_cards", 20)

                logger.info(f"\n[{idx+1}/{len(tasks)}] Google Maps: {city} / {category}")

                urls = await search_cards(
                    page, city, category, min_cards,
                    self.delay_min, self.delay_max,
                )
                logger.info(f"  Найдено URL: {len(urls)}")

                for url in urls:
                    clean = _clean_url(url)
                    if clean in saved_urls:
                        continue

                    await asyncio.sleep(random.uniform(self.delay_min, self.delay_max))

                    card, revs = await parse_card(
                        page, url, city, category,
                        max_reviews=self.max_reviews,
                        pain_keywords=self.pain_keywords,
                    )
                    if card is None:
                        continue

                    # rating filter
                    if card.rating is not None:
                        if not (self.rating_min <= card.rating <= self.rating_max):
                            logger.info(f"  Пропуск (рейтинг {card.rating}): {card.name}")
                            continue

                    # booking filter
                    if self.filter_no_booking and card.has_booking:
                        logger.info(f"  Пропуск (есть букинг): {card.name}")
                        continue

                    # messenger filter
                    if self.filter_require_messenger and not card.has_messenger:
                        logger.info(f"  Пропуск (нет мессенджера): {card.name}")
                        continue

                    saved_urls.add(clean)
                    on_card(card)
                    if revs:
                        on_reviews(revs)
                    logger.info(f"  ✅ {card.name} | ★{card.rating} | {card.phones_str}")

            await browser.close()