diff --git a/tradein-mvp/backend/app/services/scraper_adapters.py b/tradein-mvp/backend/app/services/scraper_adapters.py index 8d31a4cf..55ec419f 100644 --- a/tradein-mvp/backend/app/services/scraper_adapters.py +++ b/tradein-mvp/backend/app/services/scraper_adapters.py @@ -23,6 +23,9 @@ from app.core.config import settings as _settings from app.core.db import SessionLocal as _SessionLocal from app.services.house_dedup_merge import run_house_dedup_merge as _run_house_dedup_merge from app.services.house_imv_backfill import backfill_house_imv as _backfill_house_imv +from app.services.house_imv_backfill import ( + process_houses_imv_batch as _process_houses_imv_batch, +) from app.services.matching import match_or_create_house as _match_or_create_house from app.services.matching import upsert_listing_source as _upsert_listing_source from app.services.yandex_address_backfill import ( @@ -154,6 +157,43 @@ class RealScraperConfig: def glitchtip_dsn(self) -> str | None: return _settings.glitchtip_dsn + # ── Оркестрация sweep-pipeline (#2135) ─────────────────────────────── + @property + def avito_serp_ok_not_banned(self) -> bool: + return _settings.avito_serp_ok_not_banned + + @property + def avito_proxy_rotate_settle_s(self) -> float: + return _settings.avito_proxy_rotate_settle_s + + @property + def proxy_rotate_attempts(self) -> int: + return _settings.proxy_rotate_attempts + + @property + def proxy_rotate_attempt_timeout_s(self) -> float: + return _settings.proxy_rotate_attempt_timeout_s + + @property + def cian_proxy_rotate_url(self) -> str | None: + return _settings.cian_proxy_rotate_url + + @property + def cian_proxy_max_rotations(self) -> int: + return _settings.cian_proxy_max_rotations + + @property + def yandex_proxy_max_rotations(self) -> int: + return _settings.yandex_proxy_max_rotations + + @property + def scraper_skip_seen_today(self) -> bool: + return _settings.scraper_skip_seen_today + + @property + def cian_full_load_per_fetch_timeout_s(self) -> float: + return _settings.cian_full_load_per_fetch_timeout_s + class RealSessionFactory: """SessionFactory-адаптер над `app.core.db.SessionLocal`.""" @@ -206,6 +246,21 @@ class RealEnrichmentJobs: request_delay_sec=request_delay_sec, ) + async def process_houses_imv_batch( + self, + db: Session, + house_ids: set[int], + *, + request_delay_sec: float = 5.0, + heartbeat: Callable[[], None] | None = None, + ) -> HouseIMVBackfillResult: + return await _process_houses_imv_batch( + db, + house_ids, + request_delay_sec=request_delay_sec, + heartbeat=heartbeat, + ) + __all__ = [ "RealEnrichmentJobs", diff --git a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py new file mode 100644 index 00000000..92f43b95 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py @@ -0,0 +1,383 @@ +"""Golden-parity: `scraper_kit.orchestration.pipeline` ≡ `app.services.scrape_pipeline`. + +Strangler-инвариант (#2135): новая scraper_kit-копия оркестрации sweep-pipeline +должна давать ТОТ ЖЕ наблюдаемый эффект, что и старый боевой orchestrator, на +одинаковом мокнутом I/O. Развязка `app.*` (config/matcher/enrichment/runs/shutdown +инжектируются) не должна менять ветвление. + +Фокус — КРИТИЧНАЯ логика оркестрации (то, ради чего PR): + - ban/rotation state-machine: SERP-блок на anchor'е → abort sweep; + - partial-ban intake (#1950): SERP intake сохранён + detail заблокирован → + mark_done (не mark_banned) под флагом avito_serp_ok_not_banned; + - detail-фаза: N подряд блоков → propagate → anchor-handler; + - counters aggregation по anchor'ам + IMV-фаза; + - последовательность вызовов scrape_runs (heartbeat / mark_done / mark_banned). + +Метод: гоняем ОБА orchestrator'а (`_drive_old` / `_drive_new`) на идентичном +описании сценария, перехватывая scrape_runs recording-mock'ом. Сравниваем: + - итоговый counters.to_dict(); + - последовательность (method, numeric-counters) вызовов scrape_runs. + +Без сети, без БД. +""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.avito_exceptions import ( + AvitoBlockedError as NewAvitoBlockedError, +) +from scraper_kit.orchestration.pipeline import ( + run_avito_city_sweep as new_city_sweep, +) + +from app.services.scrape_pipeline import run_avito_city_sweep as old_city_sweep +from app.services.scrapers.avito_exceptions import ( + AvitoBlockedError as OldAvitoBlockedError, +) + +# ── recording scrape_runs ──────────────────────────────────────────────────── + +_NON_NUMERIC_KEYS = {"enrichment_abort_note"} + + +class _RunsRecorder: + """Записывает вызовы scrape_runs-финализаторов в общий список. + + is_cancelled всегда False (кооп-cancel не тестируем здесь). Каждый терминальный + вызов пишет (method, counters_dict) — сравниваем последовательности old/new. + """ + + def __init__(self) -> None: + self.calls: list[tuple[str, dict[str, Any]]] = [] + + def is_cancelled(self, db: Any, run_id: int) -> bool: + return False + + def update_heartbeat(self, db: Any, run_id: int, counters: dict[str, Any]) -> None: + self.calls.append(("update_heartbeat", dict(counters))) + + def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None: + self.calls.append(("mark_done", dict(counters))) + + def mark_banned(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: + self.calls.append(("mark_banned", dict(counters))) + + def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: + self.calls.append(("mark_failed", dict(counters))) + + +_DriveResult = tuple[dict[str, int], list[tuple[str, dict[str, int]]]] + + +def _normalize(calls: list[tuple[str, dict[str, Any]]]) -> list[tuple[str, dict[str, int]]]: + """Оставить только имя метода + числовые counters (убрать note-строки).""" + out: list[tuple[str, dict[str, int]]] = [] + for method, counters in calls: + numeric = {k: v for k, v in counters.items() if k not in _NON_NUMERIC_KEYS} + out.append((method, numeric)) + return out + + +# ── scenario description ───────────────────────────────────────────────────── + + +class _Scenario: + """Декларативное описание одного прогона city-sweep для обоих orchestrator'ов.""" + + def __init__( + self, + *, + anchors: list[tuple[float, float, str]], + # per-anchor: ("lots", n_lots, ins, upd) | ("block",) + per_anchor: list[tuple[Any, ...]], + enrich_houses: bool = False, + detail_top_n: int = 0, + detail_rows: int = 0, + detail_behavior: str = "ok", # "ok" | "block_all" + enrich_imv: bool = False, + imv_result: tuple[int, int, int] | None = None, + avito_serp_ok_not_banned: bool = True, + avito_proxy_max_rotations: int = 0, + lots_have_house_url: bool = False, + ) -> None: + self.anchors = anchors + self.per_anchor = per_anchor + self.enrich_houses = enrich_houses + self.detail_top_n = detail_top_n + self.detail_rows = detail_rows + self.detail_behavior = detail_behavior + self.enrich_imv = enrich_imv + self.imv_result = imv_result + self.avito_serp_ok_not_banned = avito_serp_ok_not_banned + self.avito_proxy_max_rotations = avito_proxy_max_rotations + self.lots_have_house_url = lots_have_house_url + + def _config(self) -> SimpleNamespace: + return SimpleNamespace( + scraper_fetch_mode="curl_cffi", + browser_http_endpoint="http://browser.test/fetch", + scraper_proxy_url=None, + avito_proxy_rotate_url=None, + avito_proxy_max_rotations=self.avito_proxy_max_rotations, + avito_serp_ok_not_banned=self.avito_serp_ok_not_banned, + avito_proxy_rotate_settle_s=0.0, + proxy_rotate_attempts=1, + proxy_rotate_attempt_timeout_s=1.0, + cian_proxy_rotate_url=None, + cian_proxy_max_rotations=0, + yandex_proxy_rotate_url=None, + yandex_proxy_max_rotations=0, + scraper_skip_seen_today=False, + ) + + def _fetch_around_side_effects(self, blocked_exc: type[Exception]) -> list[Any]: + effects: list[Any] = [] + for spec in self.per_anchor: + if spec[0] == "block": + effects.append(blocked_exc("SERP blocked")) + else: + _, n_lots, _ins, _upd = spec + house_url = "/catalog/houses/ekb/h1/100" if self.lots_have_house_url else None + effects.append([MagicMock(house_url=house_url) for _ in range(n_lots)]) + return effects + + def _save_side_effects(self) -> list[tuple[int, int]]: + return [(spec[2], spec[3]) for spec in self.per_anchor if spec[0] == "lots" and spec[1] > 0] + + def _detail_rows(self) -> list[dict[str, str]]: + return [{"source_url": f"https://www.avito.ru/x/{i}"} for i in range(self.detail_rows)] + + def _fetch_detail_side_effects(self, blocked_exc: type[Exception]) -> Any: + if self.detail_behavior == "block_all": + return blocked_exc("detail blocked") + return MagicMock(house_catalog_url=None) + + +def _make_db(scenario: _Scenario) -> MagicMock: + db = MagicMock() + # detail-фаза: db.execute(...).mappings().all() → priority rows + db.execute.return_value.mappings.return_value.all.return_value = scenario._detail_rows() + return db + + +def _make_scraper(scenario: _Scenario, blocked_exc: type[Exception]) -> MagicMock: + scraper = MagicMock() + scraper._cffi = None + scraper._browser = None + scraper.fetch_around = AsyncMock(side_effect=scenario._fetch_around_side_effects(blocked_exc)) + return scraper + + +def _async_session_cm() -> MagicMock: + sess = MagicMock() + sess.__aenter__ = AsyncMock(return_value=sess) + sess.__aexit__ = AsyncMock(return_value=None) + sess.close = AsyncMock() + return sess + + +async def _drive_old(scenario: _Scenario) -> _DriveResult: + recorder = _RunsRecorder() + db = _make_db(scenario) + scraper = _make_scraper(scenario, OldAvitoBlockedError) + save_mock = MagicMock(side_effect=scenario._save_side_effects()) + + imv_res = None + if scenario.imv_result is not None: + checked, saved, errors = scenario.imv_result + imv_res = SimpleNamespace(checked=checked, saved=saved, errors=errors) + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper), + patch("app.services.scrape_pipeline.save_listings", save_mock), + patch( + "app.services.scrape_pipeline.fetch_house_catalog", + AsyncMock(return_value=MagicMock()), + ), + patch( + "app.services.scrape_pipeline.save_house_catalog_enrichment", + return_value={"house_id": 1}, + ), + patch( + "app.services.scrape_pipeline.fetch_detail", + AsyncMock(side_effect=scenario._fetch_detail_side_effects(OldAvitoBlockedError)), + ), + patch("app.services.scrape_pipeline.save_detail_enrichment", return_value=True), + patch("app.services.scrape_pipeline.scrape_runs", recorder), + patch( + "app.services.house_imv_backfill.process_houses_imv_batch", + AsyncMock(return_value=imv_res), + ), + patch("app.services.scrape_pipeline.settings", scenario._config()), + patch("app.services.scrape_pipeline.shutdown_requested", return_value=False), + patch("app.services.scrape_pipeline.asyncio.sleep", AsyncMock()), + patch("app.services.scrape_pipeline.AsyncSession", return_value=_async_session_cm()), + ): + counters = await old_city_sweep( + db, + run_id=1, + radius_m=1000, + anchors=scenario.anchors, + pages_per_anchor=1, + enrich_houses=scenario.enrich_houses, + detail_top_n=scenario.detail_top_n, + request_delay_sec=0.0, + enrich_imv=scenario.enrich_imv, + ) + return counters.to_dict(), _normalize(recorder.calls) + + +async def _drive_new(scenario: _Scenario) -> _DriveResult: + recorder = _RunsRecorder() + db = _make_db(scenario) + scraper = _make_scraper(scenario, NewAvitoBlockedError) + save_mock = MagicMock(side_effect=scenario._save_side_effects()) + + imv_res = None + if scenario.imv_result is not None: + checked, saved, errors = scenario.imv_result + imv_res = SimpleNamespace(checked=checked, saved=saved, errors=errors) + + enrichment = MagicMock() + enrichment.process_houses_imv_batch = AsyncMock(return_value=imv_res) + + pfx = "scraper_kit.orchestration.pipeline" + with ( + patch(f"{pfx}.AvitoScraper", return_value=scraper), + patch(f"{pfx}.save_listings", save_mock), + patch(f"{pfx}.fetch_house_catalog", AsyncMock(return_value=MagicMock())), + patch(f"{pfx}.save_house_catalog_enrichment", return_value={"house_id": 1}), + patch( + f"{pfx}.fetch_detail", + AsyncMock(side_effect=scenario._fetch_detail_side_effects(NewAvitoBlockedError)), + ), + patch(f"{pfx}.save_detail_enrichment", return_value=True), + patch(f"{pfx}.runs", recorder), + patch(f"{pfx}.asyncio.sleep", AsyncMock()), + patch(f"{pfx}.AsyncSession", return_value=_async_session_cm()), + ): + counters = await new_city_sweep( + db, + run_id=1, + config=scenario._config(), + matcher=MagicMock(), + enrichment=enrichment, + shutdown_requested=lambda: False, + radius_m=1000, + anchors=scenario.anchors, + pages_per_anchor=1, + enrich_houses=scenario.enrich_houses, + detail_top_n=scenario.detail_top_n, + request_delay_sec=0.0, + enrich_imv=scenario.enrich_imv, + ) + return counters.to_dict(), _normalize(recorder.calls) + + +async def _assert_parity(scenario: _Scenario) -> dict[str, int]: + old_counters, old_calls = await _drive_old(scenario) + new_counters, new_calls = await _drive_new(scenario) + assert ( + new_counters == old_counters + ), f"counters diverged:\n old={old_counters}\n new={new_counters}" + assert ( + new_calls == old_calls + ), f"scrape_runs calls diverged:\n old={old_calls}\n new={new_calls}" + return new_counters + + +# ── scenarios ──────────────────────────────────────────────────────────────── + +_ANCHORS_2 = [(56.84, 60.60, "A1"), (56.79, 60.53, "A2")] + + +@pytest.mark.asyncio +async def test_parity_happy_path_counters_aggregation() -> None: + """2 anchor'а, SERP+save, без houses/detail/imv → mark_done + агрегированные counters.""" + scenario = _Scenario( + anchors=_ANCHORS_2, + per_anchor=[("lots", 10, 8, 2), ("lots", 6, 5, 1)], + ) + counters = await _assert_parity(scenario) + # Убеждаемся что сценарий действительно прогнал агрегацию (не пустой прогон). + assert counters["lots_fetched"] == 16 + assert counters["lots_inserted"] == 13 + assert counters["lots_updated"] == 3 + assert counters["anchors_done"] == 2 + + +@pytest.mark.asyncio +async def test_parity_full_ban_no_intake_marks_banned() -> None: + """Первый же anchor SERP-блок, лоты не сохранены → mark_banned (не done).""" + scenario = _Scenario( + anchors=_ANCHORS_2, + per_anchor=[("block",), ("block",)], + ) + await _assert_parity(scenario) + + +@pytest.mark.asyncio +async def test_parity_partial_ban_intake_marks_done() -> None: + """Anchor1 intake сохранён, anchor2 SERP-блок → partial-intake → mark_done не banned.""" + scenario = _Scenario( + anchors=_ANCHORS_2, + per_anchor=[("lots", 10, 9, 1), ("block",)], + avito_serp_ok_not_banned=True, + ) + await _assert_parity(scenario) + + +@pytest.mark.asyncio +async def test_parity_partial_ban_flag_off_marks_banned() -> None: + """avito_serp_ok_not_banned=False: даже при intake SERP-блок → mark_banned.""" + scenario = _Scenario( + anchors=_ANCHORS_2, + per_anchor=[("lots", 10, 9, 1), ("block",)], + avito_serp_ok_not_banned=False, + ) + await _assert_parity(scenario) + + +@pytest.mark.asyncio +async def test_parity_detail_consecutive_block_propagates() -> None: + """detail-фаза: 3 подряд блока (rotations=0) → propagate → anchor-handler. + + Лоты сохранены на этом же anchor'е → partial-intake → mark_done. + """ + scenario = _Scenario( + anchors=[(56.84, 60.60, "A1")], + per_anchor=[("lots", 5, 5, 0)], + detail_top_n=5, + detail_rows=3, + detail_behavior="block_all", + avito_serp_ok_not_banned=True, + avito_proxy_max_rotations=0, + ) + await _assert_parity(scenario) + + +@pytest.mark.asyncio +async def test_parity_imv_phase_counters() -> None: + """IMV-фаза: touched houses → process_houses_imv_batch → imv-counters агрегированы.""" + scenario = _Scenario( + anchors=[(56.84, 60.60, "A1")], + per_anchor=[("lots", 4, 4, 0)], + enrich_houses=True, + lots_have_house_url=True, + enrich_imv=True, + imv_result=(3, 2, 1), + ) + counters = await _assert_parity(scenario) + # touched house → IMV-фаза отработала, imv-counters агрегированы одинаково. + assert counters["imv_attempted"] == 3 + assert counters["imv_enriched"] == 2 + assert counters["imv_failed"] == 1 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py index de6f5d54..a865d384 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/contracts.py @@ -112,6 +112,28 @@ class ScraperConfig(Protocol): cookie_encryption_key: str # Sentry/GlitchTip DSN (опционально). glitchtip_dsn: str | None + # ── Оркестрация sweep-pipeline (#2135) ─────────────────────────────────── + # Поля, читаемые слоем оркестрации (scraper_kit.orchestration.pipeline). + # Собраны grep'ом `settings.` по app/services/scrape_pipeline.py. + # Проксируются той же продуктовой реализацией (RealScraperConfig). + # + # Partial-ban intake: SERP уже собрал лоты, а detail/houses заблокировало — + # run помечается 'done' (не 'banned'), intake сохранён (#1950). + avito_serp_ok_not_banned: bool + # changeip settle-пауза (секунды) после ротации IP. + avito_proxy_rotate_settle_s: float + # Ретраи changeip-GET: N попыток по proxy_rotate_attempt_timeout_s каждая (#1950). + proxy_rotate_attempts: int + proxy_rotate_attempt_timeout_s: float + # Cian rotate-прокси + лимит ротаций. + cian_proxy_rotate_url: str | None + cian_proxy_max_rotations: int + # Yandex лимит ротаций (rotate_url уже выше). + yandex_proxy_max_rotations: int + # Пропускать листинги, уже обновлённые сегодня по МСК (full-load дедуп). + scraper_skip_seen_today: bool + # Per-fetch watchdog-таймаут Cian full-load (секунды). + cian_full_load_per_fetch_timeout_s: float @runtime_checkable @@ -171,6 +193,21 @@ class EnrichmentJobs(Protocol): """Догрузить точные адреса Yandex. Проксирует backfill_yandex_addresses.""" ... + async def process_houses_imv_batch( + self, + db: Session, + house_ids: set[int], + *, + request_delay_sec: float = ..., + heartbeat: Callable[[], None] | None = ..., + ) -> Any: + """Avito IMV-оценка явного набора house_id (финальная IMV-фаза city-sweep). + + Проксирует process_houses_imv_batch. Возвращаемый результат имеет поля + .checked/.saved/.errors (тип остаётся Any — не тянем app.* в контракты). + """ + ... + __all__ = [ "EnrichmentJobs", diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/__init__.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/__init__.py new file mode 100644 index 00000000..ab31f59d --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/__init__.py @@ -0,0 +1,15 @@ +"""Оркестрация sweep-pipeline scraper_kit — strangler-копия `app.services.scrape_pipeline`. + +Координирует полный прогон скрапа (SERP → save → houses → detail → IMV) поверх +провайдеров `scraper_kit.providers.*` и базового слоя `scraper_kit.base`. Развязана +от `app.*` через Protocol-инжекцию (`scraper_kit.contracts`): + + - `app.core.config.settings` → инжектируемый `ScraperConfig` + - `app.services.matching` → инжектируемый `HouseMatcher` + - `app.services.house_imv_backfill` → инжектируемый `EnrichmentJobs` + - `app.core.shutdown.shutdown_requested` → инжектируемый `Callable[[], bool]` + - `app.services.scrape_runs` → локальная копия `scraper_kit.orchestration.runs` + +Боевой scheduler по-прежнему зовёт старый `app.services.scrape_pipeline` (strangler: +COPY, не MOVE). Переключение рантайма на этот модуль — отдельный поздний шаг. +""" diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py new file mode 100644 index 00000000..c2e5dcbf --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -0,0 +1,1412 @@ +"""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 diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py new file mode 100644 index 00000000..f8941597 --- /dev/null +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py @@ -0,0 +1,392 @@ +"""scrape_runs helpers — tracking long-running pipeline runs (strangler-копия #2135). + +Байт-эквивалент `app.services.scrape_runs` — чистые SQL-хелперы поверх таблицы +`scrape_runs` (миграции 015 + 051). Развязка от `app.*`: единственное намеренное +отличие — `sentry_sdk` импортируется опционально (kit standalone-импортируем, а +`sentry-sdk` не входит в его зависимости). Если пакет не установлен — alert-хук +best-effort no-op, поведение SQL-финализаторов идентично старому. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from sqlalchemy import text +from sqlalchemy.orm import Session + +try: # sentry опционален — kit не тянет его в зависимостях + import sentry_sdk +except ImportError: # pragma: no cover - в проде backend-env sentry_sdk присутствует + sentry_sdk = None # type: ignore[assignment] + +logger = logging.getLogger(__name__) + +# Количество последовательных неудачных запусков (failed/banned), при достижении +# которого отправляется Sentry alert. Алерт срабатывает ровно при N-й подряд ошибке +# (anti-spam: не на каждой последующей). +CONSECUTIVE_FAILURE_ALERT_THRESHOLD = 3 + + +def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]: + """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters. + + Все sweep-counters (YandexCitySweepCounters / CitySweepCounters / …) пишут число + объявлений в выдаче как ``lots_fetched`` и число впервые вставленных как + ``lots_inserted``. Раньше эти значения попадали ТОЛЬКО в counters jsonb, а + выделенные колонки total_seen/new_count оставались 0 → admin/observability + показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926). + + Приоритет ключей: + - total_seen ← 'total_seen' (если уже есть в counters) иначе 'lots_fetched' + - new_count ← 'new_count' (если уже есть) иначе 'lots_inserted' + + Возвращает (total_seen, new_count); None для ключа, которого нет в counters — + тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE). + """ + + def _pick(*keys: str) -> int | None: + for key in keys: + val = counters.get(key) + if val is not None: + try: + return int(val) + except (TypeError, ValueError): + return None + return None + + return _pick("total_seen", "lots_fetched"), _pick("new_count", "lots_inserted") + + +def _alert_if_consecutive_failures(db: Session, source: str) -> None: + """Отправить Sentry alert если последние CONSECUTIVE_FAILURE_ALERT_THRESHOLD + завершённых запусков для данного source имеют статус 'failed' или 'banned'. + + Anti-spam: алерт срабатывает ТОЛЬКО когда стрик РОВНО равен порогу — т.е. запрос + возвращает ровно N последних (failed|banned) и (N+1)-й, если существует, НЕ является + failed/banned. Это предотвращает повторный алерт на каждой ошибке сверх порога. + + Best-effort: весь блок обёрнут в try/except — сбой запроса или неинициализированный + Sentry НЕ должен нарушать вызывающий mark_* путь. + """ + if sentry_sdk is None: + return + n = CONSECUTIVE_FAILURE_ALERT_THRESHOLD + try: + # Берём последние N+1 завершённых (non-running) запусков по source. + # Сортируем по finished_at DESC чтобы самые свежие шли первыми. + rows = db.execute( + text( + """ + SELECT status FROM scrape_runs + WHERE source = :source + AND status IN ('failed', 'banned', 'done', 'cancelled') + ORDER BY finished_at DESC NULLS LAST + LIMIT :limit + """ + ), + {"source": source, "limit": n + 1}, + ).fetchall() + + if len(rows) < n: + # Ещё не набралось N завершённых запусков вообще — алерт не нужен. + return + + # Первые N должны быть все failed/banned. + first_n = rows[:n] + if not all(r.status in ("failed", "banned") for r in first_n): + return + + # (N+1)-й запуск, если есть, тоже должен НЕ быть failed/banned — иначе мы уже + # должны были отправить алерт раньше и не стоит дублировать. + if len(rows) > n and rows[n].status in ("failed", "banned"): + return + + # Стрик ровно достиг порога — отправляем алерт. + sentry_sdk.capture_message( + f"Scraper source '{source}' has {n} consecutive failed/banned runs — " + "manual intervention may be required (expired cookies / ban / broken parser).", + level="error", + ) + logger.error( + "sentry alert sent: source=%s has %d consecutive failed/banned runs", source, n + ) + except Exception: + pass # sentry_sdk not initialised in dev, or query failed — best-effort only + + +def _alert_on_run_id(db: Session, run_id: int) -> None: + """Вспомогательная обёртка: извлекает source по run_id и вызывает + _alert_if_consecutive_failures. Best-effort — не бросает исключений. + """ + try: + row = db.execute( + text("SELECT source FROM scrape_runs WHERE id = :run_id"), + {"run_id": run_id}, + ).fetchone() + if row is None: + return + _alert_if_consecutive_failures(db, str(row.source)) + except Exception: + pass + + +def create_run(db: Session, *, source: str, params: dict[str, Any]) -> int: + """INSERT scrape_runs(source, status='running', params, started_at=NOW()). + + run_type DEFAULT 'city_sweep' (из 051 миграции). + Returns run_id (bigint). + """ + row = db.execute( + text( + """ + INSERT INTO scrape_runs (source, status, params, started_at, heartbeat_at) + VALUES (:source, 'running', CAST(:params AS jsonb), NOW(), NOW()) + RETURNING id + """ + ), + {"source": source, "params": json.dumps(params, ensure_ascii=False)}, + ).fetchone() + db.commit() + assert row is not None, "scrape_runs INSERT returned no id" + return int(row.id) + + +def update_heartbeat(db: Session, run_id: int, counters: dict[str, int]) -> None: + """UPDATE heartbeat_at=NOW(), counters=:counters + total_seen/new_count колонки. + + total_seen/new_count извлекаются из counters (lots_fetched/lots_inserted) и + пишутся в выделенные колонки, чтобы observability не показывала 0 (audit #1926). + COALESCE: если ключа нет в counters — старое значение колонки сохраняется. + """ + total_seen, new_count = _column_counts(counters) + db.execute( + text( + """ + UPDATE scrape_runs + SET heartbeat_at = NOW(), + counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) + WHERE id = :run_id + """ + ), + { + "run_id": run_id, + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, + ) + db.commit() + + +def is_cancelled(db: Session, run_id: int) -> bool: + """Проверить status='cancelled' (cooperative cancel в long-running pipeline).""" + row = db.execute( + text("SELECT status FROM scrape_runs WHERE id = :id"), + {"id": run_id}, + ).fetchone() + return row is not None and row.status == "cancelled" + + +def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None: + """Финализация run: status='done', finished_at=NOW(), counters + total_seen/new_count. + + total_seen/new_count извлекаются из counters (lots_fetched/lots_inserted) и пишутся + в выделенные колонки — иначе admin/observability показывает 0 (audit #1926). + """ + total_seen, new_count = _column_counts(counters) + row = db.execute( + text( + """ + UPDATE scrape_runs + SET status = 'done', finished_at = NOW(), heartbeat_at = NOW(), + counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) + WHERE id = :run_id AND status = 'running' + RETURNING id + """ + ), + { + "run_id": run_id, + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, + ).first() + if row is None: + logger.warning("mark_done no-op: run_id=%d not in 'running' state", run_id) + db.commit() + + +def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None: + """Финализация run: status='failed', error (первые 1000 символов). + + Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE, + он мог оставить сессию в error state — rollback сбрасывает состояние. + """ + try: + db.rollback() # defensive: cascade-safe if prior UPDATE left txn in error state + except Exception: + pass + total_seen, new_count = _column_counts(counters) + row = db.execute( + text( + """ + UPDATE scrape_runs + SET status = 'failed', finished_at = NOW(), heartbeat_at = NOW(), + error = :error, counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) + WHERE id = :run_id AND status = 'running' + RETURNING id + """ + ), + { + "run_id": run_id, + "error": error[:1000], + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, + ).first() + if row is None: + logger.warning("mark_failed no-op: run_id=%d not in 'running' state", run_id) + db.commit() + # Алерт после коммита — статус уже персистирован в БД; best-effort (не бросает). + _alert_on_run_id(db, run_id) + + +def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) -> None: + """Финализация run: status='banned' (IP заблокирован Avito — 403/captcha). + + Per migration 015 — 'banned' задокументирован как 'Avito вернул 403/captcha'. + Отличается от 'failed': это external constraint, не наш bug. Cooldown 2-4 часа. + + Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE, + он мог оставить сессию в error state — rollback сбрасывает состояние. + """ + try: + db.rollback() # defensive: cascade-safe if prior UPDATE left txn in error state + except Exception: + pass + total_seen, new_count = _column_counts(counters) + row = db.execute( + text( + """ + UPDATE scrape_runs + SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(), + error = :error, counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) + WHERE id = :run_id AND status = 'running' + RETURNING id + """ + ), + { + "run_id": run_id, + "error": error[:1000], + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, + ).first() + if row is None: + logger.warning("mark_banned no-op: run_id=%d not in 'running' state", run_id) + db.commit() + # Алерт после коммита — статус уже персистирован в БД; best-effort (не бросает). + _alert_on_run_id(db, run_id) + + +def mark_cancelled(db: Session, run_id: int) -> bool: + """Set status='cancelled' если currently 'running'. Returns True если cancelled.""" + result = db.execute( + text( + """ + UPDATE scrape_runs + SET status = 'cancelled', finished_at = NOW() + WHERE id = :run_id AND status = 'running' + RETURNING id + """ + ), + {"run_id": run_id}, + ).fetchone() + db.commit() + return result is not None + + +def list_recent(db: Session, *, source: str, limit: int = 10) -> list[dict[str, Any]]: + """Список последних N runs по source, в порядке убывания started_at.""" + rows = ( + db.execute( + text( + """ + SELECT id AS run_id, source, status, params, counters, error, + started_at, finished_at, heartbeat_at + FROM scrape_runs + WHERE source = :source + ORDER BY started_at DESC NULLS LAST + LIMIT :limit + """ + ), + {"source": source, "limit": limit}, + ) + .mappings() + .all() + ) + return [dict(r) for r in rows] + + +def list_all( + db: Session, + *, + source: str | None = None, + status: str | None = None, + limit: int = 50, + offset: int = 0, +) -> tuple[int, list[dict[str, Any]]]: + """Unified-список runs по всем source'ам с опц. фильтрами source/status. + + Возвращает (total, rows): total — кол-во строк под фильтром (без limit/offset), + rows — текущая страница (ORDER BY started_at DESC, LIMIT/OFFSET). + + Используется единой scrapers-страницей вместо 3 per-source /runs. + """ + where = ["1=1"] + params: dict[str, Any] = {"limit": limit, "offset": offset} + if source is not None: + where.append("source = :source") + params["source"] = source + if status is not None: + where.append("status = :status") + params["status"] = status + where_sql = " AND ".join(where) + + total_row = db.execute( + text(f"SELECT COUNT(*) AS n FROM scrape_runs WHERE {where_sql}"), + params, + ).fetchone() + total = int(total_row.n) if total_row is not None else 0 + + rows = ( + db.execute( + text( + f""" + SELECT id AS run_id, source, run_type, status, params, counters, + total_seen, new_count, started_at, finished_at, + heartbeat_at, error AS error_text + FROM scrape_runs + WHERE {where_sql} + ORDER BY started_at DESC NULLS LAST + LIMIT CAST(:limit AS int) OFFSET CAST(:offset AS int) + """ + ), + params, + ) + .mappings() + .all() + ) + return total, [dict(r) for r in rows]