"""Боевой парсер Яндекс.Карт. Алгоритм: 1. Открываем `https://yandex.ru/maps/{city_id}/moscow/search/{query}/` 2. Скроллим .scroll__container пока появляются новые .search-snippet-view (или пока не упрёмся в MAX_SCROLLS / MAX_CARDS_PER_CATEGORY) 3. Из каждой карточки забираем href ссылки + категорию из листинга (категория есть в листинге, но НЕТ на детальной странице) 4. По каждому href: driver.get(href) + парсинг боковой панели 5. Возвращаем list[dict] под схему leads Селекторы подтверждены эмпирически в Phase 0 (test_yandex_v3.py, 3/3 карточек). См. sessions/2026-05-01_phase0_research.md. """ import logging from typing import Optional from botasaurus.browser import browser, Driver import config from normalization import ( is_garbage_social, normalize_phone, parse_rating, parse_reviews_count, ) logger = logging.getLogger(__name__) # ─────────────────────────────────────────────────────────────────────── # СЕЛЕКТОРЫ — финал из Phase 0 # ─────────────────────────────────────────────────────────────────────── SEL = { # листинг "list_container": ".scroll__container", "card_in_list": ".search-snippet-view", "card_link": "a", # внутри .search-snippet-view "card_category": ".search-business-snippet-view__category", # детальная страница "title": "h1.orgpage-header-view__header", "address": ".orgpage-header-view__address", "phone_container": ".orgpage-phones-view__phone-number", # текст, не tel:! "website_link": ".business-urls-view__link", "rating": ".business-rating-badge-view__rating-text", "reviews_count": ".business-header-rating-view__text", } # ─────────────────────────────────────────────────────────────────────── # Этап 1: собрать ссылки + категории из листинга # ─────────────────────────────────────────────────────────────────────── def _scroll_list_via_js(driver: Driver, by: int = 2000) -> bool: """Скроллит панель списка чистым JS. Возвращает True если контейнер найден и проскроллен.""" js = f""" const containers = document.querySelectorAll('{SEL["list_container"]}'); let scrolled = false; for (const c of containers) {{ // Берём только тот контейнер, у которого есть прокручиваемый контент if (c.scrollHeight > c.clientHeight) {{ c.scrollTop = c.scrollTop + {by}; scrolled = true; }} }} return scrolled; """ try: return bool(driver.run_js(js)) except Exception as e: logger.warning(f" JS-скролл упал: {e}") return False def _collect_card_links(driver: Driver, max_cards: int) -> list[dict]: """Скроллит список и собирает [{href, category}, ...]""" cards_before = driver.count(SEL["card_in_list"]) logger.info(f" Стартовое количество карточек: {cards_before}") stagnant_iterations = 0 # счётчик итераций без новых карточек MAX_STAGNANT = 3 # после стольких безуспешных скроллов — выходим for i in range(1, config.MAX_SCROLLS + 1): scrolled_ok = _scroll_list_via_js(driver, by=2000) if not scrolled_ok: logger.warning(f" Скролл #{i}: контейнер не найден / не прокручиваемый") break driver.sleep(1.0) # ждём ленивую подгрузку (2026-05-18: 1.5 → 1.0) cards_now = driver.count(SEL["card_in_list"]) delta = cards_now - cards_before logger.info(f" Скролл #{i}: карточек {cards_now} (+{delta})") if cards_now >= max_cards: logger.info(f" Достигнут лимит карточек ({max_cards}) — стоп") break if delta == 0: stagnant_iterations += 1 if stagnant_iterations >= MAX_STAGNANT: logger.info( f" Карточки не растут {MAX_STAGNANT} итерации подряд → конец листинга" ) break else: stagnant_iterations = 0 cards_before = cards_now # Сбор ссылок cards = driver.select_all(SEL["card_in_list"])[:max_cards] items: list[dict] = [] for card in cards: link_el = card.select(SEL["card_link"]) if not link_el: continue href = link_el.get_attribute("href") or "" if not href: continue if href.startswith("/"): href = "https://yandex.ru" + href # Пропускаем рекомендательные подборки Яндекса (не organization-карточки). # Признаки: /maps/discovery/ — подборки, /maps/category/ — каталог категорий. if "/maps/discovery/" in href or "/maps/category/" in href: continue cat_el = card.select(SEL["card_category"]) category = cat_el.text.strip() if (cat_el and cat_el.text) else None items.append({"href": href, "category": category}) logger.info(f" Собрано ссылок на детальные карточки: {len(items)}") return items # ─────────────────────────────────────────────────────────────────────── # Этап 2: парсинг детальной карточки # ─────────────────────────────────────────────────────────────────────── def _safe_text(driver: Driver, selector: str) -> Optional[str]: el = driver.select(selector, wait=None) # wait=None: не висеть 4с на отсутствующем if el and el.text: return el.text.strip() return None def _safe_attr(driver: Driver, selector: str, attr: str) -> Optional[str]: el = driver.select(selector, wait=None) return el.get_attribute(attr) if el else None def _parse_phones_on_page(driver: Driver) -> list[str]: """В Яндекс.Картах телефон — текст в .orgpage-phones-view__phone-number, не tel:.""" phones: list[str] = [] # Может быть несколько телефонов на одной карточке elements = driver.select_all(SEL["phone_container"], wait=None) for el in elements: if el.text: normalized = normalize_phone(el.text.strip()) if normalized and normalized not in phones: phones.append(normalized) return phones def _parse_socials_on_page(driver: Driver) -> dict[str, str]: """Соцсети с фильтром мусора (vk.com/yandex.maps и т.п.).""" socials: dict[str, str] = {} patterns = [ ("vk_url", 'a[href*="vk.com"]'), ("telegram_url", 'a[href*="t.me"]'), ("instagram_url", 'a[href*="instagram.com"]'), ("youtube_url", 'a[href*="youtube.com"]'), ] for field, selector in patterns: # Берём все ссылки и ищем первую "не мусорную" links = driver.select_all(selector, wait=None) for link in links: href = link.get_attribute("href") if href and not is_garbage_social(href): socials[field] = href break return socials def _parse_card_detail( driver: Driver, card_meta: dict, city: str = "Москва", region: str | None = None, district: str | None = None, ) -> Optional[dict]: """Открыть детальную страницу и собрать поля. Возвращает dict под схему leads.""" href = card_meta["href"] category = card_meta.get("category") driver.get(href) # Ждём появления заголовка организации (адаптивно вместо фикс. sleep(2.5)). # wait_for_complete_page_load=False → get возвращается рано; заголовок ждём явно. driver.select(SEL["title"], wait=8) # Базовые поля. ВАЖНО: все select ниже — с wait=None (мгновенный возврат). # К моменту появления заголовка панель организации уже в DOM, а отсутствующие # поля (особенно соцсети — у кафе их обычно нет) НЕ должны висеть по 4с # default-таймаута каждое. Именно это давало ~11с/карточку (а не навигация). name = _safe_text(driver, SEL["title"]) if not name: logger.warning(f" Не нашли name на {href[:80]}") return None address = _safe_text(driver, SEL["address"]) rating_raw = _safe_text(driver, SEL["rating"]) reviews_raw = _safe_text(driver, SEL["reviews_count"]) # Телефоны phones = _parse_phones_on_page(driver) # Сайт (с фильтром на яндексовские внутренние ссылки) website = None web_el = driver.select(SEL["website_link"], wait=None) if web_el: href_web = web_el.get_attribute("href") if href_web and "yandex" not in href_web.lower(): website = href_web # Соцсети socials = _parse_socials_on_page(driver) # Контакты НЕ собираем из document.body.innerText — это ВЕСЬ видимый текст # страницы (реклама, блок «Похожие места», футер Яндекса), откуда регулярно # подмешивались ЧУЖИЕ телефоны/email и мусор: placeholder-номера (+73333333333), # несуществующие коды, имена картинок-как-email (asset@4x.png), почты # разработчиков тем сайтов (lella@elated-themes.com). # Достоверный телефон компании — только из спец-селектора (phones выше). # Email и доп.телефоны добираются отдельно с САЙТА компании через # enricher/website_analyzer (Tier 2): там приоритет mailto:/tel: и привязка # к домену сайта → контакты гарантированно принадлежат компании. return { "name": name, "phones": phones, "phone_primary": phones[0] if phones else None, "emails": [], "email_primary": None, "website": website, "vk_url": socials.get("vk_url"), "telegram_url": socials.get("telegram_url"), "instagram_url": socials.get("instagram_url"), "youtube_url": socials.get("youtube_url"), "address": address, "city": city, "region": region, "district": district, "category": category, "reviews_count": parse_reviews_count(reviews_raw), "reviews_avg": parse_rating(rating_raw) or 0.0, "source": "yandex_maps", "source_url": href, } # ─────────────────────────────────────────────────────────────────────── # ПУБЛИЧНАЯ ФУНКЦИЯ — это её вызывает main.py # ─────────────────────────────────────────────────────────────────────── @browser( headless=True, chrome_executable_path="/home/aks1om/.cache/ms-playwright/chromium-1223/chrome-linux64/chrome", block_images_and_css=True, reuse_driver=True, wait_for_complete_page_load=False, remove_default_browser_check_argument=True, ) def parse_yandex_maps(driver: Driver, data: dict) -> list[dict]: """ Парсит 1 категорию в 1 городе. data: { "query": "кафе", # категория "city": "Москва", "district": "Митино", # опц. район внутри города (добавляется в query) "max_cards": 30, } Возвращает list[dict] под схему leads. """ query = data.get("query", "кафе") city = data.get("city", "Москва") district = (data.get("district") or "").strip() or None max_cards = data.get("max_cards", config.MAX_CARDS_PER_CATEGORY) city_cfg = config.CITIES.get(city) if not city_cfg: # Fallback: неизвестный город → используем "Москва и МО" как geo, # а сам город уходит в district (как часть поискового запроса). if district: logger.error( f"Город '{city}' не в config.CITIES, и при этом указан --district '{district}'. " f"Так нельзя — добавь '{city}' в config.CITIES с правильным yandex_id " f"либо убери --district. Доступные города: {list(config.CITIES.keys())}" ) return [] fallback_name = "Москва и МО" fallback_cfg = config.CITIES.get(fallback_name) if not fallback_cfg: logger.error( f"Город '{city}' не найден в config.CITIES и нет fallback '{fallback_name}'. " f"Доступные: {list(config.CITIES.keys())}" ) return [] logger.warning( f"⚠️ Город '{city}' не в config.CITIES. Использую fallback: " f"'{fallback_name}' + '{city}' как район. Точность поиска ниже. " f"Для точного парсинга добавь '{city}' в config.CITIES (yandex_id с yandex.ru/maps)." ) district = city # неизвестный город теперь в районе city = fallback_name city_cfg = fallback_cfg yandex_id = city_cfg["yandex_id"] yandex_slug = city_cfg.get("yandex_slug", "moscow") region = city_cfg.get("region", city) # Если указан район — добавляем его к поисковому запросу. # Яндекс ищет в зоне района ("кафе Митино" в geo_id=213 → кафе в Митино) search_query = f"{query} {district}" if district else query url = f"https://yandex.ru/maps/{yandex_id}/{yandex_slug}/search/{search_query}/" region_label = f"{region}, район {district}" if district else region logger.info(f"\n→ Открываю: {url} (регион: {region_label})") driver.get(url) # Ждём появления карточек списка (адаптивно вместо фикс. sleep(5)). # Если капча/блок — карточек не будет, select вернёт None по таймауту, # и нижние проверки (bot_detected / пустой card_links) штатно остановят прогон. driver.select(SEL["card_in_list"], wait=12) # Анти-бот проверки detected = driver.get_bot_detected_by() if detected: logger.error(f"⚠️ Бот-детектор: {detected}. Останавливаемся.") return [] if "showcaptcha" in driver.current_url or driver.select(".CheckboxCaptcha"): logger.error("⚠️ CAPTCHA. Останавливаемся.") return [] # 1. Собрать ссылки на детальные карточки card_links = _collect_card_links(driver, max_cards) if not card_links: logger.warning("Карточек не найдено") return [] # 2. По каждой ссылке — переход и парсинг. # Подменяем category на поисковый запрос Яна (то что он искал) — # для CRM удобнее видеть "кафе" вместо "Ресторан, бар", а у HH/WB # такая же логика (category = query) для консистентности. leads: list[dict] = [] for idx, meta in enumerate(card_links, start=1): meta = {**meta, "category": query} logger.info(f"\n[{idx}/{len(card_links)}] {meta['href'][:80]}...") try: lead = _parse_card_detail(driver, meta, city=city, region=region, district=district) if lead: leads.append(lead) phones_str = ", ".join(lead["phones"]) if lead["phones"] else "—" logger.info(f" ✓ {lead['name']} | tel: {phones_str} | site: {lead.get('website') or '—'}") # Анти-бот пауза между карточками import random, time time.sleep(random.uniform(config.MIN_DELAY, config.MAX_DELAY)) except Exception as e: logger.exception(f" Ошибка парсинга {meta['href']}: {e}") logger.info(f"\n✅ Собрано лидов: {len(leads)} из {len(card_links)} карточек") return leads