"""pipeline.py — Avito sweep orchestrator (strangler-копия #2135). Strangler-COPY боевого `app.services.scrape_pipeline` (шаг оркестрации, #2135). Логика фаз/бан-ротации/счётчиков идентична оригиналу — единственное отличие в развязке от `app.*` через инжекцию (см. `scraper_kit.contracts`): - `app.core.config.settings` → аргумент `config: ScraperConfig` - `app.services.matching` → аргумент `matcher: HouseMatcher` - `app.services.house_imv_backfill` → аргумент `enrichment: EnrichmentJobs` - `app.core.shutdown.shutdown_requested` → аргумент `shutdown_requested: Callable` - `app.services.scrape_runs` → локальный `scraper_kit.orchestration.runs` - провайдеры/база → `scraper_kit.providers.*` / `scraper_kit.base` Скопирована КРИТИЧНАЯ логика оркестрации (#2135): - `run_avito_pipeline` — одиночный anchor (SEARCH→SAVE→GROUP→HOUSES→DETAIL) - `run_avito_city_sweep` — полный city-sweep + ban/rotation + partial-intake + IMV Остаток (yandex/cian/domclick city-sweep, *_full_load, newbuilding sweep) — follow-up: та же механическая развязка, но не входит в первый срез #2135. Ban-rotation state-machine сведена в единый хелпер `_try_rotate_within_budget` (в оригинале инлайн-дублировалась в house/detail/house-detail фазах). """ from __future__ import annotations import asyncio import logging import random from collections.abc import Callable from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields from typing import TYPE_CHECKING from urllib.parse import urlparse from curl_cffi.requests import AsyncSession from sqlalchemy import text from scraper_kit.avito_exceptions import AvitoBlockedError, AvitoRateLimitedError from scraper_kit.base import ScrapedLot, save_listings from scraper_kit.browser_fetcher import BrowserFetcher from scraper_kit.orchestration import runs from scraper_kit.providers.avito.detail import fetch_detail, save_detail_enrichment from scraper_kit.providers.avito.houses import ( fetch_house_catalog, save_house_catalog_enrichment, ) from scraper_kit.providers.avito.serp import AvitoScraper if TYPE_CHECKING: from sqlalchemy.orm import Session from scraper_kit.contracts import EnrichmentJobs, HouseMatcher, ScraperConfig logger = logging.getLogger(__name__) # Регион по умолчанию для listings.region_code (в старом save_listings хардкод 66). # Вынесен параметром run_*-функций — kit не знает про конкретный регион. DEFAULT_REGION_CODE: int = 66 def _max_rotations(config: ScraperConfig, source: str) -> int: """Лимит IP-ротаций по провайдеру (avito/cian/yandex).""" if source == "cian": return config.cian_proxy_max_rotations if source == "yandex": return config.yandex_proxy_max_rotations return config.avito_proxy_max_rotations async def _rotate_proxy_ip( config: ScraperConfig, *, 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 = config.cian_proxy_rotate_url or config.avito_proxy_rotate_url max_rot = config.cian_proxy_max_rotations elif source == "yandex": rotate_url = config.yandex_proxy_rotate_url or config.avito_proxy_rotate_url max_rot = config.yandex_proxy_max_rotations else: # avito (default) — backward compat rotate_url = config.avito_proxy_rotate_url max_rot = config.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 = config.proxy_rotate_attempts _attempt_timeout = config.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(config.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 async def _try_rotate_within_budget( config: ScraperConfig, *, reason: str, rotations_done: int, source: str = "avito", ) -> tuple[bool, int]: """Единый шаг ban-rotation state-machine (свод инлайн-дублей #2135). Вызывается при достижении порога ПОСЛЕДОВАТЕЛЬНЫХ блоков. Если бюджет ротаций ещё не исчерпан (`rotations_done < max_rotations(source)`) — дёргает changeip и увеличивает счётчик ротаций. Возвращает `(rotated, rotations_done')`: - rotated=True → IP сменён; вызывающий сбрасывает consecutive и продолжает. - rotated=False → бюджет исчерпан ЛИБО changeip не удался; вызывающий делает abort/propagate (листинги уже сохранены). Поведение идентично прежним инлайн-веткам: инкремент rotations_done происходит только при наличии бюджета (когда реально пытаемся ротировать). """ if rotations_done < _max_rotations(config, source): rotated = await _rotate_proxy_ip( config, reason=reason, rotations_done=rotations_done, source=source ) return rotated, rotations_done + 1 return False, rotations_done # Per-anchor watchdog timeout (секунды). Если обработка одного anchor'а занимает # дольше этого значения (зависший HTTP-запрос, прокси hang), asyncio.wait_for # прерывает её с TimeoutError, sweep продолжается со следующим anchor'ом. # Диапазон 180-300 с; 240 — разумный середина (4 мин > любой нормальный anchor). ANCHOR_TIMEOUT_SEC: int = 240 # Константы для расчёта 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). _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). _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5 # #1820: порог ПОСЛЕДОВАТЕЛЬНЫХ блоков (AvitoBlockedError/AvitoRateLimitedError) внутри # enrichment-фаз run_avito_pipeline. Один заблокированный фетч (house/detail) больше НЕ # роняет весь прогон — блок ловится пер-item (как обычная ошибка), и лишь N подряд # прерывают enrichment-фазу. Listings из SEARCH+SAVE сохраняются всегда. _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3 # 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(config: ScraperConfig) -> 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). Читает config.scraper_proxy_url (SCRAPER_PROXY_URL > AVITO_PROXY_URL fallback). """ url = config.scraper_proxy_url return {"http": url, "https": url} if url else None # ── 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, *, config: ScraperConfig, matcher: HouseMatcher, 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, region_code: int = DEFAULT_REGION_CODE, ) -> 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 = config.scraper_fetch_mode == "browser" session: AsyncSession | None = None browser_fetcher: BrowserFetcher | None = None own_session = False own_browser = False scraper = AvitoScraper(config) if browser_mode: browser_fetcher = shared_browser if browser_fetcher is None: browser_fetcher = BrowserFetcher( source="avito", endpoint=config.browser_http_endpoint ) await browser_fetcher.__aenter__() own_browser = True scraper._browser = browser_fetcher else: own_session = shared_session is None session = shared_session or AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, proxies=_avito_proxies(config), ) 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, matcher=matcher, region_code=region_code ) 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, matcher=matcher) 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 уже сохранены). rotated, enrich_rotations_done = await _try_rotate_within_budget( config, reason=f"house consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) if rotated: enrich_consecutive_blocks = 0 logger.info( "pipeline:enrich ROTATE — IP changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, config.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 уже сохранены). rotated, enrich_rotations_done = await _try_rotate_within_budget( config, reason=f"detail consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) if rotated: enrich_consecutive_blocks = 0 logger.info( "pipeline:enrich ROTATE — IP changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, config.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, matcher=matcher ) 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). rotated, enrich_rotations_done = await _try_rotate_within_budget( config, reason=f"house-detail consecutive={enrich_consecutive_blocks}", rotations_done=enrich_rotations_done, ) if rotated: enrich_consecutive_blocks = 0 logger.info( "pipeline:enrich ROTATE — IP changed, " "reset consecutive blocks (rotation #%d/%d)", enrich_rotations_done, config.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) @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, config: ScraperConfig, matcher: HouseMatcher, enrichment: EnrichmentJobs, shutdown_requested: Callable[[], bool] = lambda: False, 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, region_code: int = DEFAULT_REGION_CODE, ) -> 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-результаты не теряются. Инжекция (#2135): config/matcher/enrichment/shutdown_requested приходят снаружи вместо прямых импортов app.* (см. scraper_kit.contracts). process_houses_imv_batch зовётся через enrichment (EnrichmentJobs Protocol). """ _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 = config.scraper_fetch_mode == "browser" # #1790: общий бюджет IP-ротаций на весь sweep (разделяется по всем anchor'ам). sweep_rotations_done = 0 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", endpoint=config.browser_http_endpoint) ) else: session = await stack.enter_async_context( AsyncSession( impersonate="chrome120", timeout=25, headers=_CHROME_HEADERS, proxies=_avito_proxies(config), ) ) try: for idx, (lat, lon, name) in enumerate(_anchors, start=1): if 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, ) 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, ) runs.update_heartbeat(db, run_id, counters.to_dict()) 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 # ── Phase 1+2: SERP + save ───────────────────────────────── scraper = AvitoScraper(config) 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, matcher=matcher, region_code=region_code ) 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_hc = await fetch_house_catalog( house_path, cffi_session=session, browser_fetcher=shared_bf, ) hc = save_house_catalog_enrichment( db, enrichment_hc, matcher=matcher ) 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 < config.avito_proxy_max_rotations ): rotated, sweep_rotations_done = ( await _try_rotate_within_budget( config, reason=( f"city-sweep house " f"consecutive={sweep_consecutive_house_blocks}" ), rotations_done=sweep_rotations_done, ) ) if rotated: sweep_consecutive_house_blocks = 0 logger.info( "city-sweep run_id=%d: house ROTATE — " "IP changed, reset consecutive " "(rotation #%d/%d)", run_id, sweep_rotations_done, config.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'ом. rotated, sweep_rotations_done = ( await _try_rotate_within_budget( config, reason=( f"city-sweep detail " f"consecutive={consecutive_blocks}" ), rotations_done=sweep_rotations_done, ) ) if rotated: consecutive_blocks = 0 logger.info( "city-sweep run_id=%d: detail ROTATE — " "IP changed, reset consecutive " "(rotation #%d/%d)", run_id, sweep_rotations_done, config.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_hc = await fetch_house_catalog( house_path, cffi_session=session, browser_fetcher=shared_bf, ) hc = save_house_catalog_enrichment( db, enrichment_hc, matcher=matcher ) 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 < config.avito_proxy_max_rotations ): rotated, sweep_rotations_done = ( await _try_rotate_within_budget( config, reason=( "city-sweep house(detail) " f"consecutive=" f"{sweep_consecutive_detail_house_blocks}" ), rotations_done=sweep_rotations_done, ) ) if rotated: sweep_consecutive_detail_house_blocks = 0 logger.info( "city-sweep run_id=%d: house(detail) " "ROTATE — IP changed (rotation #%d/%d)", run_id, sweep_rotations_done, config.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 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 config.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, ) runs.mark_done( db, run_id, {**counters.to_dict(), "enrichment_abort_note": _note}, # type: ignore[arg-type] ) else: 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 runs.update_heartbeat(db, run_id, counters.to_dict()) # ── IMV-фаза: финальный обход тронутых домов ────────── if enrich_imv and all_touched_house_ids: if 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), ) 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), ) 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: runs.update_heartbeat(db, run_id, counters.to_dict()) imv_result = await enrichment.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 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 runs.update_heartbeat(db, run_id, counters.to_dict()) 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) runs.mark_failed(db, run_id, str(exc), counters.to_dict()) raise