"""scrape_pipeline.py — Avito full orchestrator (Stage 2e + city sweep). Координирует 5 шагов одного pipeline run: 1. SEARCH — AvitoScraper().fetch_around(lat, lon, radius_m, pages=N) 2. SAVE — save_listings(db, lots) → (inserted, updated) 3. GROUP — dedupe house_url из lots → unique set 4. ENRICH_HOUSES — fetch_house_catalog + save_house_catalog_enrichment per unique house 5. ENRICH_DETAIL — top-N listings (detail_enriched_at IS NULL) → fetch_detail + save City sweep: run_avito_city_sweep — итерирует EKB_ANCHORS × pages_per_anchor, с cooperative cancel через scrape_runs.status и heartbeat update каждый anchor. Anti-bot hardening (2023-05-23): - Shared AsyncSession на весь sweep (один TLS handshake) - asyncio.sleep с random +-20% jitter между detail requests - AvitoBlockedError/AvitoRateLimitedError propagation → early abort - После 3 consecutive blocks — abort sweep + mark_banned (status='banned') Graceful degradation на каждом step: exception в одной house/listing не валит весь pipeline (кроме blocked exceptions — они abort весь sweep). IMV-фаза (финальная, после всех anchor'ов): если enrich_imv=True — для домов, тронутых в этом sweep'е (touched house_id из houses-enrichment), вызывается house_imv_backfill.process_houses_imv_batch. Ошибки per-house логируются и считаются, но не валят sweep. """ from __future__ import annotations import asyncio import logging import random from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields from datetime import date, timedelta from typing import Any from urllib.parse import urlparse from curl_cffi.requests import AsyncSession from sqlalchemy import text from sqlalchemy.orm import Session from app.core.config import settings from app.core.shutdown import shutdown_requested from app.services import proxy_pool, scrape_runs from app.services.scrapers.avito import AvitoScraper from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment from app.services.scrapers.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment from app.services.scrapers.base import ScrapedLot, save_listings from app.services.scrapers.browser_fetcher import BrowserFetcher from app.services.scrapers.yandex_realty import YandexRealtyScraper from app.services.yandex_price_history import record_yandex_price_history logger = logging.getLogger(__name__) async def _rotate_proxy_ip(*, reason: str, rotations_done: int, source: str = "avito") -> bool: """Вызвать changeip-ссылку mobileproxy и подождать settle (#1790/#1848). Provider-aware: выбирает rotate_url и max_rotations по параметру `source` (avito / cian / yandex). Fallback-цепочка для rotate_url: avito: avito_proxy_rotate_url cian: cian_proxy_rotate_url → avito_proxy_rotate_url yandex: yandex_proxy_rotate_url → avito_proxy_rotate_url Аналог AvitoScraper._rotate_ip, но на уровне pipeline — используется в enrichment-фазах где нет доступа к экземпляру scraper'а. Returns True при успешной смене IP, False если rotate_url не задан или ошибка. Логирует каждую ротацию (причина, порядковый номер, source, остаток лимита). """ if source == "cian": rotate_url = settings.cian_proxy_rotate_url or settings.avito_proxy_rotate_url max_rot = settings.cian_proxy_max_rotations elif source == "yandex": rotate_url = settings.yandex_proxy_rotate_url or settings.avito_proxy_rotate_url max_rot = settings.yandex_proxy_max_rotations else: # avito (default) — backward compat rotate_url = settings.avito_proxy_rotate_url max_rot = settings.avito_proxy_max_rotations if not rotate_url: return False sep = "&" if "?" in rotate_url else "?" # #1950: retry changeip-GET короткими попытками вместо одношотного 30s-timeout. # Если changeip-сервер завис на 30s → весь прогон ждёт зря + ротация считается # успешной (False возвращался). Теперь: proxy_rotate_attempts попыток по # proxy_rotate_attempt_timeout_s каждая; возвращаем True при первом успехе. _attempts = settings.proxy_rotate_attempts _attempt_timeout = settings.proxy_rotate_attempt_timeout_s for attempt in range(_attempts): try: async with AsyncSession(timeout=_attempt_timeout) as rot: resp = await rot.get(f"{rotate_url}{sep}format=json") new_ip: str = "" try: data = resp.json() new_ip = str(data.get("new_ip", "")) except Exception: pass await asyncio.sleep(settings.avito_proxy_rotate_settle_s) logger.info( "pipeline: IP rotated via changeip — source=%s reason=%s " "rotation=#%d attempt=%d/%d remaining=%d new_ip=%s", source, reason, rotations_done + 1, attempt + 1, _attempts, max_rot - rotations_done - 1, new_ip or "unknown", ) return True except Exception: logger.warning( "pipeline: IP rotation attempt %d/%d failed " "(source=%s reason=%s rotation=#%d)", attempt + 1, _attempts, source, reason, rotations_done + 1, exc_info=True, ) if attempt < _attempts - 1: await asyncio.sleep(1.0) logger.error( "pipeline: all %d changeip attempts failed " "(source=%s reason=%s rotation=#%d)", _attempts, source, reason, rotations_done + 1, ) return False # Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает # дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for # прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом. # Диапазон 180-300 с; 240 — разумный середина (4 мин > любой нормальный anchor). ANCHOR_TIMEOUT_SEC: int = 240 # Константы для расчёта watchdog-таймаута combos-режима Yandex sweep. # В combos-режиме единственный "anchor" делает num_combos × max_pages fetch'ей; # каждый fetch занимает request_delay_sec + сетевой overhead. ADDRESS_ENRICH_BUDGET_S # добавляет буфер на address-enrich фазу (per-listing HTTP + sleep × N листингов). # Пример: 30 combos × 3 pages × (9+12) + 300 ≈ 2190s (~37 мин) — надёжно укладывается # в окно 02:00-05:00 без риска перекрытия следующего sweep'а. _YANDEX_COMBOS_PER_FETCH_S: float = 12.0 # network + parse margin (секунды на страницу) _YANDEX_ADDRESS_ENRICH_BUDGET_S: float = 300.0 # бюджет на address-enrich фазу # Константы для расчёта watchdog-таймаута Avito city sweep. # Avito detail-fetch идёт через браузерный сервис с page.goto timeout 60s, и # anti-bot страницы нередко доходят до этого лимита. Фиксированный ANCHOR_TIMEOUT_SEC=240 # guillotine'ит anchor ещё в detail-фазе → SERP-счётчики теряются, run показывает # lots_fetched=0 при реально сохранённых листингах. Таймаут масштабируется: # _avito_anchor_timeout = ANCHOR_TIMEOUT_SEC # + detail_top_n × _AVITO_PER_DETAIL_S # + (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0) # Пример: detail_top_n=20 + houses → 240 + 1000 + 180 = 1420s/anchor (worst-case). # 5 anchor'ов × 1420s ≈ 2ч worst-case (нормальный прогон быстрее: не каждый detail # достигает 60s timeout). Окно лишь ограничивает старт следующего sweep'а — scheduler # ждёт финиша задачи и не запустит параллельный run. _AVITO_PER_DETAIL_S: float = 50.0 # секунд на один detail-fetch (incl. occasional 60s timeout) _AVITO_HOUSES_BUDGET_S: float = 180.0 # бюджет на house-enrich фазу # Fail-fast: если detail phase накапливает подряд N timeout/ошибок, прерываем её # (аналог consecutive_blocks для AvitoBlockedError в step 5). Срабатывает только при # явной деградации browser-сервиса, не мешает нормальным transient-ошибкам. _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5 # #1820: порог ПОСЛЕДОВАТЕЛЬНЫХ блоков (AvitoBlockedError/AvitoRateLimitedError) внутри # enrichment-фаз run_avito_pipeline. Один заблокированный фетч (house/detail) больше НЕ # роняет весь прогон — блок ловится пер-item (как обычная ошибка), и лишь N подряд # прерывают enrichment-фазу. Listings из SEARCH+SAVE сохраняются всегда. Зеркалит # паттерн avito_detail_backfill (max_consecutive_blocks). _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3 # #2160: константы для расчёта watchdog-таймаута Cian city sweep. При # USE_PROXY_POOL_BROWSER=true каждый SERP-фетч идёт через camoufox с relaunch при смене # прокси (page.goto timeout 60s + overhead) = 13-45s/страница, а якорь cian = 4 room-buckets # × pages_per_anchor SERP-фетчей + detail_top_n detail-фетчей + houses-enrich. Фиксированный # ANCHOR_TIMEOUT_SEC=240 гильотинит якорь ещё в SERP-фазе → save_listings не успевает, # lots_fetched=0, после N подряд → mark_banned. Таймаут масштабируется: # _cian_anchor_timeout = max( # ANCHOR_TIMEOUT_SEC, # _CIAN_ROOM_BUCKETS × pages_per_anchor × (request_delay_sec + _CIAN_PER_SERP_FETCH_S) # + detail_top_n × _CIAN_PER_DETAIL_S # + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0)) # Пример (дефолты): 4×3×(5+70) + 10×50 + 180 = 900 + 500 + 180 = 1580s/anchor (worst-case, # это watchdog от зависания, не ожидаемая длительность). _CIAN_ROOM_BUCKETS: int = 4 # fetch_around_multi_room итерирует ((1,),(2,),(3,),(4,)), # см. providers/cian/serp.py:215 _CIAN_PER_SERP_FETCH_S: float = 70.0 # browser fetch через пул: page.goto 60s + camoufox relaunch _CIAN_PER_DETAIL_S: float = 50.0 # секунд на один detail-fetch (как avito) _CIAN_HOUSES_BUDGET_S: float = 180.0 # бюджет на house-enrich фазу (как avito) def _cian_anchor_timeout_s( pages_per_anchor: int, request_delay_sec: float, detail_top_n: int, enrich_houses: bool, ) -> float: """Масштабируемый watchdog-таймаут одного Cian anchor'а (секунды). Чистая функция для тестируемости формулы (#2160). Никогда не опускается ниже ANCHOR_TIMEOUT_SEC (max-ветка): при малых параметрах фиксированный минимум остаётся. """ return max( float(ANCHOR_TIMEOUT_SEC), _CIAN_ROOM_BUCKETS * pages_per_anchor * (request_delay_sec + _CIAN_PER_SERP_FETCH_S) + detail_top_n * _CIAN_PER_DETAIL_S + (_CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0), ) # Default anchors ЕКБ — 5 точек покрытия города EKB_ANCHORS: list[tuple[float, float, str]] = [ (56.8400, 60.6050, "Центр"), (56.7950, 60.5300, "ЮЗ"), (56.8970, 60.6100, "Уралмаш"), (56.7700, 60.5500, "Академический"), (56.8650, 60.6200, "Пионерский"), ] _CHROME_HEADERS = { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", "Cache-Control": "max-age=0", "Sec-Fetch-Dest": "document", "Sec-Fetch-Mode": "navigate", "Sec-Fetch-Site": "none", "Sec-Fetch-User": "?1", "Upgrade-Insecure-Requests": "1", } def _avito_proxies() -> dict[str, str] | None: """#623/#806: mobile-proxy egress для shared-сессии sweep'а. КРИТИЧНО: sweep создаёт собственную shared AsyncSession и присваивает её `scraper._cffi`, перезатирая проксированную сессию из `AvitoScraper.__aenter__`. Через эту же сессию идёт и SERP, и detail-энричмент (fetch_detail). Без прокси detail-страницы летят с datacenter-IP → HTTP 429 → весь run mark_banned. Пусто (env не задан) → прямое подключение (dev/no-op). Читает settings.scraper_proxy_url (SCRAPER_PROXY_URL > AVITO_PROXY_URL fallback). """ url = settings.scraper_proxy_url return {"http": url, "https": url} if url else None def _mask_proxy_url(url: str) -> str: """Скрыть пароль в proxy-url для логов (scheme://user:***@host).""" if "@" not in url or "//" not in url: return url scheme, rest = url.split("//", 1) creds, host = rest.split("@", 1) if ":" in creds: user, _pwd = creds.split(":", 1) creds = f"{user}:***" return f"{scheme}//{creds}@{host}" def _pool_proxy_dict( db: Session, provider: str, *, run_id: int | None = None ) -> tuple[dict[str, str] | None, proxy_pool.ProxyLease | None]: """#2202: health-filtered lease из scrape_proxies вместо статического env-прокси. За флагом settings.use_proxy_pool_curl (ship-dark, default False — прод не меняется, пока флаг не включён явно). Флаг off ИЛИ пул пуст/ошибка acquire → fallback на статический _avito_proxies() (settings.scraper_proxy_url), поведение как раньше — сбор никогда не падает из-за проблем пула. run_id — id текущего scrape_runs-прогона (если он есть у вызывающего, см. run_avito_city_sweep/run_avito_newbuilding_sweep). Прокидывается в proxy_pool.acquire(), чтобы scrape_proxies.leased_by нёс реальный run_id для диагностики банов вместо NON_RUN_LEASE_MARKER. None (default) — прежнее поведение для вызывающих без run_id в scope (run_avito_pipeline standalone). Returns (proxies-dict для curl_cffi AsyncSession, lease | None). lease нужен вызывающему для release+mark_health в конце run (см. _release_pool_lease) и для последующей ротации на бан (см. _rotate_pool_proxy). """ # getattr-defensive: некоторые тестовые/adapter settings-объекты (SimpleNamespace) # не несут этого поля — трактуем отсутствие как off (то же ship-dark поведение). if not getattr(settings, "use_proxy_pool_curl", False): return _avito_proxies(), None try: lease = proxy_pool.acquire(db, provider, run_id=run_id) except Exception: logger.warning( "pipeline: proxy_pool.acquire(%s) failed — fallback to env proxy", provider, exc_info=True, ) lease = None if lease is None: logger.warning( "pipeline: proxy pool empty/unhealthy for provider=%s — fallback to env proxy", provider, ) return _avito_proxies(), None logger.info( "pipeline: using pool proxy id=%d provider=%s url=%s for this run", lease.id, provider, _mask_proxy_url(lease.url), ) return {"http": lease.url, "https": lease.url}, lease def _release_pool_lease(db: Session, lease: proxy_pool.ProxyLease | None, ok: bool) -> None: """Best-effort release+mark_health в конце run (#2202) — проблема пула не должна ронять уже собранные данные. lease=None (пул выключен/не выдал прокси) → no-op. """ if lease is None: return try: proxy_pool.mark_health(db, lease.id, ok) except Exception: logger.warning("pipeline: mark_health failed for lease id=%d", lease.id, exc_info=True) try: proxy_pool.release(db, lease.id) except Exception: logger.warning("pipeline: release failed for lease id=%d", lease.id, exc_info=True) async def _rotate_pool_proxy( db: Session, provider: str, session: AsyncSession, current_lease: proxy_pool.ProxyLease | None, ) -> proxy_pool.ProxyLease | None: """#2202: переключиться на ДРУГОЙ здоровый прокси из пула при бане (не changeip одного и того же IP). Помечает current_lease нездоровым (mark_health ok=False), затем acquire()-ит следующий свободный здоровый прокси, и ТОЛЬКО ПОСЛЕ этого освобождает current_lease (release). Порядок здесь принципиален (#2202 MAJOR-фикс, review-репро): release ДО acquire() убирает current_lease из leased_by-фильтра немедленно, и один бан поднимает consecutive_fails только до 1 (порог исключения — proxy_pool.MAX_CONSECUTIVE_FAILS=3) — только что забаненный прокси тут же снова проходит health-фильтр и acquire() может вернуть его же обратно (ORDER BY last_ok_at NULLS LAST, id — никак не гарантирует, что он не окажется первым). Пока current_lease НЕ освобождён, acquire()'s `leased_by IS NULL` условие гарантированно его исключает — никакой отдельный exclude_id-параметр не нужен. Успешная ротация мутирует session.proxies — curl_cffi AsyncSession читает self.proxies заново на каждый request (не кэширует на конструкторе), так что переприсвоение атрибута действует на следующий вызов без пересоздания сессии. Returns новый lease, или None если пул пуст/ошибка (вызывающий код в этом случае падает обратно на changeip-ротацию одного IP — см. _rotate_avito_proxy). current_lease освобождается в обоих случаях (успех/провал) — он подтверждённо нездоров, дальше эта функция за него не отвечает. """ if current_lease is not None: try: proxy_pool.mark_health(db, current_lease.id, ok=False) except Exception: logger.warning( "pipeline: mark_health(False) failed for lease id=%d", current_lease.id, exc_info=True, ) try: new_lease = proxy_pool.acquire(db, provider) except Exception: logger.warning( "pipeline: proxy_pool.acquire(%s) failed during rotation", provider, exc_info=True ) new_lease = None if current_lease is not None: try: proxy_pool.release(db, current_lease.id) except Exception: logger.warning( "pipeline: release failed for lease id=%d", current_lease.id, exc_info=True ) if new_lease is None: logger.warning( "pipeline: proxy pool exhausted for provider=%s during rotation — " "no healthy proxy to rotate to", provider, ) return None session.proxies = {"http": new_lease.url, "https": new_lease.url} logger.info( "pipeline: ROTATED to pool proxy id=%d provider=%s url=%s (previous lease id=%s)", new_lease.id, provider, _mask_proxy_url(new_lease.url), current_lease.id if current_lease is not None else None, ) return new_lease async def _rotate_avito_proxy( db: Session, session: AsyncSession | None, current_lease: proxy_pool.ProxyLease | None, *, reason: str, rotations_done: int, ) -> tuple[bool, proxy_pool.ProxyLease | None]: """#2202: единая точка ротации на бан для Avito. Сначала пробует переключить прокси ПУЛА (health-фильтр, гарантированно другой узел) — только когда settings.use_proxy_pool_curl включён И есть curl-сессия (session is not None, т.е. не browser-mode). Если пул выключен, пуст, или это browser-mode (session=None) — падает на старый changeip (смена exit-IP ОДНОГО прокси-аккаунта, как раньше). NB: пока делит общий rotations_done/avito_proxy_max_rotations бюджет с changeip-путём — раздельные бюджеты (пул своё, changeip своё) можно завести отдельным follow-up, если пул-ротации на практике окажутся частым событием. Returns (rotated, lease для последующих вызовов — обновлён при успешной пул-ротации, иначе тот же current_lease). """ if getattr(settings, "use_proxy_pool_curl", False) and session is not None: new_lease = await _rotate_pool_proxy(db, "avito", session, current_lease) if new_lease is not None: return True, new_lease logger.warning( "pipeline: pool rotation unavailable (reason=%s) — falling back to changeip", reason ) rotated = await _rotate_proxy_ip(reason=reason, rotations_done=rotations_done, source="avito") return rotated, current_lease # ── Result dataclasses ────────────────────────────────────────── @dataclass class PipelineCounters: """Counters одного pipeline run для логов/админки.""" lots_fetched: int = 0 lots_inserted: int = 0 lots_updated: int = 0 unique_houses: int = 0 houses_enriched: int = 0 houses_failed: int = 0 detail_attempted: int = 0 detail_enriched: int = 0 detail_failed: int = 0 errors: list[str] = field(default_factory=list) @dataclass class PipelineResult: """Полный результат full Avito pipeline.""" anchor_lat: float anchor_lon: float radius_m: int counters: PipelineCounters enrich_houses: bool enrich_detail_top_n: int touched_house_ids: set[int] = field(default_factory=set) # ── Main orchestrator ─────────────────────────────────────────── async def run_avito_pipeline( db: Session, lat: float, lon: float, radius_m: int = 1500, *, enrich_houses: bool = True, enrich_detail_top_n: int = 10, pages: int = 1, request_delay_sec: float | None = None, shared_session: AsyncSession | None = None, shared_browser: BrowserFetcher | None = None, ) -> PipelineResult: """Full Avito search → houses → detail enrichment pipeline. Steps: 1. SEARCH: AvitoScraper().fetch_around(lat, lon, radius_m, pages=pages) 2. SAVE: save_listings(db, lots) → (inserted, updated) 3. GROUP: dedupe house_url из новых lots → unique house path set 4. ENRICH_HOUSES: fetch_house_catalog + save per house (with sleep between) 5. ENRICH_DETAIL: top-N listings → fetch_detail + save (abort on 3 blocks) shared_session — если передан, используется без закрытия (lifecycle у вызывающего). shared_browser — если передан в browser-mode, используется без закрытия. Блокировки (AvitoBlockedError / AvitoRateLimitedError) в SEARCH-фазе (шаг 1) по-прежнему propagate наружу (без листингов прогон бессмыслен). В enrichment-фазах (houses шаг 4 + detail шаг 5 + house-detail шаг 5b) блок ловится ПЕР-ITEM как обычная ошибка (#1820): один заблокированный фетч НЕ роняет весь прогон. Лишь _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT блоков ПОДРЯД прерывают enrichment — но PipelineResult всё равно возвращается с уже собранными listings/обогащениями. """ counters = PipelineCounters() lots: list[ScrapedLot] = [] detail_delay = request_delay_sec if request_delay_sec is not None else 7.0 # #1820: общий счётчик ПОСЛЕДОВАТЕЛЬНЫХ блоков на все enrichment-фазы (4/5/5b). # Сбрасывается на любом успешном fetch; при достижении порога — ротация IP (#1790) # или abort если бюджет ротаций исчерпан. enrich_consecutive_blocks = 0 enrich_rotations_done = 0 # #1790: сколько раз уже сменили IP в этом run enrichment_aborted = False browser_mode = settings.scraper_fetch_mode == "browser" session: AsyncSession | None = None browser_fetcher: BrowserFetcher | None = None own_session = False own_browser = False # #2202: pool-lease текущей curl-сессии (None если browser-mode/pool off/shared # session переданный извне — тогда владелец проксей не мы, не трогаем). _lease: proxy_pool.ProxyLease | None = None scraper = AvitoScraper() if browser_mode: browser_fetcher = shared_browser if browser_fetcher is None: browser_fetcher = BrowserFetcher(source="avito") await browser_fetcher.__aenter__() own_browser = True scraper._browser = browser_fetcher else: own_session = shared_session is None if shared_session is not None: session = shared_session else: _proxies, _lease = _pool_proxy_dict(db, "avito") session = AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, proxies=_proxies, ) scraper._cffi = session try: # ── Step 1: search ────────────────────────────────────── try: lots = await scraper.fetch_around( lat, lon, radius_m, pages=pages, delay_override_sec=request_delay_sec, ) counters.lots_fetched = len(lots) logger.info( "pipeline:search anchor=(%.4f,%.4f) r=%dm fetched=%d", lat, lon, radius_m, len(lots), ) except (AvitoBlockedError, AvitoRateLimitedError): logger.error("pipeline:search BLOCKED by Avito at anchor=(%.4f,%.4f)", lat, lon) raise except Exception as e: logger.exception("pipeline:search failed") counters.errors.append(f"search: {e}") return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n) # ── Step 2: save listings ─────────────────────────────── if lots: try: counters.lots_inserted, counters.lots_updated = save_listings(db, lots) except Exception as e: logger.exception("pipeline:save_listings failed") counters.errors.append(f"save_listings: {e}") # ── Step 3: group by house ────────────────────────────── unique_house_paths: set[str] = set() for lot in lots: if lot.house_url: try: parsed = urlparse(lot.house_url) path = parsed.path if parsed.path else lot.house_url unique_house_paths.add(path) except Exception: continue counters.unique_houses = len(unique_house_paths) logger.info("pipeline:group_by_house unique=%d", counters.unique_houses) # ── Step 4: enrich houses с shared session + sleep ────── touched_house_ids: set[int] = set() if enrich_houses and unique_house_paths and not enrichment_aborted: house_paths_list = list(unique_house_paths) for idx, house_path in enumerate(house_paths_list): try: enrichment = await fetch_house_catalog( house_path, cffi_session=session, browser_fetcher=browser_fetcher ) house_counters = save_house_catalog_enrichment(db, enrichment) counters.houses_enriched += 1 enrich_consecutive_blocks = 0 hid = house_counters.get("house_id") if hid: touched_house_ids.add(int(hid)) except (AvitoBlockedError, AvitoRateLimitedError) as e: # #1820: блок ловим пер-item как обычную ошибку — НЕ роняем прогон. enrich_consecutive_blocks += 1 counters.houses_failed += 1 counters.errors.append(f"BLOCKED_HOUSE {house_path}: {e}") logger.warning( "pipeline:house BLOCKED at %s (consecutive=%d)", house_path, enrich_consecutive_blocks, ) if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: # #1790: Перед abort'ом пробуем сменить IP (changeip). # После ротации сбрасываем счётчик и продолжаем — НЕ abort. # Если ротации исчерпаны — abort enrichment (listings уже сохранены). if enrich_rotations_done < settings.avito_proxy_max_rotations: rotated, _lease = await _rotate_avito_proxy( db, session, _lease, reason=f"house consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) enrich_rotations_done += 1 if rotated: enrich_consecutive_blocks = 0 logger.info( "pipeline:enrich ROTATE — proxy changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, settings.avito_proxy_max_rotations, ) continue logger.error( "pipeline:enrich ABORT — %d consecutive blocks (IP rate-limited), " "listings preserved", enrich_consecutive_blocks, ) counters.errors.append( f"AVITO_BLOCKED — enrichment aborted after " f"{enrich_consecutive_blocks} consecutive blocks (house phase)" ) enrichment_aborted = True break except Exception as e: logger.warning("pipeline:house_enrich failed for %s: %s", house_path, e) counters.houses_failed += 1 counters.errors.append(f"house {house_path}: {e}") # #1368: save_house_catalog_enrichment не делает внутренний rollback, # поэтому упавший промежуточный statement оставляет общий Session в # InFailedSqlTransaction. Без сброса — каскад «current transaction is # aborted» на все оставшиеся дома/listings этого якоря. try: db.rollback() except Exception: pass if idx < len(house_paths_list) - 1: await asyncio.sleep(detail_delay) # ── Step 5: enrich detail для top-N priority listings ─── if enrich_detail_top_n > 0 and not enrichment_aborted: priority_rows = ( db.execute( text(""" SELECT source_url FROM listings WHERE source = 'avito' AND source_url IS NOT NULL AND ( ( detail_enriched_at IS NULL AND price_rub > 0 AND ST_DWithin( geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius ) ) OR ( detail_enriched_at IS NULL AND scraped_at > NOW() - INTERVAL '2 hours' ) ) ORDER BY scraped_at DESC NULLS LAST LIMIT :limit """), { "lat": lat, "lon": lon, "radius": radius_m * 2, "limit": enrich_detail_top_n, }, ) .mappings() .all() ) detail_house_paths: set[str] = set() for idx, row in enumerate(priority_rows): source_url: str = row["source_url"] counters.detail_attempted += 1 try: item_url = ( urlparse(source_url).path if source_url.startswith("http") else source_url ) enrichment_detail = await fetch_detail( item_url, cffi_session=session, browser_fetcher=browser_fetcher ) if save_detail_enrichment(db, enrichment_detail): counters.detail_enriched += 1 # #871: house-link есть в detail SSR (в SERP-карточке — JS-only). # Копим catalog-пути для house-enrich ниже (Step 5b). if enrichment_detail.house_catalog_url: hp = urlparse(enrichment_detail.house_catalog_url).path if hp: detail_house_paths.add(hp) enrich_consecutive_blocks = 0 except (AvitoBlockedError, AvitoRateLimitedError) as e: # #1820: блок ловим пер-item, НЕ роняем прогон. Счётчик общий с house-фазой. enrich_consecutive_blocks += 1 counters.detail_failed += 1 counters.errors.append(f"BLOCKED_DETAIL #{idx}: {e}") logger.warning( "pipeline:detail BLOCKED #%d/%d (consecutive=%d): %s", idx + 1, len(priority_rows), enrich_consecutive_blocks, e, ) if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: # #1790: Перед abort'ом пробуем сменить IP (changeip). # После ротации сбрасываем счётчик — продолжаем detail-обход. # Если ротации исчерпаны — abort detail-фазы (listings уже сохранены). if enrich_rotations_done < settings.avito_proxy_max_rotations: rotated, _lease = await _rotate_avito_proxy( db, session, _lease, reason=f"detail consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) enrich_rotations_done += 1 if rotated: enrich_consecutive_blocks = 0 logger.info( "pipeline:enrich ROTATE — proxy changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, settings.avito_proxy_max_rotations, ) continue logger.error( "pipeline:detail ABORT — %d consecutive blocks (IP rate-limited), " "listings preserved", enrich_consecutive_blocks, ) counters.errors.append( f"AVITO_BLOCKED — enrichment aborted after " f"{idx + 1}/{len(priority_rows)} details (consecutive blocks)" ) # #1820: НЕ raise — прерываем только detail-фазу (и Step 5b), # PipelineResult возвращается с уже сохранёнными listings/houses. enrichment_aborted = True break except Exception as e: counters.detail_failed += 1 counters.errors.append(f"detail {source_url}: {e}") logger.warning("pipeline:detail_enrich failed for %s: %s", source_url, e) # #1386: НЕ сбрасываем consecutive_blocks здесь. Транзиентная # сетевая/parse-ошибка (5xx, timeout, connection reset) — правдоподобный # побочный эффект активного IP-блока (Avito не отдаёт 403/429 единообразно). # Сброс маскировал бы продолжающийся блок (block, block, timeout→reset, ...) # и не давал бы сработать early-abort на 3 consecutive blocks. # Сброс на успехе (см. выше) остаётся. # #1368: save_detail_enrichment без внутреннего rollback мог оставить # общий Session в aborted-state — сбрасываем, иначе каскад на оставшиеся # detail-строки и Step 5b. try: db.rollback() except Exception: pass if idx < len(priority_rows) - 1: jitter = random.uniform(0.8, 1.2) await asyncio.sleep(detail_delay * jitter) # ── Step 5b: enrich houses, найденные через detail-страницы (#871) ── # SERP-карточки больше не несут house-link (JS-render) → дома берём из # detail SSR. Дедуп против обработанных в Step 4 (unique_house_paths). new_house_paths = detail_house_paths - unique_house_paths if enrich_houses and new_house_paths and not enrichment_aborted: counters.unique_houses += len(new_house_paths) nh_list = list(new_house_paths) for h_idx, house_path in enumerate(nh_list): try: enrichment = await fetch_house_catalog( house_path, cffi_session=session, browser_fetcher=browser_fetcher ) house_counters = save_house_catalog_enrichment(db, enrichment) counters.houses_enriched += 1 enrich_consecutive_blocks = 0 hid = house_counters.get("house_id") if hid: touched_house_ids.add(int(hid)) except (AvitoBlockedError, AvitoRateLimitedError) as e: # #1820: блок пер-item, общий счётчик; не роняем прогон. enrich_consecutive_blocks += 1 counters.houses_failed += 1 counters.errors.append(f"BLOCKED_HOUSE(detail) {house_path}: {e}") logger.warning( "pipeline:house(detail) BLOCKED at %s (consecutive=%d)", house_path, enrich_consecutive_blocks, ) if enrich_consecutive_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: # #1790: попытка ротации перед abort'ом (общий бюджет run). if enrich_rotations_done < settings.avito_proxy_max_rotations: rotated, _lease = await _rotate_avito_proxy( db, session, _lease, reason=f"house-detail consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) enrich_rotations_done += 1 if rotated: enrich_consecutive_blocks = 0 logger.info( "pipeline:enrich ROTATE — proxy changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, settings.avito_proxy_max_rotations, ) continue logger.error( "pipeline:enrich ABORT — %d consecutive blocks " "(IP rate-limited), listings preserved", enrich_consecutive_blocks, ) counters.errors.append( f"AVITO_BLOCKED — enrichment aborted after " f"{enrich_consecutive_blocks} consecutive blocks " f"(house-detail phase)" ) enrichment_aborted = True break except Exception as e: logger.warning( "pipeline:house(detail) enrich failed for %s: %s", house_path, e ) counters.houses_failed += 1 counters.errors.append(f"house(detail) {house_path}: {e}") # #1368: см. Step 4 — сбрасываем aborted-state общего Session, # иначе каскад на оставшиеся дома detail-фазы. try: db.rollback() except Exception: pass if h_idx < len(nh_list) - 1: await asyncio.sleep(detail_delay) logger.info( "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) " "houses=%d/%d detail=%d/%d touched_house_ids=%d errors=%d", lat, lon, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, counters.houses_enriched, counters.unique_houses, counters.detail_enriched, counters.detail_attempted, len(touched_house_ids), len(counters.errors), ) return PipelineResult( lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n, touched_house_ids=touched_house_ids, ) finally: if own_session and session is not None: await session.close() if own_browser and browser_fetcher is not None: await browser_fetcher.__aexit__(None, None, None) if own_session: _release_pool_lease(db, _lease, ok=not enrichment_aborted) @dataclass class CitySweepCounters: """Aggregate counters для full city sweep run.""" anchors_total: int = 0 anchors_done: int = 0 lots_fetched: int = 0 lots_inserted: int = 0 lots_updated: int = 0 unique_houses: int = 0 houses_enriched: int = 0 houses_failed: int = 0 detail_attempted: int = 0 detail_enriched: int = 0 detail_failed: int = 0 imv_attempted: int = 0 imv_enriched: int = 0 imv_failed: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} async def run_avito_city_sweep( db: Session, *, run_id: int, anchors: list[tuple[float, float, str]] | None = None, radius_m: int = 1500, pages_per_anchor: int = 3, enrich_houses: bool = True, detail_top_n: int = 10, request_delay_sec: float = 7.0, enrich_imv: bool = True, ) -> CitySweepCounters: """Full city sweep: iterate anchors × pages → save → enrich houses + detail → IMV. - Single shared AsyncSession на весь sweep (один TLS fingerprint) - AvitoBlockedError/AvitoRateLimitedError → ABORT sweep + mark_banned (status='banned') - Прочие errors per-anchor логируются, не валят весь sweep - После всех anchor'ов: если enrich_imv=True — IMV-оценка тронутых домов (cooperative cancel + per-house graceful error handling) SERP-счётчики (lots_fetched/inserted/updated) записываются в sweep-level counters НЕМЕДЛЕННО после save_listings — до начала detail/houses фазы. Даже если anchor превышает watchdog-таймаут в detail-фазе, SERP-результаты не теряются. """ from app.services.house_imv_backfill import process_houses_imv_batch _anchors = anchors if anchors is not None else EKB_ANCHORS counters = CitySweepCounters(anchors_total=len(_anchors)) all_touched_house_ids: set[int] = set() # Масштабируемый watchdog-таймаут для Avito anchor'а. В отличие от Yandex/Cian, # Avito detail-fetch идёт через browser service с page.goto timeout 60s. При # detail_top_n=20 detail-фаза может занять до N×60s — фиксированный ANCHOR_TIMEOUT_SEC # обрезает её раньше SERP, теряя счётчики. Масштабирование по detail_top_n + houses # гарантирует, что SERP+save всегда успевает до watchdog. _avito_anchor_timeout = ( float(ANCHOR_TIMEOUT_SEC) + detail_top_n * _AVITO_PER_DETAIL_S + (_AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0) ) logger.info( "city-sweep run_id=%d: avito anchor_timeout=%.0fs " "(base=%d + detail_top_n=%d×%.0fs + houses=%.0fs)", run_id, _avito_anchor_timeout, ANCHOR_TIMEOUT_SEC, detail_top_n, _AVITO_PER_DETAIL_S, _AVITO_HOUSES_BUDGET_S if enrich_houses else 0.0, ) browser_mode = settings.scraper_fetch_mode == "browser" # #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам). sweep_rotations_done = 0 # #2202: pool-lease текущей curl-сессии (None в browser-mode/pool off). _lease: proxy_pool.ProxyLease | None = None async with AsyncExitStack() as stack: session: AsyncSession | None = None shared_bf: BrowserFetcher | None = None if browser_mode: shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito")) else: _proxies, _lease = _pool_proxy_dict(db, "avito", run_id=run_id) session = await stack.enter_async_context( AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, proxies=_proxies, ) ) try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): if scrape_runs.is_cancelled(db, run_id): logger.info( "city-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", run_id, idx, len(_anchors), name, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters elif shutdown_requested(): # Кооперативный SIGTERM-drain (#1182 Phase 3a): останавливаемся # на границе anchor'а и финализируем run как mark_done(partial) — # это НЕ user-cancel, поэтому НЕ mark_cancelled. Дальнейшие anchor'ы # и IMV-фаза не выполняются. logger.info( "city-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", run_id, idx, len(_anchors), name, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters logger.info( "city-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", run_id, idx, len(_anchors), name, lat, lon, ) # Capture loop variables in default args (B023): prevents stale binding # if the coroutine is scheduled after the loop variable changes. _a_lat, _a_lon, _a_name = lat, lon, name async def _avito_anchor_phases( _lat: float = _a_lat, _lon: float = _a_lon, _name: str = _a_name, ) -> None: """Все фазы одного Avito anchor'а (SERP+save → houses → detail). Обновляет sweep-level counters и all_touched_house_ids через nonlocal НЕМЕДЛЕННО после SERP+save (до detail-фазы). Это гарантирует, что TimeoutError из wait_for теряет только detail-счётчики, но не SERP. """ nonlocal all_touched_house_ids, sweep_rotations_done, _lease # ── Phase 1+2: SERP + save ───────────────────────────────── scraper = AvitoScraper() if browser_mode: scraper._browser = shared_bf # Shared-browser режим: _cffi=None → curl_cffi-fallback на # firewall пропускается (by design). Логируем для observability. logger.debug("avito pipeline-mode: shared browser only, no cffi fallback") else: scraper._cffi = session try: anchor_lots: list[ScrapedLot] = await scraper.fetch_around( _lat, _lon, radius_m, pages=pages_per_anchor, delay_override_sec=request_delay_sec, ) except (AvitoBlockedError, AvitoRateLimitedError): logger.error( "city-sweep run_id=%d: SERP BLOCKED at anchor (%s, %.4f, %.4f)", run_id, _name, _lat, _lon, ) raise counters.lots_fetched += len(anchor_lots) if anchor_lots: try: ins, upd = save_listings(db, anchor_lots) counters.lots_inserted += ins counters.lots_updated += upd except Exception as save_exc: logger.exception( "city-sweep run_id=%d: save_listings failed at anchor %s: %s", run_id, _name, save_exc, ) counters.errors_count += 1 logger.info( "city-sweep run_id=%d anchor %s: SERP fetched=%d ins=%d upd=%d", run_id, _name, len(anchor_lots), counters.lots_inserted, counters.lots_updated, ) # SERP-счётчики зафиксированы в sweep-level counters. # Дальнейший TimeoutError из wait_for их не сотрёт. # ── Phase 3: group by house ──────────────────────────────── unique_house_paths: set[str] = set() for lot in anchor_lots: if lot.house_url: try: parsed = urlparse(lot.house_url) path = parsed.path if parsed.path else lot.house_url unique_house_paths.add(path) except Exception: continue if unique_house_paths: counters.unique_houses += len(unique_house_paths) # ── Phase 4: enrich houses ───────────────────────────────── touched_house_ids: set[int] = set() sweep_consecutive_house_blocks = 0 # #1790: per-anchor счётчик if enrich_houses and unique_house_paths: house_paths_list = list(unique_house_paths) h_idx = 0 while h_idx < len(house_paths_list): house_path = house_paths_list[h_idx] try: enrichment = await fetch_house_catalog( house_path, cffi_session=session, browser_fetcher=shared_bf, ) hc = save_house_catalog_enrichment(db, enrichment) counters.houses_enriched += 1 sweep_consecutive_house_blocks = 0 hid = hc.get("house_id") if hid: touched_house_ids.add(int(hid)) if h_idx < len(house_paths_list) - 1: await asyncio.sleep(request_delay_sec) h_idx += 1 except (AvitoBlockedError, AvitoRateLimitedError) as be: # #1790: пробуем сменить IP перед re-raise. sweep_consecutive_house_blocks += 1 counters.houses_failed += 1 counters.errors_count += 1 logger.warning( "city-sweep run_id=%d: house BLOCKED at %s " "(consecutive=%d) — %s", run_id, house_path, sweep_consecutive_house_blocks, be, ) if ( sweep_consecutive_house_blocks >= _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT and sweep_rotations_done < settings.avito_proxy_max_rotations ): rotated, _lease = await _rotate_avito_proxy( db, session, _lease, reason=( f"city-sweep house " f"consecutive={sweep_consecutive_house_blocks}" ), rotations_done=sweep_rotations_done, ) sweep_rotations_done += 1 if rotated: sweep_consecutive_house_blocks = 0 logger.info( "city-sweep run_id=%d: house ROTATE — " "proxy changed, reset consecutive " "(rotation #%d/%d)", run_id, sweep_rotations_done, settings.avito_proxy_max_rotations, ) # Не инкрементируем h_idx — retry текущего дома continue elif sweep_consecutive_house_blocks >= 3: logger.error( "city-sweep run_id=%d: house ABORT — " "%d consecutive blocks, propagating", run_id, sweep_consecutive_house_blocks, ) raise h_idx += 1 except Exception as he: logger.warning( "city-sweep run_id=%d: house_enrich failed %s: %s", run_id, house_path, he, ) counters.houses_failed += 1 counters.errors_count += 1 try: db.rollback() except Exception: pass h_idx += 1 # ── Phase 5: enrich detail (top-N) ──────────────────────── if detail_top_n > 0 and anchor_lots: priority_rows = ( db.execute( text(""" SELECT source_url FROM listings WHERE source = 'avito' AND source_url IS NOT NULL AND ( ( detail_enriched_at IS NULL AND price_rub > 0 AND ST_DWithin( geom::geography, ST_MakePoint(:lon, :lat)::geography, :radius ) ) OR ( detail_enriched_at IS NULL AND scraped_at > NOW() - INTERVAL '2 hours' ) ) ORDER BY scraped_at DESC NULLS LAST LIMIT :limit """), { "lat": _lat, "lon": _lon, "radius": radius_m * 2, "limit": detail_top_n, }, ) .mappings() .all() ) consecutive_blocks = 0 consecutive_timeouts = 0 detail_house_paths: set[str] = set() for d_idx, row in enumerate(priority_rows): source_url: str = row["source_url"] counters.detail_attempted += 1 try: item_url = ( urlparse(source_url).path if source_url.startswith("http") else source_url ) enrichment_detail = await fetch_detail( item_url, cffi_session=session, browser_fetcher=shared_bf, ) if save_detail_enrichment(db, enrichment_detail): counters.detail_enriched += 1 if enrichment_detail.house_catalog_url: hp = urlparse(enrichment_detail.house_catalog_url).path if hp: detail_house_paths.add(hp) consecutive_blocks = 0 consecutive_timeouts = 0 except (AvitoBlockedError, AvitoRateLimitedError) as be: consecutive_blocks += 1 counters.detail_failed += 1 counters.errors_count += 1 logger.warning( "city-sweep run_id=%d: detail BLOCKED #%d/%d " "(consecutive=%d): %s", run_id, d_idx + 1, len(priority_rows), consecutive_blocks, be, ) if consecutive_blocks >= 3: # #1790: попытка ротации перед abort'ом. if sweep_rotations_done < settings.avito_proxy_max_rotations: rotated, _lease = await _rotate_avito_proxy( db, session, _lease, reason=( f"city-sweep detail " f"consecutive={consecutive_blocks}" ), rotations_done=sweep_rotations_done, ) sweep_rotations_done += 1 if rotated: consecutive_blocks = 0 logger.info( "city-sweep run_id=%d: detail ROTATE — " "proxy changed, reset consecutive " "(rotation #%d/%d)", run_id, sweep_rotations_done, settings.avito_proxy_max_rotations, ) continue logger.error( "city-sweep run_id=%d: detail ABORT — " "%d consecutive blocks (IP rate-limited)", run_id, consecutive_blocks, ) raise except Exception as de: counters.detail_failed += 1 counters.errors_count += 1 consecutive_timeouts += 1 logger.warning( "city-sweep run_id=%d: detail failed #%d/%d " "(consecutive_timeouts=%d): %s", run_id, d_idx + 1, len(priority_rows), consecutive_timeouts, de, ) # (C) Fail-fast: браузер явно деградирует — прерываем # detail-фазу, не тратим время на оставшиеся запросы. if consecutive_timeouts >= _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: logger.error( "city-sweep run_id=%d: detail ABORT — " "%d consecutive timeouts/errors, browser degraded", run_id, consecutive_timeouts, ) break try: db.rollback() except Exception: pass if d_idx < len(priority_rows) - 1: jitter = random.uniform(0.8, 1.2) await asyncio.sleep(request_delay_sec * jitter) # ── Phase 5b: houses найденные через detail-страницы ── new_house_paths = detail_house_paths - unique_house_paths if enrich_houses and new_house_paths: counters.unique_houses += len(new_house_paths) nh_list = list(new_house_paths) nh_idx = 0 sweep_consecutive_detail_house_blocks = 0 # #1790 while nh_idx < len(nh_list): house_path = nh_list[nh_idx] try: enrichment = await fetch_house_catalog( house_path, cffi_session=session, browser_fetcher=shared_bf, ) hc = save_house_catalog_enrichment(db, enrichment) counters.houses_enriched += 1 sweep_consecutive_detail_house_blocks = 0 hid = hc.get("house_id") if hid: touched_house_ids.add(int(hid)) if nh_idx < len(nh_list) - 1: await asyncio.sleep(request_delay_sec) nh_idx += 1 except (AvitoBlockedError, AvitoRateLimitedError) as beh: # #1790: ротация перед re-raise. sweep_consecutive_detail_house_blocks += 1 counters.houses_failed += 1 counters.errors_count += 1 logger.warning( "city-sweep run_id=%d: house(detail) BLOCKED %s" " (consecutive=%d) — %s", run_id, house_path, sweep_consecutive_detail_house_blocks, beh, ) if ( sweep_consecutive_detail_house_blocks >= 3 and sweep_rotations_done < settings.avito_proxy_max_rotations ): rotated, _lease = await _rotate_avito_proxy( db, session, _lease, reason=( "city-sweep house(detail) " f"consecutive=" f"{sweep_consecutive_detail_house_blocks}" ), rotations_done=sweep_rotations_done, ) sweep_rotations_done += 1 if rotated: sweep_consecutive_detail_house_blocks = 0 logger.info( "city-sweep run_id=%d: house(detail) " "ROTATE — proxy changed (rotation #%d/%d)", run_id, sweep_rotations_done, settings.avito_proxy_max_rotations, ) continue elif sweep_consecutive_detail_house_blocks >= 3: logger.error( "city-sweep run_id=%d: house(detail) ABORT" " — %d consecutive blocks, propagating", run_id, sweep_consecutive_detail_house_blocks, ) raise nh_idx += 1 except Exception as he: logger.warning( "city-sweep run_id=%d: house(detail) failed %s: %s", run_id, house_path, he, ) counters.houses_failed += 1 counters.errors_count += 1 try: db.rollback() except Exception: pass nh_idx += 1 all_touched_house_ids.update(touched_house_ids) logger.info( "city-sweep run_id=%d anchor %s done: " "lots=%d houses=%d/%d detail=%d/%d touched=%d errors=%d", run_id, _name, len(anchor_lots), counters.houses_enriched, counters.unique_houses, counters.detail_enriched, counters.detail_attempted, len(touched_house_ids), counters.errors_count, ) try: await asyncio.wait_for(_avito_anchor_phases(), timeout=_avito_anchor_timeout) except TimeoutError: logger.warning( "city-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " "timed out after %.0fs — SERP counters preserved, " "detail phase incomplete", run_id, idx, len(_anchors), name, lat, lon, _avito_anchor_timeout, ) counters.errors_count += 1 except (AvitoBlockedError, AvitoRateLimitedError) as e: logger.error( "city-sweep run_id=%d ABORT at anchor #%d/%d (%s) — blocked: %s", run_id, idx, len(_anchors), name, e, ) counters.errors_count += 1 counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # #1950: если SERP уже собрал лоты и заблокировало только detail/houses, # ставим 'done' (не 'banned') — partial intake сохранён. # За флагом avito_serp_ok_not_banned (default True). _serp_ok = (counters.lots_inserted + counters.lots_updated) > 0 if settings.avito_serp_ok_not_banned and _serp_ok: _note = ( f"detail enrichment aborted ({e}) но SERP intake завершён: " f"{counters.lots_inserted} ins {counters.lots_updated} upd" ) logger.warning( "city-sweep run_id=%d: SERP OK (ins=%d upd=%d), " "detail blocked — DONE not banned. %s", run_id, counters.lots_inserted, counters.lots_updated, _note, ) scrape_runs.mark_done( db, run_id, {**counters.to_dict(), "enrichment_abort_note": _note}, # type: ignore[arg-type] ) else: scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict()) return counters except Exception: logger.exception("city-sweep run_id=%d: anchor %s failed", run_id, name) counters.errors_count += 1 counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # ── IMV-фаза: финальный обход тронутых домов ────────── if enrich_imv and all_touched_house_ids: if scrape_runs.is_cancelled(db, run_id): logger.info( "city-sweep run_id=%d: cancelled before IMV phase (%d houses skipped)", run_id, len(all_touched_house_ids), ) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters elif shutdown_requested(): # SIGTERM-drain до IMV-фазы: финализируем без дорогой IMV-оценки. logger.info( "city-sweep run_id=%d: SIGTERM-drain — skipping IMV phase " "(%d houses skipped)", run_id, len(all_touched_house_ids), ) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters logger.info( "city-sweep run_id=%d: IMV phase — %d touched houses", run_id, len(all_touched_house_ids), ) try: # Периодический heartbeat внутри длинной IMV-фазы (#1363): # без него десятки/сотни домов × request_delay_sec проходят # без update_heartbeat, и reap_zombies помечает живой run # 'zombie' → последующий mark_done становится no-op (дубль-sweep). def _imv_heartbeat() -> None: scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) imv_result = await process_houses_imv_batch( db, all_touched_house_ids, request_delay_sec=request_delay_sec, heartbeat=_imv_heartbeat, ) counters.imv_attempted += imv_result.checked counters.imv_enriched += imv_result.saved counters.imv_failed += imv_result.errors counters.errors_count += imv_result.errors scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) logger.info( "city-sweep run_id=%d: IMV phase done — attempted=%d enriched=%d failed=%d", run_id, imv_result.checked, imv_result.saved, imv_result.errors, ) except Exception as exc: logger.error( "city-sweep run_id=%d: IMV phase crashed — %r (sweep still done)", run_id, exc, ) counters.errors_count += 1 # #1521: IMV-фаза могла оставить Session в aborted-state # (необёрнутый eligibility SELECT / _mark_status на отравленной # сессии). update_heartbeat и mark_done — единственные финализаторы # БЕЗ defensive rollback, поэтому сбрасываем txn-state сами. Иначе # PendingRollbackError пролетает мимо внутреннего try во внешний # except (mark_failed) → успешный sweep ложно помечается 'failed'. try: db.rollback() except Exception: pass scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( "city-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " "houses=%d/%d detail=%d/%d imv=%d/%d errors=%d", run_id, counters.anchors_done, counters.anchors_total, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, counters.houses_enriched, counters.unique_houses, counters.detail_enriched, counters.detail_attempted, counters.imv_enriched, counters.imv_attempted, counters.errors_count, ) return counters except Exception as exc: logger.exception("city-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise finally: # #2202: best-effort release+mark_health — не должен ронять уже # собранный sweep. ok=False только если весь run завершился с ошибками. _release_pool_lease(db, _lease, ok=(counters.errors_count == 0)) # ── Avito newbuilding (novostroyka) citywide sweep ──────────────── @dataclass class NewbuildingSweepCounters: """Aggregate counters для Avito novostroyka citywide sweep run.""" lots_fetched: int = 0 lots_inserted: int = 0 lots_updated: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} async def run_avito_newbuilding_sweep( db: Session, *, run_id: int, pages: int = 20, request_delay_sec: float = 7.0, ) -> NewbuildingSweepCounters: """Citywide-обход ЕКБ-выборки только новостроек (novostroyka-filter) → save. В отличие от run_avito_city_sweep (anchor-based fetch_around + houses/detail/IMV), это простой citywide paginated-обход через AvitoScraper.fetch_newbuildings: dedicated novostroyka-SERP отдаёт 100% new-build карточки (listing_segment= "novostroyki", newbuilding_id/newbuilding_url заполнены парсером). Сохраняем через стандартный save_listings, ведём SERP-счётчики через scrape_runs. - Single shared AsyncSession/BrowserFetcher на весь sweep (один TLS fingerprint) - AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned') - is_cancelled-чек до старта SERP-фазы """ counters = NewbuildingSweepCounters() browser_mode = settings.scraper_fetch_mode == "browser" # #2202: pool-lease текущей curl-сессии (None в browser-mode/pool off). _lease: proxy_pool.ProxyLease | None = None async with AsyncExitStack() as stack: session: AsyncSession | None = None shared_bf: BrowserFetcher | None = None if browser_mode: shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito")) else: _proxies, _lease = _pool_proxy_dict(db, "avito", run_id=run_id) session = await stack.enter_async_context( AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, proxies=_proxies, ) ) try: if scrape_runs.is_cancelled(db, run_id): logger.info("nb-sweep run_id=%d: cancelled before SERP phase", run_id) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters elif shutdown_requested(): logger.info( "nb-sweep run_id=%d: SIGTERM-drain — stopping before SERP phase", run_id ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters scraper = AvitoScraper() if browser_mode: scraper._browser = shared_bf # Shared-browser режим: _cffi=None → curl_cffi-fallback на firewall # пропускается (by design). Логируем для observability. logger.debug("avito pipeline-mode: shared browser only, no cffi fallback") else: scraper._cffi = session try: lots: list[ScrapedLot] = await scraper.fetch_newbuildings( pages=pages, delay_override_sec=request_delay_sec, ) except (AvitoBlockedError, AvitoRateLimitedError) as e: logger.error("nb-sweep run_id=%d: SERP BLOCKED — %s", run_id, e) scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict()) return counters counters.lots_fetched += len(lots) if lots: try: ins, upd = save_listings(db, lots) counters.lots_inserted += ins counters.lots_updated += upd except Exception as save_exc: logger.exception( "nb-sweep run_id=%d: save_listings failed: %s", run_id, save_exc ) counters.errors_count += 1 try: db.rollback() except Exception: pass scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( "nb-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) errors=%d", run_id, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, counters.errors_count, ) return counters except Exception as exc: logger.exception("nb-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise finally: _release_pool_lease(db, _lease, ok=(counters.errors_count == 0)) # ── Yandex city sweep ─────────────────────────────────────────── @dataclass class YandexCitySweepCounters: """Aggregate counters для Yandex.Недвижимость city sweep run. Фазы: SERP (fetch_around_multi_room) → save_listings инкрементально → address-enrich. address_* — результат T10-обогащения (detail → полный адрес с номером дома). combos_skipped — сколько (room×price) combo пропущено из-за таймаутов gate-API. """ anchors_total: int = 0 anchors_done: int = 0 lots_fetched: int = 0 lots_inserted: int = 0 lots_updated: int = 0 price_history_rows: int = 0 combos_skipped: int = 0 address_attempted: int = 0 address_enriched: int = 0 address_failed: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} # Defensive abort threshold: у YandexRealtyScraper нет block/rate-exception классов # (fetch_around глотает ошибки и возвращает []). Чтобы не молотить Yandex впустую при # системном блоке/бане IP, прерываем sweep после N подряд anchor'ов, упавших с # исключением. NB: пустой результат (0 lots) — это НЕ ошибка, счётчик не растёт. YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 async def run_yandex_city_sweep( db: Session, *, run_id: int, anchors: list[tuple[float, float, str]] | None = None, radius_m: int = 25000, pages_per_anchor: int = 2, request_delay_sec: float | None = None, enrich_address: bool = True, rooms_list: list[str] | None = None, price_ranges: list[tuple[int | None, int | None]] | None = None, segments: list[str] | None = None, ) -> YandexCitySweepCounters: """Yandex.Недвижимость city sweep: rooms × price combos от центра ЕКБ → save → address-enrich. Вместо 5 географических anchor'ов итерирует ВСЕ combinations (room × price_range) из одной центральной точки ЕКБ (56.8400, 60.6050). radius_m=25000 охватывает весь город; price/room segmentation обходит Yandex SERP ~575-card cap без geo-шрапнели. Фазы (единственный центральный якорь): 1. SERP: YandexRealtyScraper.fetch_around_multi_room(lat, lon, radius_m, max_pages=pages_per_anchor, rooms_list=..., price_ranges=...) — combos mode. 2. SAVE: save_listings(db, lots, run_id=run_id). 3. ADDRESS-ENRICH (если enrich_address=True): для yandex-листингов без номера дома в address (address IS NOT NULL AND NOT (address ~ ',\\s*\\d+')) запрашиваем detail-страницу и извлекаем полный адрес из <title>. Параметр `anchors` сохранён для backward-compat тестов (anchors=[...]). В prod (anchors=None) всегда используется один EKB_CENTER anchor. Anti-bot / cancel семантика: - Cooperative cancel: is_cancelled(db, run_id) проверяется перед фазой SERP. - YandexRealtyScraper не propagate'ит block/rate exceptions (глотает → []). Ловим broad Exception; при N подряд — mark_banned + abort. - request_delay_sec: inter-request delay для address-enrich (поверх per-page sleep внутри scraper'а). Возвращает YandexCitySweepCounters. mark_done вызывается ВСЕГДА (кроме cancel / ban / fatal). """ from sqlalchemy import text as _text from app.services.scrapers.yandex_realty import DEFAULT_PRICE_RANGES, ROOM_PATH from app.services.yandex_address_backfill import ( _RE_HAS_HOUSE_NUMBER, _extract_address_from_title, ) # EKB center — единственный anchor для citywide combos sweep. # anchors-параметр принимается для backward-compat тестов. if anchors is not None: _anchors: list[tuple[float, float, str]] = anchors else: _anchors = [(56.8400, 60.6050, "Центр (combos)")] _rooms_list = rooms_list or list(ROOM_PATH.keys()) _price_ranges = price_ranges or DEFAULT_PRICE_RANGES # newFlat-сегменты: оба прохода (vtorichka + novostroyki) по одним combos. # Дефолт ["NO", "YES"] — оба; dedup по source_id span'ит оба прохода (один seen # внутри fetch_around_multi_room). Counters (lots_fetched/inserted/updated) # агрегируют оба прохода через len(anchor_lots). _segments = segments or ["NO", "YES"] counters = YandexCitySweepCounters(anchors_total=len(_anchors)) inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0 enrich_delay = request_delay_sec if request_delay_sec is not None else 3.0 _resolved_delay = request_delay_sec if request_delay_sec is not None else 9.0 consecutive_failures = 0 yandex_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None). # В combos-режиме один "anchor" выполняет num_segments × num_combos × max_pages # fetch'ей — каждый занимает примерно _resolved_delay + _YANDEX_COMBOS_PER_FETCH_S сек. # Для 2 segments × 30 combos × 3 pages × (9+12) + 300 ≈ 4080s (~68 мин). # Множитель len(_segments) масштабирует watchdog пропорционально числу проходов # (vtorichka + novostroyki ≈ ×2 wall-clock) — иначе второй проход гильотинится. # В explicit-anchor (тест/override) режиме оставляем ANCHOR_TIMEOUT_SEC (240s) — # там каждый anchor небольшой и watchdog работает как старый защитный барьер. _num_combos = len(_rooms_list) * len(_price_ranges) if anchors is None and _num_combos > 0: _sweep_timeout = max( ANCHOR_TIMEOUT_SEC, len(_segments) * _num_combos * pages_per_anchor * (_resolved_delay + _YANDEX_COMBOS_PER_FETCH_S) + _YANDEX_ADDRESS_ENRICH_BUDGET_S, ) else: # Explicit anchors (backward-compat тесты или ручной override) — legacy timeout. _sweep_timeout = ANCHOR_TIMEOUT_SEC # Прокси для address-enrich curl_cffi сессии (зеркало yandex_address_backfill). _proxy_url = settings.scraper_proxy_url _proxies = {"http": _proxy_url, "https": _proxy_url} if _proxy_url else None try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): # ── Cooperative cancel ─────────────────────────────────────────── if scrape_runs.is_cancelled(db, run_id): logger.info( "yandex-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", run_id, idx, len(_anchors), name, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters elif shutdown_requested(): # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial). logger.info( "yandex-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", run_id, idx, len(_anchors), name, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters logger.info( "yandex-sweep run_id=%d center-combos #%d/%d (%s, %.4f, %.4f) " "radius_m=%d rooms=%d price_ranges=%d", run_id, idx, len(_anchors), name, lat, lon, radius_m, len(_rooms_list), len(_price_ranges), ) # ── Per-anchor phases под watchdog-таймаутом ──────────────────── # Весь SERP + address-enrich одного anchor'а оборачиваем в wait_for. # TimeoutError → log + errors_count++ + continue (sweep не прерывается). anchor_lots: list[ScrapedLot] = [] _anchor_timed_out = False # Capture loop variables in default args to satisfy B023 (flake8/ruff): # без этого inner async def захватывает изменяемые переменные цикла # по ссылке → неправильные значения при asyncio scheduling. _lat, _lon, _name = lat, lon, name # _combo_lots_accumulator: per-anchor мутируемый контейнер для on_combo. # Захватывается по ссылке и в _yandex_anchor_phases и в _on_combo — # избегаем B023 (loop-variable capture) через явный bind в default-arg. _combo_lots_accumulator: list[ScrapedLot] = anchor_lots async def _yandex_anchor_phases( _a_lat: float = _lat, _a_lon: float = _lon, _a_name: str = _name, _al: list[ScrapedLot] = _combo_lots_accumulator, ) -> None: """Все фазы одного anchor'а (SERP + address-enrich). FIX: инкрементальный save per-combo — on_combo вызывается scraper'ом после каждого (room×price) combo, сохраняет порцию сразу в БД + обновляет heartbeat. anchor_lots накапливается (через _al) для address-enrich и price-history. """ nonlocal consecutive_failures, yandex_rotations_done # ── Phase 1+2: SERP + инкрементальный save per-combo ────── def _on_combo( combo_label: str, new_lots: list[ScrapedLot], _accumulator: list[ScrapedLot] = _al, ) -> None: """Callback: вызывается scraper'ом после каждого combo с новыми лотами. Сохраняет порцию новых лотов немедленно (анти-зомби: данные в БД до конца anchor'а) и обновляет heartbeat для мониторинга прогресса. """ _accumulator.extend(new_lots) counters.lots_fetched += len(new_lots) try: ins, upd = save_listings(db, new_lots, run_id=run_id) counters.lots_inserted += ins counters.lots_updated += upd except Exception as save_exc: counters.errors_count += 1 logger.warning( "yandex-sweep run_id=%d combo %s: save_listings failed: %s", run_id, combo_label, save_exc, ) try: db.rollback() except Exception: pass scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) logger.debug( "yandex-sweep run_id=%d combo %s: saved %d lots " "(ins=%d upd=%d total_fetched=%d)", run_id, combo_label, len(new_lots), counters.lots_inserted, counters.lots_updated, counters.lots_fetched, ) async with YandexRealtyScraper() as scraper: await scraper.fetch_around_multi_room( _a_lat, _a_lon, radius_m, max_pages=pages_per_anchor, rooms_list=_rooms_list, price_ranges=_price_ranges, segments=_segments, on_combo=_on_combo, ) # _al накоплен on_combo; save уже вызван per-combo. # Price-history из gate price.previous/trend (graceful — провал # истории НЕ должен ронять сейв листингов). if _al: try: counters.price_history_rows += record_yandex_price_history(db, _al) except Exception as ph_exc: logger.warning( "yandex-sweep run_id=%d: price-history failed: %s", run_id, ph_exc, ) try: db.rollback() except Exception: pass consecutive_failures = 0 logger.info( "yandex-sweep run_id=%d center-combos %s: fetched=%d ins=%d upd=%d", run_id, _a_name, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, ) # ── Phase 3: address-enrich ──────────────────────────────── if not (enrich_address and _al): return source_ids = [lot.source_id for lot in _al if lot.source_id] if not source_ids: return id_rows = ( db.execute( _text(""" SELECT id, source_id, address, source_url FROM listings WHERE source = 'yandex' AND source_id = ANY(CAST(:ids AS text[])) AND address IS NOT NULL AND NOT (address ~ ',\\s*\\d+') AND source_url IS NOT NULL """), {"ids": source_ids}, ) .mappings() .all() ) items = [dict(r) for r in id_rows] counters.address_attempted += len(items) if items: from curl_cffi.requests import AsyncSession as _AsyncSession # #1848: rotation tracking для Yandex address-enrich (stuck-proxy) _yandex_enrich_consec_fail = 0 _yandex_enrich_abort = 3 # abort address-enrich после N подряд async with _AsyncSession( impersonate="chrome120", timeout=30.0, proxies=_proxies, headers={"Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8"}, ) as enrich_session: eidx = 0 while eidx < len(items): item = items[eidx] lid: int = item["id"] old_addr: str = item["address"] url: str = item["source_url"] try: resp = await enrich_session.get(url, allow_redirects=True) if resp.status_code not in {200}: logger.warning( "yandex-sweep run_id=%d address-enrich:" " HTTP %d listing_id=%d", run_id, resp.status_code, lid, ) counters.address_failed += 1 # 503/429 = прокси заблокирован → считаем как failure if resp.status_code in {429, 503}: _yandex_enrich_consec_fail += 1 else: _yandex_enrich_consec_fail = 0 else: new_addr = _extract_address_from_title(resp.text) if ( new_addr is None or new_addr == old_addr or not _RE_HAS_HOUSE_NUMBER.search(new_addr) ): pass else: try: db.execute( _text(""" UPDATE listings SET address = :addr, geocode_tried_at = NULL WHERE id = CAST(:id AS bigint) """), {"addr": new_addr, "id": lid}, ) db.commit() counters.address_enriched += 1 logger.info( "yandex-sweep run_id=%d: " "address updated listing_id=%d " "%r -> %r", run_id, lid, old_addr, new_addr, ) except Exception as upd_exc: counters.address_failed += 1 counters.errors_count += 1 logger.warning( "yandex-sweep run_id=%d: " "address DB update failed " "listing_id=%d: %s", run_id, lid, upd_exc, ) try: db.rollback() except Exception: pass _yandex_enrich_consec_fail = 0 except Exception as fetch_exc: counters.address_failed += 1 _yandex_enrich_consec_fail += 1 logger.warning( "yandex-sweep run_id=%d: address-enrich " "fetch failed listing_id=%d " "(consecutive=%d): %s", run_id, lid, _yandex_enrich_consec_fail, fetch_exc, ) # Ротация Yandex-прокси при stuck/ban (#1848) if _yandex_enrich_consec_fail >= _yandex_enrich_abort: if yandex_rotations_done < settings.yandex_proxy_max_rotations: rotated = await _rotate_proxy_ip( reason=( f"yandex enrich consecutive=" f"{_yandex_enrich_consec_fail}" ), rotations_done=yandex_rotations_done, source="yandex", ) yandex_rotations_done += 1 if rotated: _yandex_enrich_consec_fail = 0 logger.info( "yandex-sweep run_id=%d: enrich ROTATE — " "IP changed, reset consecutive " "(rotation #%d/%d)", run_id, yandex_rotations_done, settings.yandex_proxy_max_rotations, ) # retry текущего item после ротации continue logger.error( "yandex-sweep run_id=%d: address-enrich ABORT — " "%d consecutive failures (proxy likely stuck/banned)", run_id, _yandex_enrich_consec_fail, ) break if eidx < len(items) - 1: await asyncio.sleep(enrich_delay) eidx += 1 logger.info( "yandex-sweep run_id=%d anchor %s: address enrich=%d/%d failed=%d", run_id, _a_name, counters.address_enriched, counters.address_attempted, counters.address_failed, ) try: await asyncio.wait_for(_yandex_anchor_phases(), timeout=_sweep_timeout) except TimeoutError: logger.warning( "yandex-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " "timed out after %.0fs — skipping", run_id, idx, len(_anchors), name, lat, lon, _sweep_timeout, ) counters.errors_count += 1 consecutive_failures += 1 _anchor_timed_out = True if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES: logger.error( "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " "%d consecutive failures (last: timeout): " "IP likely blocked or proxy hung", run_id, idx, len(_anchors), name, consecutive_failures, ) counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_banned( db, run_id, f"yandex sweep aborted: {consecutive_failures} consecutive " f"anchor failures (last: anchor timeout {_sweep_timeout:.0f}s)", counters.to_dict(), ) return counters counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) if idx < len(_anchors): await asyncio.sleep(inter_anchor_delay) continue except Exception as e: # YandexRealtyScraper не propagate'ит block/rate exceptions — ловим broad. logger.exception("yandex-sweep run_id=%d: anchor %s SERP failed", run_id, name) counters.errors_count += 1 consecutive_failures += 1 if consecutive_failures >= YANDEX_SWEEP_MAX_CONSECUTIVE_FAILURES: logger.error( "yandex-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " "%d consecutive failures (IP likely blocked): %s", run_id, idx, len(_anchors), name, consecutive_failures, e, ) counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_banned( db, run_id, f"yandex sweep aborted: {consecutive_failures} consecutive " f"anchor failures (last: {e})", counters.to_dict(), ) return counters counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # Пауза перед следующим anchor'ом даже при ошибке. if idx < len(_anchors): await asyncio.sleep(inter_anchor_delay) continue counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # Доп. пауза между anchor'ами (поверх per-page sleep внутри scraper'а). if idx < len(_anchors): await asyncio.sleep(inter_anchor_delay) scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( "yandex-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " "address=%d/%d failed=%d errors=%d", run_id, counters.anchors_done, counters.anchors_total, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, counters.address_enriched, counters.address_attempted, counters.address_failed, counters.errors_count, ) return counters except Exception as exc: logger.exception("yandex-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise # ── Cian city sweep (#860) ────────────────────────────────────────────────── @dataclass class CianCitySweepCounters: """Aggregate counters для Cian city sweep run. Поля симметричны CitySweepCounters, но без imv_* (у Cian нет IMV-фазы). detail_* — обогащение через cian_detail (вкл. price-history). houses_* — обогащение newbuilding/ЖК через cian_newbuilding. """ anchors_total: int = 0 anchors_done: int = 0 lots_fetched: int = 0 lots_dropped_secondary: int = 0 # вторичка, отброшенная при newbuilding_only=True lots_inserted: int = 0 lots_updated: int = 0 detail_attempted: int = 0 detail_enriched: int = 0 detail_failed: int = 0 houses_attempted: int = 0 houses_enriched: int = 0 houses_failed: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} # Defensive abort threshold: если N подряд anchor'ов упали с исключением # (не просто вернули пустой результат) — прерываем sweep. CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES = 3 def _mark_cian_sweep_abort( db: Session, run_id: int, counters: CianCitySweepCounters, reason: str, ) -> None: """#1949: честный финальный статус при defensive abort run_cian_city_sweep. Зеркалит avito_serp_ok_not_banned (#1950): если SERP-фаза уже сохранила лоты (lots_inserted + lots_updated > 0) до момента abort'а, основной сбор состоялся — 'banned' лишний шум в мониторинге. Ставим 'done' с заметкой о причине abort'а. Если лотов нет (SERP сам заблокирован/завис на первом же anchor'е) — 'banned' как раньше. За флагом settings.cian_sweep_ok_not_banned (default True). """ _serp_ok = (counters.lots_inserted + counters.lots_updated) > 0 if settings.cian_sweep_ok_not_banned and _serp_ok: logger.warning( "cian-sweep run_id=%d: SERP OK (ins=%d upd=%d) before abort — " "DONE not banned. %s", run_id, counters.lots_inserted, counters.lots_updated, reason, ) scrape_runs.mark_done( db, run_id, {**counters.to_dict(), "abort_note": reason}, # type: ignore[arg-type] ) else: scrape_runs.mark_banned(db, run_id, reason, counters.to_dict()) async def run_cian_city_sweep( db: Session, *, run_id: int, anchors: list[tuple[float, float, str]] | None = None, radius_m: int = 1500, pages_per_anchor: int = 3, request_delay_sec: float = 5.0, detail_top_n: int = 10, enrich_houses: bool = True, newbuilding_only: bool = True, ) -> CianCitySweepCounters: """Cian newbuilding city sweep: SERP → detail(+price-history) → newbuilding/houses. Симметрично run_avito_city_sweep. Использует EKB_ANCHORS (общий список). newbuilding_only (default True): в SERP-фазе сохраняются только лоты с listing_segment == "novostroyki" — вторичку отбрасываем, ею авторитетно владеет run_cian_full_load (exhaustive региональный сбор). Зеркало secondary_only в fetch_all_secondary, которое фильтрует обратное (!= "novostroyki"). DETAIL-фаза (обогащение cian-листингов без detail_enriched_at) и HOUSES-фаза (newbuilding-enrich) не затронуты фильтром. Фазы per anchor: 1. SERP: CianScraper.fetch_around_multi_room(lat, lon, radius_m, pages=pages_per_anchor) 2. FILTER (newbuilding_only): отбросить вторичку (listing_segment != "novostroyki") 3. SAVE: save_listings(db, lots, run_id=run_id) 4. DETAIL: top-N свежих cian-листингов без detail_enriched_at → fetch_detail(source_url) + save_detail_enrichment(db, listing_id, enrichment) (save_detail_enrichment уже пишет price_changes → offer_price_history) 5. HOUSES (если enrich_houses=True): для домов из этого anchor'а с cian_zhk_url → fetch_newbuilding(zhk_url) + save_newbuilding_enrichment(db, house_id, enr) Anti-bot: - CianScraper управляет своей curl_cffi-сессией (прокси уже внутри). - asyncio.sleep(request_delay_sec) между detail-запросами (jitter ±20%). - Defensive abort при CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES SERP-ошибок подряд → _mark_cian_sweep_abort (#1949): 'done' если SERP уже сохранил лоты, иначе 'banned'. Cooperative cancel: scrape_runs.is_cancelled(db, run_id) перед каждым anchor. mark_done вызывается ВСЕГДА (finally outer). """ from app.services.scrapers.cian import CianScraper from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment from app.services.scrapers.cian_newbuilding import ( fetch_newbuilding, save_newbuilding_enrichment, ) _anchors = anchors if anchors is not None else EKB_ANCHORS counters = CianCitySweepCounters(anchors_total=len(_anchors)) consecutive_failures = 0 cian_rotations_done = 0 # #1848: бюджет IP-ротаций на весь sweep # #2160: масштабируемый watchdog-таймаут для cian anchor'а. При proxy-pool browser каждый # SERP-фетч 13-45s (camoufox relaunch), фиксированный ANCHOR_TIMEOUT_SEC=240 гильотинит # якорь в SERP-фазе. См. _cian_anchor_timeout_s + константы выше. _cian_anchor_timeout = _cian_anchor_timeout_s( pages_per_anchor, request_delay_sec, detail_top_n, enrich_houses ) logger.info( "cian-sweep run_id=%d: anchor_timeout=%.0fs " "(base=%d, serp=%d×%d×(%.0f+%.0f)s + detail_top_n=%d×%.0fs + houses=%.0fs)", run_id, _cian_anchor_timeout, ANCHOR_TIMEOUT_SEC, _CIAN_ROOM_BUCKETS, pages_per_anchor, request_delay_sec, _CIAN_PER_SERP_FETCH_S, detail_top_n, _CIAN_PER_DETAIL_S, _CIAN_HOUSES_BUDGET_S if enrich_houses else 0.0, ) try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): # Cooperative cancel перед каждым anchor if scrape_runs.is_cancelled(db, run_id): logger.info( "cian-sweep run_id=%d: cancelled at anchor #%d/%d (%s)", run_id, idx, len(_anchors), name, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters elif shutdown_requested(): # SIGTERM-drain: чистый стоп на границе anchor'а → mark_done(partial). logger.info( "cian-sweep run_id=%d: SIGTERM-drain — stopping at anchor #%d/%d (%s)", run_id, idx, len(_anchors), name, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters logger.info( "cian-sweep run_id=%d anchor #%d/%d (%s, %.4f, %.4f)", run_id, idx, len(_anchors), name, lat, lon, ) # ── Per-anchor phases под watchdog-таймаутом ──────────────────── # Весь SERP + detail + houses одного anchor'а оборачиваем в wait_for. # TimeoutError → log + errors_count++ + continue (sweep не прерывается). anchor_lots: list[ScrapedLot] = [] # Capture loop variables in default args (B023): prevents stale binding # if the coroutine is scheduled after the loop variable changes. _c_lat, _c_lon, _c_name = lat, lon, name async def _cian_anchor_phases( _a_lat: float = _c_lat, _a_lon: float = _c_lon, _a_name: str = _c_name, ) -> None: """Все фазы одного cian anchor'а (SERP + detail + houses).""" nonlocal anchor_lots, consecutive_failures, cian_rotations_done from sqlalchemy import text as _text # ── Phase 1+2: SERP + save ───────────────────────────────── async with CianScraper() as scraper: anchor_lots = await scraper.fetch_around_multi_room( _a_lat, _a_lon, radius_m, pages=pages_per_anchor, ) counters.lots_fetched += len(anchor_lots) # ── Фильтр вторички (newbuilding_only) ───────────────────── # Зеркало secondary_only в fetch_all_secondary: тут оставляем # только новостройки — вторичкой авторитетно владеет run_cian_full_load. if newbuilding_only: _before = len(anchor_lots) anchor_lots = [ lot for lot in anchor_lots if lot.listing_segment == "novostroyki" ] counters.lots_dropped_secondary += _before - len(anchor_lots) if anchor_lots: inserted, updated = save_listings(db, anchor_lots, run_id=run_id) counters.lots_inserted += inserted counters.lots_updated += updated consecutive_failures = 0 logger.info( "cian-sweep run_id=%d anchor %s: SERP fetched=%d nb_kept=%d " "dropped_secondary=%d ins=%d upd=%d", run_id, _a_name, counters.lots_fetched, len(anchor_lots), counters.lots_dropped_secondary, counters.lots_inserted, counters.lots_updated, ) # ── Phase 3: detail enrichment (top-N) ──────────────────── if detail_top_n > 0: priority_rows = ( db.execute( _text(""" SELECT id, source_url FROM listings WHERE source = 'cian' AND source_url IS NOT NULL AND detail_enriched_at IS NULL AND scraped_at > NOW() - INTERVAL '2 hours' ORDER BY scraped_at DESC NULLS LAST LIMIT :lim """), {"lim": detail_top_n}, ) .mappings() .all() ) # #1848: Cian bans/stuck-proxy tracking + rotation (зеркало avito) _cian_detail_consec_failures = 0 _cian_detail_abort = 3 # abort detail-фазы после N подряд didx = 0 while didx < len(priority_rows): row = priority_rows[didx] listing_id: int = row["id"] source_url: str = row["source_url"] counters.detail_attempted += 1 try: enrichment = await fetch_detail(source_url) if enrichment is not None: save_detail_enrichment(db, listing_id, enrichment) counters.detail_enriched += 1 else: counters.detail_failed += 1 logger.warning( "cian-sweep run_id=%d: detail returned None for " "listing_id=%d url=%s", run_id, listing_id, source_url, ) _cian_detail_consec_failures = 0 except Exception as exc: counters.detail_failed += 1 counters.errors_count += 1 _cian_detail_consec_failures += 1 logger.warning( "cian-sweep run_id=%d: detail failed listing_id=%d " "(consecutive=%d): %s", run_id, listing_id, _cian_detail_consec_failures, exc, ) if _cian_detail_consec_failures >= _cian_detail_abort: # Попытка ротации Cian-прокси перед abort'ом (#1848) if cian_rotations_done < settings.cian_proxy_max_rotations: rotated = await _rotate_proxy_ip( reason=( f"cian detail consecutive=" f"{_cian_detail_consec_failures}" ), rotations_done=cian_rotations_done, source="cian", ) cian_rotations_done += 1 if rotated: _cian_detail_consec_failures = 0 logger.info( "cian-sweep run_id=%d: detail ROTATE — " "IP changed, reset consecutive " "(rotation #%d/%d)", run_id, cian_rotations_done, settings.cian_proxy_max_rotations, ) # retry текущего detail (не инкрементируем didx) continue logger.error( "cian-sweep run_id=%d: detail ABORT — " "%d consecutive failures (proxy likely banned/stuck)", run_id, _cian_detail_consec_failures, ) break if didx < len(priority_rows) - 1: jitter = random.uniform(0.8, 1.2) await asyncio.sleep(request_delay_sec * jitter) didx += 1 logger.info( "cian-sweep run_id=%d anchor %s: detail=%d/%d failed=%d", run_id, _a_name, counters.detail_enriched, counters.detail_attempted, counters.detail_failed, ) # ── Phase 4: newbuilding/houses enrichment ───────────────── if not (enrich_houses and anchor_lots): return cian_nb_ext_ids = { lot.house_ext_id for lot in anchor_lots if lot.house_source == "cian_newbuilding" and lot.house_ext_id } if not cian_nb_ext_ids: return nb_id_list = list(cian_nb_ext_ids) try: house_rows = ( db.execute( _text(""" SELECT h.id, h.cian_zhk_url FROM houses h JOIN house_sources hs ON hs.house_id = h.id WHERE hs.ext_source = 'cian_newbuilding' AND hs.ext_id = ANY(CAST(:ids AS text[])) AND h.cian_zhk_url IS NOT NULL """), {"ids": nb_id_list}, ) .mappings() .all() ) except Exception as exc: logger.exception( "cian-sweep run_id=%d anchor %s: houses DB query failed — " "skipping houses phase (%d ids): %s", run_id, _a_name, len(nb_id_list), exc, ) counters.houses_failed += len(nb_id_list) try: db.rollback() except Exception: pass return for hidx, hrow in enumerate(house_rows): house_id_val: int = hrow["id"] zhk_url: str = hrow["cian_zhk_url"] counters.houses_attempted += 1 try: nb_enrichment = await fetch_newbuilding(zhk_url) if nb_enrichment is not None: save_newbuilding_enrichment(db, house_id_val, nb_enrichment) counters.houses_enriched += 1 else: counters.houses_failed += 1 logger.warning( "cian-sweep run_id=%d: newbuilding returned None " "house_id=%d url=%s", run_id, house_id_val, zhk_url, ) except Exception as exc: counters.houses_failed += 1 counters.errors_count += 1 logger.warning( "cian-sweep run_id=%d: houses failed house_id=%d: %s", run_id, house_id_val, exc, ) try: db.rollback() except Exception: pass if hidx < len(house_rows) - 1: await asyncio.sleep(request_delay_sec) logger.info( "cian-sweep run_id=%d anchor %s: houses=%d/%d failed=%d", run_id, _a_name, counters.houses_enriched, counters.houses_attempted, counters.houses_failed, ) try: await asyncio.wait_for(_cian_anchor_phases(), timeout=_cian_anchor_timeout) except TimeoutError: logger.warning( "cian-sweep run_id=%d: anchor #%d/%d (%s, %.4f, %.4f) " "timed out after %ds — skipping", run_id, idx, len(_anchors), name, lat, lon, _cian_anchor_timeout, ) counters.errors_count += 1 consecutive_failures += 1 if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES: logger.error( "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " "%d consecutive failures (last: timeout): " "IP likely blocked or proxy hung", run_id, idx, len(_anchors), name, consecutive_failures, ) counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) _abort_reason = ( f"cian sweep aborted: {consecutive_failures} consecutive " f"anchor failures (last: anchor timeout {_cian_anchor_timeout:.0f}s)" ) _mark_cian_sweep_abort(db, run_id, counters, _abort_reason) return counters counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) continue except Exception as e: logger.exception("cian-sweep run_id=%d: anchor %s SERP failed", run_id, name) counters.errors_count += 1 consecutive_failures += 1 if consecutive_failures >= CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES: logger.error( "cian-sweep run_id=%d ABORT at anchor #%d/%d (%s) — " "%d consecutive SERP failures (IP likely blocked): %s", run_id, idx, len(_anchors), name, consecutive_failures, e, ) counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) _abort_reason = ( f"cian sweep aborted: {consecutive_failures} consecutive " f"anchor SERP failures (last: {e})" ) _mark_cian_sweep_abort(db, run_id, counters, _abort_reason) return counters counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) continue counters.anchors_done = idx scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # Пауза между anchor'ами (поверх per-request sleep внутри scraper'а) if idx < len(_anchors): await asyncio.sleep(request_delay_sec) scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( "cian-sweep run_id=%d done: anchors=%d/%d lots=%d (ins=%d/upd=%d) " "detail=%d/%d houses=%d/%d errors=%d", run_id, counters.anchors_done, counters.anchors_total, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, counters.detail_enriched, counters.detail_attempted, counters.houses_enriched, counters.houses_attempted, counters.errors_count, ) return counters except Exception as exc: logger.exception("cian-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise # ── Cian exhaustive full load (region-wide, без anchor'ов) ───────────────── @dataclass class CianFullLoadCounters: """Aggregate counters для run_cian_full_load. Без anchor-полей — единственный региональный проход. detail_* — опциональное обогащение detail-страниц (если enrich_detail=True). """ unique_fetched: int = 0 saved_inserted: int = 0 saved_updated: int = 0 detail_attempted: int = 0 detail_enriched: int = 0 detail_failed: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} async def run_cian_full_load( db: Session, *, run_id: int, price_cap_per_bucket: int = 1400, detail_top_n: int = 0, request_delay_sec: float = 4.0, enrich_detail: bool = False, concurrency: int = 5, resume_run_id: int | None = None, ) -> CianFullLoadCounters: """Exhaustive региональный сбор Cian ЕКБ вторички (БЕЗ anchor'ов). Cian SERP — region-wide, не geo-bbox. Anchor-loop избыточен (5 anchor'ов возвращают одну и ту же выдачу ×5). Один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление) охватывает весь ЕКБ. Параметры: price_cap_per_bucket: максимум офферов на price-бакет (< Cian SERP-cap ~1500). detail_top_n: сколько свежих cian-листингов без detail_enriched_at обогащать (актуально только при enrich_detail=True). request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default). enrich_detail: включить detail-обогащение (по умолчанию отключено — тяжело). concurrency: параллельных page-фетчей в leaf-бакете (default=5). Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора. Краш/бан больше не приводит к потере всего накопленного. Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket. """ from app.services.scrapers.cian import CianScraper counters = CianFullLoadCounters() # ── Checkpoint/resume: читаем done_buckets из прошлого run ─────────────── skip_set: set[str] = set() if resume_run_id is not None: _prev_row = db.execute( text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"), {"rid": resume_run_id}, ).fetchone() if _prev_row is not None and _prev_row.counters: _prev_counters: dict = ( _prev_row.counters if isinstance(_prev_row.counters, dict) else {} ) skip_set = set(_prev_counters.get("done_buckets", [])) logger.info( "cian-full-load run_id=%d: resuming from run %d — %d buckets already done", run_id, resume_run_id, len(skip_set), ) done: set[str] = set(skip_set) # накапливаем завершённые бакеты этого прогона def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg] """Инкрементальный save после каждого leaf-бакета. Дописывает bucket_key в done.""" nonlocal done if scrape_runs.is_cancelled(db, run_id): logger.info("cian-full-load run_id=%d: cancel detected in on_bucket", run_id) raise RuntimeError("cancelled") elif shutdown_requested(): # SIGTERM-drain: тот же sentinel-механизм, что и user-cancel, но отдельный # маркер "shutdown" → handler финализирует mark_done(partial), не mark_failed. logger.info( "cian-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", run_id, bucket_key, ) raise RuntimeError("shutdown") if not lots: done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} ) return inserted, updated = save_listings( db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today ) # save_listings вызывает db.commit() внутри — данные в БД сразу counters.saved_inserted += inserted counters.saved_updated += updated counters.unique_fetched += len(lots) done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} ) logger.info( "cian-full-load run_id=%d: bucket %s saved ins=%d upd=%d total_unique=%d", run_id, bucket_key, inserted, updated, counters.unique_fetched, ) def _on_progress(unique_count: int) -> None: """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # #1949: фоновый heartbeat — обновляет heartbeat_at каждые 60s независимо от on_bucket. # Без него зависший asyncio.gather внутри bucket-loop замораживает heartbeat на часы # → reap_zombies помечает живой run zombie. Отменяется в finally при любом завершении. async def _background_heartbeat() -> None: """Периодический heartbeat каждые 60s, не завязанный на прогресс бакетов.""" while True: await asyncio.sleep(60.0) try: scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) except Exception: pass _hb_task: asyncio.Task[None] | None = None try: _hb_task = asyncio.create_task(_background_heartbeat()) async with CianScraper() as scraper: scraper.request_delay_sec = request_delay_sec # #1949: per-fetch hard-cancel для browser-fetch'ей внутри bucket-gather. # BrowserFetcher.fetch имеет httpx-timeout 120s, но browser-сервис иногда # виснет так что httpx не получает ответ → gather морозит → heartbeat стоп. # Оборачиваем _browser.fetch в asyncio.wait_for(timeout=X) через monkey-patch: # TimeoutError → _fetch_page_html ловит как Exception → None → _one_page → [] # → gather завершается нормально → on_bucket вызывается → heartbeat обновляется. # За флагом cian_full_load_per_fetch_timeout_s (0 = отключить). _fetch_timeout = settings.cian_full_load_per_fetch_timeout_s if _fetch_timeout > 0 and scraper._browser is not None: _orig_fetch = scraper._browser.fetch async def _timed_fetch( url: str, _f: Any = _orig_fetch, _t: float = _fetch_timeout, _rid: int = run_id, ) -> str: try: return await asyncio.wait_for(_f(url), timeout=_t) except TimeoutError: logger.warning( "cian-full-load run_id=%d: browser-fetch timed out " "after %.0fs — url=%s", _rid, _t, url, ) raise scraper._browser.fetch = _timed_fetch # type: ignore[method-assign] logger.info( "cian-full-load run_id=%d: per-fetch timeout %.0fs enabled", run_id, _fetch_timeout, ) await scraper.fetch_all_secondary( price_cap_per_bucket=price_cap_per_bucket, concurrency=concurrency, secondary_only=True, on_bucket=_on_bucket, on_progress=_on_progress, skip_buckets=skip_set if skip_set else None, ) logger.info( "cian-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # ── Опциональный detail-энричмент ───────────────────────────────────── if enrich_detail and detail_top_n > 0: from sqlalchemy import text as _text from app.services.scrapers.cian_detail import ( fetch_detail, save_detail_enrichment, ) priority_rows = ( db.execute( _text(""" SELECT id, source_url FROM listings WHERE source = 'cian' AND source_url IS NOT NULL AND detail_enriched_at IS NULL AND scraped_at > NOW() - INTERVAL '4 hours' ORDER BY scraped_at DESC NULLS LAST LIMIT :lim """), {"lim": detail_top_n}, ) .mappings() .all() ) for didx, row in enumerate(priority_rows): if scrape_runs.is_cancelled(db, run_id): logger.info("cian-full-load run_id=%d: cancelled during detail enrich", run_id) break elif shutdown_requested(): logger.info( "cian-full-load run_id=%d: SIGTERM-drain — stopping detail at #%d/%d", run_id, didx + 1, len(priority_rows), ) break listing_id: int = row["id"] source_url: str = row["source_url"] counters.detail_attempted += 1 try: enrichment = await fetch_detail(source_url) if enrichment is not None: save_detail_enrichment(db, listing_id, enrichment) counters.detail_enriched += 1 else: counters.detail_failed += 1 except Exception as exc: counters.detail_failed += 1 counters.errors_count += 1 logger.warning( "cian-full-load run_id=%d: detail failed listing_id=%d: %s", run_id, listing_id, exc, ) if didx < len(priority_rows) - 1: await asyncio.sleep(request_delay_sec * random.uniform(0.8, 1.2)) logger.info( "cian-full-load run_id=%d: detail done — enriched=%d/%d failed=%d", run_id, counters.detail_enriched, counters.detail_attempted, counters.detail_failed, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) logger.info( "cian-full-load run_id=%d done: unique=%d ins=%d upd=%d detail=%d/%d errors=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, counters.detail_enriched, counters.detail_attempted, counters.errors_count, ) return counters except RuntimeError as exc: # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). if str(exc) == "cancelled": logger.info( "cian-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters if str(exc) == "shutdown": logger.info( "cian-full-load run_id=%d: SIGTERM-drain — partial results unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters logger.exception("cian-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise except Exception as exc: logger.exception("cian-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise finally: # #1949: отменяем фоновый heartbeat при любом завершении (return / raise / cancel). if _hb_task is not None: _hb_task.cancel() try: await _hb_task except asyncio.CancelledError: pass # ── Yandex exhaustive full load (region-wide, без anchor'ов) ────────────────── @dataclass class YandexFullLoadCounters: """Aggregate counters для run_yandex_full_load. Без detail_* полей — Yandex full load не делает detail-обогащение. """ unique_fetched: int = 0 saved_inserted: int = 0 saved_updated: int = 0 price_history_rows: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} async def run_yandex_full_load( db: Session, *, run_id: int, price_cap_per_bucket: int = 500, request_delay_sec: float = 2.0, concurrency: int = 4, resume_run_id: int | None = None, ) -> YandexFullLoadCounters: """Exhaustive региональный сбор Yandex ЕКБ вторички (БЕЗ anchor'ов). Yandex SERP — path-based по городу, не geo-bbox. Anchor-loop избыточен (все anchor'ы дают одну и ту же городскую выдачу). Один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление) охватывает весь ЕКБ. Параметры: price_cap_per_bucket: максимум офферов на price-бакет (< Yandex SERP-cap ~575). request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default). concurrency: параллельных page-фетчей в leaf-бакете. resume_run_id: если задан — читает done_buckets из counters прошлого run и пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного) yandex_full_load прогона для resume. Без этого поля — full walk с нуля. Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора. Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket. """ from app.services.scrapers.yandex_realty import YandexRealtyScraper counters = YandexFullLoadCounters() skip_set: set[str] = set() if resume_run_id is not None: _prev_row = db.execute( text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"), {"rid": resume_run_id}, ).fetchone() if _prev_row is not None and _prev_row.counters: _prev_counters: dict = ( _prev_row.counters if isinstance(_prev_row.counters, dict) else {} ) skip_set = set(_prev_counters.get("done_buckets", [])) logger.info( "yandex-full-load run_id=%d: resuming from run %d — %d buckets already done", run_id, resume_run_id, len(skip_set), ) done: set[str] = set(skip_set) def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg] """Инкрементальный save после каждого leaf-бакета.""" nonlocal done if scrape_runs.is_cancelled(db, run_id): logger.info("yandex-full-load run_id=%d: cancel detected in on_bucket", run_id) raise RuntimeError("cancelled") elif shutdown_requested(): logger.info( "yandex-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", run_id, bucket_key, ) raise RuntimeError("shutdown") if not lots: done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} ) return inserted, updated = save_listings( db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today ) # save_listings вызывает db.commit() внутри — данные в БД сразу counters.saved_inserted += inserted counters.saved_updated += updated counters.unique_fetched += len(lots) # Price-history из gate price.previous/trend (graceful — провал истории # НЕ должен ронять сейв листингов/бакета). try: counters.price_history_rows += record_yandex_price_history(db, lots) except Exception as ph_exc: logger.warning( "yandex-full-load run_id=%d: price-history failed bucket=%s: %s", run_id, bucket_key, ph_exc, ) try: db.rollback() except Exception: pass done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} ) logger.info( "yandex-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d", run_id, bucket_key, inserted, updated, counters.unique_fetched, ) def _on_progress(unique_count: int) -> None: """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) try: async with YandexRealtyScraper() as scraper: scraper.request_delay_sec = request_delay_sec await scraper.fetch_all_secondary( price_cap_per_bucket=price_cap_per_bucket, concurrency=concurrency, on_bucket=_on_bucket, on_progress=_on_progress, skip_buckets=skip_set if skip_set else None, ) logger.info( "yandex-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) logger.info( "yandex-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, counters.errors_count, ) return counters except RuntimeError as exc: # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). if str(exc) == "cancelled": logger.info( "yandex-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters if str(exc) == "shutdown": logger.info( "yandex-full-load run_id=%d: SIGTERM-drain — partial results " "unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters logger.exception("yandex-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise except Exception as exc: logger.exception("yandex-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise # ── Avito exhaustive full load (citywide, без anchor'ов) ────────────────────── @dataclass class AvitoFullLoadCounters: """Aggregate counters для run_avito_full_load. Зеркало YandexFullLoadCounters — citywide room×price bisection без anchor/detail. """ unique_fetched: int = 0 saved_inserted: int = 0 saved_updated: int = 0 errors_count: int = 0 def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} async def run_avito_full_load( db: Session, *, run_id: int, price_cap_per_bucket: int = 1400, request_delay_sec: float = 7.0, concurrency: int = 5, secondary_only: bool = True, resume_run_id: int | None = None, incremental_days: int | None = None, ) -> AvitoFullLoadCounters: """Exhaustive ИЛИ инкрементальный региональный сбор Avito ЕКБ вторички (БЕЗ anchor'ов). Avito anchor/citywide упирается в SERP-cap ~5000 результатов на запрос → вторичка катастрофически недобрана. Решение зеркалит cian/yandex full_load: один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное деление ценового диапазона через `pmin`/`pmax`) охватывает весь ЕКБ. Параметры: price_cap_per_bucket: максимум результатов на price-бакет (< Avito SERP-cap ~5000). request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default). concurrency: параллельных page-фетчей в leaf-бакете. secondary_only: отбрасывать новостройки (listing_segment=="novostroyki") до save. resume_run_id: если задан — читает done_buckets из counters прошлого run и пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного) avito_full_load прогона для resume. Без поля — full walk с нуля. incremental_days: если задан — shallow инкрементальный обход. Считается since = date.today() - timedelta(days=incremental_days) и передаётся в fetch_all_secondary(since=since): date early-stop останавливает пагинацию до глубоких страниц (избегает page-36 429-банов). None → since=None → exhaustive bisection (полный обход, текущее поведение). Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора. Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket. AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned'), как в run_avito_city_sweep — partial results уже сохранены через on_bucket. """ counters = AvitoFullLoadCounters() # ── Инкрементальный режим: since-cutoff для shallow обхода ──────────────── since: date | None = None if incremental_days is not None: since = date.today() - timedelta(days=incremental_days) logger.info( "avito-full-load run_id=%d: INCREMENTAL mode — since=%s (last %d days)", run_id, since.isoformat(), incremental_days, ) # ── Checkpoint/resume: читаем done_buckets из прошлого run ─────────────── skip_set: set[str] = set() if resume_run_id is not None: _prev_row = db.execute( text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"), {"rid": resume_run_id}, ).fetchone() if _prev_row is not None and _prev_row.counters: _prev_counters: dict = ( _prev_row.counters if isinstance(_prev_row.counters, dict) else {} ) skip_set = set(_prev_counters.get("done_buckets", [])) logger.info( "avito-full-load run_id=%d: resuming from run %d — %d buckets already done", run_id, resume_run_id, len(skip_set), ) done: set[str] = set(skip_set) def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg] """Инкрементальный save после каждого leaf-бакета.""" nonlocal done if scrape_runs.is_cancelled(db, run_id): logger.info("avito-full-load run_id=%d: cancel detected in on_bucket", run_id) raise RuntimeError("cancelled") elif shutdown_requested(): logger.info( "avito-full-load run_id=%d: SIGTERM-drain — stopping at bucket %s", run_id, bucket_key, ) raise RuntimeError("shutdown") if not lots: done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} ) return inserted, updated = save_listings( db, lots, run_id=run_id, skip_seen_today=settings.scraper_skip_seen_today ) # save_listings вызывает db.commit() внутри — данные в БД сразу counters.saved_inserted += inserted counters.saved_updated += updated counters.unique_fetched += len(lots) done.add(bucket_key) scrape_runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)} ) logger.info( "avito-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d", run_id, bucket_key, inserted, updated, counters.unique_fetched, ) def _on_progress(unique_count: int) -> None: """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов).""" scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) try: async with AvitoScraper() as scraper: scraper.request_delay_sec = request_delay_sec await scraper.fetch_all_secondary( price_cap_per_bucket=price_cap_per_bucket, concurrency=concurrency, secondary_only=secondary_only, on_bucket=_on_bucket, on_progress=_on_progress, skip_buckets=skip_set if skip_set else None, since=since, ) logger.info( "avito-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) logger.info( "avito-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, counters.errors_count, ) return counters except (AvitoBlockedError, AvitoRateLimitedError) as exc: # IP-бан/rate-limit: partial results уже сохранены через on_bucket. logger.error("avito-full-load run_id=%d: blocked/rate-limited — %s", run_id, exc) counters.errors_count += 1 scrape_runs.mark_banned( db, run_id, f"avito full load aborted: {exc}", {**counters.to_dict(), "done_buckets": sorted(done)}, ) return counters except RuntimeError as exc: # on_bucket кидает RuntimeError("cancelled") при cooperative cancel, # RuntimeError("shutdown") при кооперативном SIGTERM-drain (#1182 Phase 3a). if str(exc) == "cancelled": logger.info( "avito-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters if str(exc) == "shutdown": logger.info( "avito-full-load run_id=%d: SIGTERM-drain — partial results " "unique=%d ins=%d upd=%d", run_id, counters.unique_fetched, counters.saved_inserted, counters.saved_updated, ) scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}) return counters logger.exception("avito-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise except Exception as exc: logger.exception("avito-full-load run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise # ── DomClick citywide sweep (без anchor'ов) ──────────────────────────────── @dataclass class DomClickCitySweepCounters: """Aggregate counters для DomClick citywide sweep run. DomClick BFF — citywide (city_id), не geo-bbox: anchor-loop отсутствует. Один проход fetch_city перебирает ROOM_BUCKETS × pages внутри scraper'а. Поля симметричны sibling-counters, но без detail_*/houses_* (SERP-only) и без anchors_* (нет anchor-фазы). pages_fetched = 6 buckets × pages worst-case. """ lots_fetched: int = 0 lots_inserted: int = 0 lots_updated: int = 0 pages_fetched: int = 0 errors_count: int = 0 blocked: int = 0 # 1 если QRATOR-блок был во время sweep geo_filtered: int = 0 # число офферов отфильтрованных geo-guard def to_dict(self) -> dict[str, int]: return {f.name: getattr(self, f.name) for f in fields(self)} # Дефолтные параметры sweep'а (EKB city_id=4). DOMCLICK_DEFAULT_CITY_ID: int = 4 DOMCLICK_DEFAULT_ROOMS: list[int] = [0, 1, 2, 3, 4] # vestigial; scraper sweeps all buckets # Число BFF-бакетов (st/1/2/3/4/5+) — фиксировано; используется для watchdog. _DOMCLICK_NUM_BUCKETS: int = 6 # Оценка времени одного fetch'а (network + parse) для watchdog. _DOMCLICK_PER_FETCH_S: float = 12.0 # Буфер сверху расчётного бюджета (cold browser start, save-фаза, price-splits). _DOMCLICK_SWEEP_BUDGET_S: float = 300.0 async def run_domclick_city_sweep( db: Session, *, run_id: int, city_id: int = DOMCLICK_DEFAULT_CITY_ID, rooms: list[int] | None = None, pages: int = 100, request_delay_sec: float | None = None, ) -> DomClickCitySweepCounters: """DomClick citywide sweep через BFF JSON API. Структурно зеркалит run_cian_city_sweep / run_yandex_city_sweep, но CITYWIDE: DomClick не поддерживает geo-radius (fetch_around → NotImplementedError), поэтому anchor-loop отсутствует. DomClickScraper.fetch_city перебирает все ROOM_BUCKETS (st/1/2/3/4/5+) через BrowserFetcher → shared mobile proxy → обходит QRATOR. Фазы (единственный citywide проход): 1. SERP: DomClickScraper().fetch_city(city_id, rooms, pages). 2. SAVE: save_listings(db, lots, run_id=run_id) — house-match по lat/lon и address-fingerprint (BFF отдаёт координаты). Watchdog: весь fetch_city оборачивается в asyncio.wait_for. Таймаут считается из 6 buckets × pages × per-fetch + buffer (минимум ANCHOR_TIMEOUT_SEC). Реальный sweep ≈ 330 страниц (rooms=2 требует price-split). Cooperative cancel: is_cancelled(db, run_id) проверяется перед SERP-фазой. ЧЕСТНЫЙ СТАТУС (#1968): если scraper сообщил QRATOR-блок И lots == 0 → mark_failed с объяснением. Иначе mark_done (в т.ч. при lots==0 без блока: genuinely empty run). Возвращает DomClickCitySweepCounters. """ from app.services.scrapers.domclick import ROOM_BUCKETS, DomClickScraper _resolved_delay = request_delay_sec if request_delay_sec is not None else 6.0 counters = DomClickCitySweepCounters() # Watchdog: 6 buckets × pages × per_fetch + budget. # Реальный worst-case (rooms=2 price-split) ≈ 330 страниц × (delay+12s). # pages=100 → 600 fetch'ей × ~18s + 300 ≈ 11100s (~3ч) — намеренно generous. _num_fetches = _DOMCLICK_NUM_BUCKETS * max(1, pages) _sweep_timeout = max( ANCHOR_TIMEOUT_SEC, int(_num_fetches * (_resolved_delay + _DOMCLICK_PER_FETCH_S) + _DOMCLICK_SWEEP_BUDGET_S), ) # Мутируемый контейнер для захвата scraper-ссылки из замыкания. _scraper_ref: list[DomClickScraper] = [] try: # ── Cooperative cancel перед SERP-фазой ────────────────────────────── if scrape_runs.is_cancelled(db, run_id): logger.info("domclick-sweep run_id=%d: cancelled before SERP", run_id) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) return counters elif shutdown_requested(): # SIGTERM-drain до SERP-фазы: финализируем run как mark_done(partial), # минуя honest-status (это чистый drain, не QRATOR-блок). logger.info("domclick-sweep run_id=%d: SIGTERM-drain — stopping before SERP", run_id) scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) scrape_runs.mark_done(db, run_id, counters.to_dict()) return counters logger.info( "domclick-sweep run_id=%d: BFF citywide sweep city_id=%d " "buckets=%d pages_cap=%d (watchdog %ds)", run_id, city_id, len(ROOM_BUCKETS), pages, _sweep_timeout, ) lots: list[ScrapedLot] = [] async def _domclick_phase() -> None: """Единственная citywide-фаза: fetch_city + save.""" nonlocal lots async with DomClickScraper() as _scraper: _scraper_ref.append(_scraper) if request_delay_sec is not None: _scraper.request_delay_sec = _resolved_delay lots = await _scraper.fetch_city(city_id=city_id, rooms=rooms, pages=pages) counters.lots_fetched += len(lots) if lots: inserted, updated = save_listings(db, lots, run_id=run_id) counters.lots_inserted += inserted counters.lots_updated += updated try: await asyncio.wait_for(_domclick_phase(), timeout=_sweep_timeout) except TimeoutError: logger.warning( "domclick-sweep run_id=%d: SERP phase timed out after %ds — partial results", run_id, _sweep_timeout, ) counters.errors_count += 1 except Exception: logger.exception("domclick-sweep run_id=%d: SERP phase failed", run_id) counters.errors_count += 1 # Перенести счётчики scraper'а в counters (если scraper был создан). if _scraper_ref: _s = _scraper_ref[0] counters.blocked = 1 if _s.blocked else 0 counters.geo_filtered = _s.geo_filtered # Не-block fetch-ошибки скрейпера учитываем в errors_count: они должны # влиять на honest-status (0 lots + fetch_errors → НЕ ложный mark_done). counters.errors_count += _s.fetch_errors # pages_fetched: worst-case число страниц (buckets × pages cap). counters.pages_fetched = _num_fetches scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) # ── ЧЕСТНЫЙ СТАТУС (#1968) ──────────────────────────────────────────── # 0 lots + (block ИЛИ любая ошибка) → failed: run без результата по # внешней причине не должен маскироваться как 'done'. Нераспознанный # block-вариант (нет литерал-маркера) приходит как fetch_error, не blocked — # поэтому errors_count тоже триггерит failed. Genuinely empty 0-error run # → done. Partial run с lots>0 остаётся done даже если часть бакетов упала. if counters.lots_fetched == 0 and (counters.blocked or counters.errors_count > 0): logger.error( "domclick-sweep run_id=%d: 0 listings with blocked=%d errors=%d " "— marking failed", run_id, counters.blocked, counters.errors_count, ) scrape_runs.mark_failed( db, run_id, "QRATOR block or fetch errors — 0 listings", counters.to_dict(), ) else: scrape_runs.mark_done(db, run_id, counters.to_dict()) logger.info( "domclick-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) " "pages=%d errors=%d blocked=%d geo_filtered=%d", run_id, counters.lots_fetched, counters.lots_inserted, counters.lots_updated, counters.pages_fetched, counters.errors_count, counters.blocked, counters.geo_filtered, ) return counters except Exception as exc: logger.exception("domclick-sweep run_id=%d: fatal error", run_id) scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise # ── Public re-exports ─────────────────────────────────────────── __all__ = [ "ANCHOR_TIMEOUT_SEC", "CIAN_SWEEP_MAX_CONSECUTIVE_FAILURES", "EKB_ANCHORS", "CianCitySweepCounters", "CianFullLoadCounters", "CitySweepCounters", "DomClickCitySweepCounters", "NewbuildingSweepCounters", "PipelineCounters", "PipelineResult", "YandexCitySweepCounters", "YandexFullLoadCounters", "run_avito_city_sweep", "run_avito_newbuilding_sweep", "run_avito_pipeline", "run_cian_city_sweep", "run_cian_full_load", "run_domclick_city_sweep", "run_yandex_city_sweep", "run_yandex_full_load", ]