Aks
f78f35fb3f
Парсер лидов МБ РФ: Яндекс.Карты + HH.ru + обогащение DaData/ЕГРЮЛ/Rusprofile + Streamlit CRM. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
227 lines
8.8 KiB
Python
227 lines
8.8 KiB
Python
"""
|
|
Phase 0 — углублённый тест: клик по карточке, парсинг боковой панели.
|
|
|
|
ЦЕЛЬ:
|
|
1. Открыть карты → кликнуть первую карточку
|
|
2. Из боковой панели вытащить: телефон, адрес, сайт, отзывы, соцсети
|
|
3. Зафиксировать рабочие селекторы для боевого парсера
|
|
|
|
═══════════════════════════════════════════════════════════════════════
|
|
ЗАПУСК:
|
|
python test_yandex_v3.py
|
|
|
|
ОЖИДАЕМЫЙ РЕЗУЛЬТАТ:
|
|
- Из 3 карточек подряд извлекаются контактные поля
|
|
- Телефон в формате "+7 ..." виден в выводе
|
|
═══════════════════════════════════════════════════════════════════════
|
|
"""
|
|
|
|
from botasaurus.browser import browser, Driver
|
|
|
|
|
|
CARD_SELECTOR = ".search-snippet-view"
|
|
|
|
# Поля в боковой панели — пробуем несколько селекторов на каждое поле.
|
|
# Самые надёжные — стандартные HTML атрибуты (tel:, http://) и itemprop микроразметка.
|
|
FIELD_SELECTORS = {
|
|
"title": [
|
|
"h1.orgpage-header-view__header",
|
|
"h1",
|
|
".card-title-view__title",
|
|
],
|
|
"phone": [
|
|
# tel:-ссылка — самый надёжный способ
|
|
'a[href^="tel:"]',
|
|
".orgpage-phones-view__phone-number",
|
|
".card-phones-view__phone-number",
|
|
'[class*="phones-view__phone"]',
|
|
],
|
|
"address": [
|
|
".orgpage-header-view__address",
|
|
".business-contacts-view__address",
|
|
'[class*="address-view"]',
|
|
'[itemprop="address"]',
|
|
],
|
|
"website": [
|
|
# http(s)-ссылки кроме самого яндекса
|
|
'.business-urls-view__link',
|
|
'[class*="urls-view"] a',
|
|
],
|
|
"category": [
|
|
".orgpage-header-view__category",
|
|
".business-card-title-view__categories",
|
|
'[class*="header-view__category"]',
|
|
],
|
|
"rating": [
|
|
".business-rating-badge-view__rating-text",
|
|
'[class*="rating-badge-view__rating"]',
|
|
],
|
|
"reviews_count": [
|
|
".business-header-rating-view__text",
|
|
".card-section-header__title",
|
|
'[class*="reviews-count"]',
|
|
'[class*="rating-amount"]',
|
|
],
|
|
"working_hours": [
|
|
".business-working-status",
|
|
'[class*="working-status"]',
|
|
],
|
|
}
|
|
|
|
|
|
def find_first(element_or_driver, selectors):
|
|
"""Перебирает селекторы, возвращает Element первого найденного либо None."""
|
|
for sel in selectors:
|
|
result = element_or_driver.select(sel)
|
|
if result:
|
|
return result, sel
|
|
return None, None
|
|
|
|
|
|
def find_first_text(element_or_driver, selectors):
|
|
el, sel = find_first(element_or_driver, selectors)
|
|
if el and el.text:
|
|
return el.text.strip(), sel
|
|
return None, None
|
|
|
|
|
|
def parse_card_panel(driver: Driver) -> dict:
|
|
"""Парсит открытую боковую панель карточки.
|
|
Возвращает словарь со всеми полями + лог сработавших селекторов."""
|
|
result = {"fields": {}, "selectors_used": {}}
|
|
|
|
# Простые текстовые поля
|
|
for field, selectors in FIELD_SELECTORS.items():
|
|
if field in ("phone", "website"):
|
|
continue # они со ссылками — обработаем отдельно
|
|
text, sel = find_first_text(driver, selectors)
|
|
result["fields"][field] = text
|
|
result["selectors_used"][field] = sel
|
|
|
|
# Телефоны: собираем все a[href^="tel:"] и текстовые элементы
|
|
phones = []
|
|
tel_links = driver.select_all('a[href^="tel:"]')
|
|
for link in tel_links:
|
|
href = link.get_attribute("href")
|
|
if href:
|
|
phones.append(href.replace("tel:", "").strip())
|
|
if not phones:
|
|
# fallback: текст из специфических селекторов
|
|
for sel in FIELD_SELECTORS["phone"][1:]: # пропускаем tel: — уже пробовали
|
|
elements = driver.select_all(sel)
|
|
for el in elements:
|
|
if el.text:
|
|
phones.append(el.text.strip())
|
|
if phones:
|
|
result["selectors_used"]["phone"] = sel
|
|
break
|
|
else:
|
|
result["selectors_used"]["phone"] = 'a[href^="tel:"]'
|
|
result["fields"]["phones"] = list(dict.fromkeys(phones)) # уникальные, порядок сохраняется
|
|
|
|
# Сайт: ищем ссылку на сторонний домен
|
|
website = None
|
|
web_sel_used = None
|
|
for sel in FIELD_SELECTORS["website"]:
|
|
link = driver.select(sel)
|
|
if link:
|
|
href = link.get_attribute("href")
|
|
if href and "yandex" not in href.lower():
|
|
website = href
|
|
web_sel_used = sel
|
|
break
|
|
result["fields"]["website"] = website
|
|
result["selectors_used"]["website"] = web_sel_used
|
|
|
|
# Соцсети — что найдём
|
|
socials = {}
|
|
for net, pattern in [("vk", "vk.com"), ("telegram", "t.me"), ("instagram", "instagram.com"), ("youtube", "youtube.com")]:
|
|
link = driver.select(f'a[href*="{pattern}"]')
|
|
if link:
|
|
socials[net] = link.get_attribute("href")
|
|
result["fields"]["socials"] = socials
|
|
|
|
return result
|
|
|
|
|
|
@browser(
|
|
headless=False,
|
|
block_images=True,
|
|
reuse_driver=True,
|
|
)
|
|
def test_card_deep_dive(driver: Driver, data: dict):
|
|
query = data.get("query", "кафе")
|
|
city_id = data.get("city_id", 213)
|
|
cards_to_parse = data.get("cards_to_parse", 3)
|
|
|
|
url = f"https://yandex.ru/maps/{city_id}/moscow/search/{query}/"
|
|
print(f"\n→ Открываю: {url}")
|
|
driver.get(url)
|
|
driver.sleep(5)
|
|
|
|
# Проверки на блокировку
|
|
detected = driver.get_bot_detected_by()
|
|
if detected:
|
|
return {"status": "bot_detected", "detected_by": detected}
|
|
|
|
# Берём ссылки на детальные карточки заранее (чтобы не зависеть от состояния DOM)
|
|
print(f"\n🔗 Собираю ссылки первых {cards_to_parse} карточек...")
|
|
cards = driver.select_all(CARD_SELECTOR)[:cards_to_parse]
|
|
if not cards:
|
|
return {"status": "no_cards"}
|
|
|
|
card_links = []
|
|
for card in cards:
|
|
link_el = card.select("a")
|
|
if link_el:
|
|
href = link_el.get_attribute("href")
|
|
if href:
|
|
# Яндекс отдаёт относительные ссылки — приводим к абсолютным
|
|
if href.startswith("/"):
|
|
href = "https://yandex.ru" + href
|
|
card_links.append(href)
|
|
print(f" Получено ссылок: {len(card_links)}")
|
|
for i, link in enumerate(card_links, 1):
|
|
print(f" [{i}] {link[:100]}...")
|
|
|
|
# Идём по каждой ссылке и парсим
|
|
results = []
|
|
for idx, link in enumerate(card_links, 1):
|
|
print(f"\n══════ Карточка [{idx}/{len(card_links)}] ══════")
|
|
driver.get(link)
|
|
driver.sleep(4) # ждём рендер боковой панели
|
|
|
|
parsed = parse_card_panel(driver)
|
|
results.append(parsed)
|
|
|
|
print(" Поля:")
|
|
for k, v in parsed["fields"].items():
|
|
if isinstance(v, (list, dict)) and not v:
|
|
continue # не показываем пустые списки/словари
|
|
print(f" {k}: {v}")
|
|
|
|
# Итог: какие селекторы стабильно срабатывают
|
|
print("\n" + "=" * 60)
|
|
print("СТАТИСТИКА ПО СЕЛЕКТОРАМ (что сработало для каждого поля):")
|
|
selector_stats = {}
|
|
for r in results:
|
|
for field, sel in r["selectors_used"].items():
|
|
if sel:
|
|
selector_stats.setdefault(field, []).append(sel)
|
|
for field, sels in selector_stats.items():
|
|
unique_sels = set(sels)
|
|
print(f" {field}: {unique_sels} (срабатывало в {len(sels)}/{len(results)} карточках)")
|
|
print("=" * 60)
|
|
|
|
return {"status": "ok", "parsed": results, "selector_stats": selector_stats}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
result = test_card_deep_dive({
|
|
"query": "кафе",
|
|
"city_id": 213,
|
|
"cards_to_parse": 3,
|
|
})
|
|
|
|
print(f"\n→ Финальный статус: {result.get('status')}")
|