"""HH employer-page parser — добирает website компании со страницы работодателя. Контекст: Сама выдача HH (`hh.ru/search/vacancy`) не показывает сайт работодателя — только название и employer_id. Но на странице компании (`hh.ru/employer/{id}`) работодатель часто указывает свой сайт. Эффект: После основного HH-парсинга → запуск этого enricher → у HH-лидов появляется поле `website` → обычный Tier 2 (`--enrich`) проходит по этим сайтам и собирает email/доп.телефоны. Workflow: HH parse → ЕГРЮЛ enrich → HH website enrich → Tier 2 enrich → email/phones Покрытие: На HH сайт компании указан у ~50-70% работодателей малого бизнеса. Селекторы: На странице employer'а сайт может быть в нескольких местах. Перебираем в порядке приоритета + fallback на body-text scan через regex. Дополнительно: Также извлекаем email со страницы employer'а (некоторые компании указывают HR-почту в описании) и доп.телефоны. """ import logging import random import re import time from typing import Optional from botasaurus.browser import browser, Driver from normalization import extract_emails_from_text, extract_phones_from_text logger = logging.getLogger(__name__) # ─────────────────────────────────────────────────────────────────────── # Селекторы сайта компании на странице employer'а # ─────────────────────────────────────────────────────────────────────── SITE_SELECTORS = [ '[data-qa="sidebar-company-site"] a', '[data-qa="sidebar-company-site"]', '[data-qa="employer-page__website"]', '[data-qa="employer-site-url"]', '[data-qa="company-site-url"]', 'a[data-qa*="company-site"]', 'a[data-qa*="employer-site"]', ] # Домены которые НЕ считаются сайтом компании (HH-инфраструктура, соцсети, # и обычные ссылки на партнёров/новости которые часто встречаются в описании). SKIP_DOMAINS = ( "hh.ru", "hhcdn.ru", "headhunter", "yastatic.net", "yandex.", "vk.com", "vk.ru", "t.me", "telegram.", "instagram.com", "facebook.com", "linkedin.com", "youtube.com", "youtu.be", "twitter.com", "x.com", "ok.ru", "rutube.ru", "google.com", "googleusercontent.com", "wikipedia.org", # Сервисы вакансий "superjob.ru", "rabota.ru", "avito.ru", ) def _is_company_site(url: str) -> bool: """True если URL похож на собственный сайт компании, а не на соцсеть/HH.""" if not url: return False u = url.lower().strip() # Должен быть http(s) и не пустой if not (u.startswith("http://") or u.startswith("https://")): return False for skip in SKIP_DOMAINS: if skip in u: return False return True def _extract_website_from_employer_page(driver: Driver) -> Optional[str]: """Перебор data-qa селекторов + fallback на body-text scan на http-ссылки. Возвращает первый найденный сайт компании, либо None. """ # 1. Структурные селекторы (приоритет — наиболее надёжные) for sel in SITE_SELECTORS: try: el = driver.select(sel, wait=None) if not el: continue # У ссылка в href; у других элементов — в тексте href = el.get_attribute("href") candidate = (href or el.text or "").strip() # Если в тексте — может быть без https:// в начале if candidate and not (candidate.startswith("http://") or candidate.startswith("https://")): if "." in candidate and " " not in candidate: candidate = "https://" + candidate if _is_company_site(candidate): return candidate except Exception as e: logger.debug(f" selector {sel} failed: {e}") continue # 2. Fallback: сканируем весь видимый текст страницы regex'ом на http-ссылки. # У employer'ов сайт часто упомянут в описании компании. try: text = driver.run_js("return document.body.innerText") or "" except Exception: text = "" if text: urls = re.findall(r"https?://[^\s)\"'>\]]+", text) for url in urls: url_clean = url.rstrip(".,;:") if _is_company_site(url_clean): return url_clean return None def _extract_extra_contacts(driver: Driver) -> tuple[list[str], list[str]]: """Извлечь email и доп.телефоны со страницы employer'а. Returns: (emails, phones) """ try: text = driver.run_js("return document.body.innerText") or "" except Exception: return [], [] if not text: return [], [] emails: list[str] = [] for e in extract_emails_from_text(text): el = e.lower() # Фильтр служебных HH адресов и явно технических if any(bad in el for bad in ( "@hh.ru", "@headhunter", "no-reply", "noreply", "postmaster", "support@hh", "feedback@hh", )): continue if el not in emails: emails.append(el) phones: list[str] = [] for p in extract_phones_from_text(text): if p and p not in phones: phones.append(p) return emails, phones # ─────────────────────────────────────────────────────────────────────── # Browser-парсер # ─────────────────────────────────────────────────────────────────────── @browser( headless=False, block_images=True, reuse_driver=True, ) def parse_hh_employer_pages(driver: Driver, data: dict) -> list[dict]: """Browser-функция: пройти по employer-page'ам, забрать website/emails/phones. data: { "leads": [{"lead_id": int, "employer_id": str, "name": str}, ...], "delay_min": float, # пауза между запросами "delay_max": float, } Возвращает список dict'ов: [{"lead_id": int, "website": str|None, "emails": [...], "phones": [...]}] """ leads = data.get("leads", []) delay_min = data.get("delay_min", 1.5) delay_max = data.get("delay_max", 4.0) results: list[dict] = [] for idx, lead in enumerate(leads, start=1): employer_id = lead.get("employer_id") lead_id = lead.get("lead_id") name = lead.get("name", "?") if not employer_id: continue url = f"https://hh.ru/employer/{employer_id}" logger.info(f"[{idx}/{len(leads)}] employer {employer_id} ({name[:40]})") try: driver.get(url) driver.sleep(1.5) # 2026-05-18: 2 → 1.5 (employer-страницы лёгкие) # Anti-bot check detected = driver.get_bot_detected_by() if detected: logger.error(f"⚠️ Бот-детектор: {detected}. Останавливаемся.") break # Сайт компании website = _extract_website_from_employer_page(driver) # Дополнительные email/телефоны со страницы employer'а emails, phones = _extract_extra_contacts(driver) res = { "lead_id": lead_id, "website": website, "emails": emails, "phones": phones, } results.append(res) # Краткий лог что нашли site_short = website[:50] if website else "—" extras = [] if emails: extras.append(f"+{len(emails)}email") if phones: extras.append(f"+{len(phones)}тел") extras_str = (" | " + " ".join(extras)) if extras else "" logger.info(f" site: {site_short}{extras_str}") except Exception as e: logger.warning(f" ⚠ ошибка: {e}") continue # Anti-rate-limit пауза if idx < len(leads): time.sleep(random.uniform(delay_min, delay_max)) logger.info(f"\n✅ HH employer-pages обработано: {len(results)} из {len(leads)}") return results