diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 50635e45..9a9056a6 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/tradein-mvp/backend/app/api/v1/admin.py @@ -2239,6 +2239,10 @@ class UnifiedScrapeRunRow(BaseModel): # не опрашивает (см. scrape_runs.honors_cancel): статус соврал бы, а # has_running_run перестал бы держать single-run guard. cancellable: bool = False + # #2686: диагноз для status='banned' — 'platform' (площадка заблокировала) или + # 'infra' (не отдал наш браузерный сайдкар). Без него оператор видит только + # «забанен» и делает вывод «площадка нас палит» на 80% наших же отказов. + ban_kind: str | None = None params: dict | None = None counters: dict | None = None total_seen: int | None = None @@ -2346,6 +2350,7 @@ def list_scrape_runs_unified( source=r["source"], status=r["status"], cancellable=runs_mod.honors_cancel(str(r["source"])), + ban_kind=r.get("ban_kind"), params=r.get("params"), counters=r.get("counters"), total_seen=r.get("total_seen"), diff --git a/tradein-mvp/backend/app/services/scrape_runs.py b/tradein-mvp/backend/app/services/scrape_runs.py index 9a6a0af3..154d19a3 100644 --- a/tradein-mvp/backend/app/services/scrape_runs.py +++ b/tradein-mvp/backend/app/services/scrape_runs.py @@ -29,6 +29,22 @@ CONSECUTIVE_FAILURE_ALERT_THRESHOLD = 3 # невидим для CONSECUTIVE_FAILURE_ALERT_THRESHOLD (тот считает только failed/banned). CONSECUTIVE_ZERO_RESULT_ALERT_THRESHOLD = 3 +# #2686: диагноз оборванного прогона. Пишется в scrape_runs.ban_kind (миграция 218) +# РЯДОМ со status='banned', а не ВМЕСТО него — сознательный выбор между «новый +# статус» и «явное поле причины»: +# 1. Побочная функция 'banned' — сохранение done_buckets-чекпоинта (mark_failed +# его теряет) — нужна ОБОИМ исходам. Оставив статус, получаем её даром; расщепив +# статус, пришлось бы дублировать её в каждом потребителе. +# 2. Новое значение статуса пришлось бы доучить пяти местам, каждое из которых +# молча даёт неверный ответ, если про него забыть: CHECK-констрейнт схемы, +# IN-списки обоих сторожей (_alert_if_consecutive_failures / _zero_results), +# Literal-фильтр admin API и хардкод-список статусов во фронте. Это ровно тот +# класс оборванной проводки, из-за которого задача и появилась. +# 3. Прогон в обоих случаях требует одного и того же обращения (оборвать, сохранить +# частичное); различается только ДИАГНОЗ — то есть метаданное, не состояние. +BAN_KIND_PLATFORM = "platform" # площадка показала firewall/403/captcha — внешнее +BAN_KIND_INFRA = "infra" # наш сайдкар/прокси не отдал страницу — внутреннее + def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]: """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters. @@ -353,11 +369,29 @@ def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) _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). +def mark_banned( + db: Session, + run_id: int, + error: str, + counters: dict[str, int], + *, + ban_kind: str = BAN_KIND_PLATFORM, +) -> None: + """Финализация run: status='banned' + диагноз ban_kind (#2686). Per migration 015 — 'banned' задокументирован как 'Avito вернул 403/captcha'. - Отличается от 'failed': это external constraint, не наш bug. Cooldown 2-4 часа. + Отличается от 'failed': прогон оборван внешним/блокирующим условием, а не нашим + багом, и — важно — СОХРАНЯЕТ done_buckets-чекпоинт в counters (mark_failed его + теряет). Cooldown 2-4 часа. + + `ban_kind` разводит два исхода, которые раньше схлопывались в один статус: + - BAN_KIND_PLATFORM — площадка нас заблокировала (firewall/403/captcha); + - BAN_KIND_INFRA — упала НАША инфраструктура (браузерный сайдкар/прокси). + Значение приходит от места ПОРОЖДЕНИЯ отказа (тип исключения), а не из разбора + текста ошибки. Default 'platform' = историческая семантика статуса, поэтому + вызывающие, которым разводить нечего, не меняются. + + Оба исхода одинаково сохраняют чекпоинт — они отличаются только диагнозом. Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE, он мог оставить сессию в error state — rollback сбрасывает состояние. @@ -373,6 +407,7 @@ def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) UPDATE scrape_runs SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(), error = :error, counters = CAST(:counters AS jsonb), + ban_kind = :ban_kind, 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' @@ -383,6 +418,7 @@ def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) "run_id": run_id, "error": error[:1000], "counters": json.dumps(counters), + "ban_kind": ban_kind, "total_seen": total_seen, "new_count": new_count, }, @@ -549,7 +585,7 @@ def list_all( text( f""" SELECT id AS run_id, source, status, params, counters, - total_seen, new_count, started_at, finished_at, + ban_kind, total_seen, new_count, started_at, finished_at, heartbeat_at, error AS error_text FROM scrape_runs WHERE {where_sql} diff --git a/tradein-mvp/backend/data/sql/218_scrape_runs_ban_kind.sql b/tradein-mvp/backend/data/sql/218_scrape_runs_ban_kind.sql new file mode 100644 index 00000000..689cee00 --- /dev/null +++ b/tradein-mvp/backend/data/sql/218_scrape_runs_ban_kind.sql @@ -0,0 +1,72 @@ +-- 218_scrape_runs_ban_kind.sql +-- scrape_runs.ban_kind — статус 'banned' перестаёт смешивать наш сбой с чужим (#2686). +-- +-- Dependencies: 015_scrape_runs.sql (создала таблицу и CHECK по status). +-- Apply after: 217_position_in_serp_unexpressible.sql +-- Идемпотентно: ADD COLUMN IF NOT EXISTS + DROP/ADD CONSTRAINT + UPDATE только там, +-- где ban_kind ещё NULL. +-- +-- ── ЧТО НАЙДЕНО ────────────────────────────────────────────────────────────── +-- Из 115 avito-прогонов со статусом 'banned' (замер на проде 2026-08-06): +-- 90 — «browser unavailable (proxy may be down)» — 503 НАШЕГО браузерного +-- сайдкара (05.07-03.08), площадка ни при чём; +-- 2 — прочие ошибки того же сайдкара (04.07, 15.07); +-- 10 — «Avito SERP firewall (browser-mode) … IP banned» — реальная блокировка +-- площадкой (16.06-06.08); +-- 11 — HTTP 429/403 (30.05-19.06); +-- 2 — прочее (21.06, 23.06). +-- То есть 92 из 115 (80%) прогонов, помеченных «нас забанили», — отказ нашей +-- собственной инфраструктуры. Оба исхода писали РАЗНЫЙ текст ошибки и получали +-- ОДИН статус: различитель лежал в данных и терялся ровно в момент присвоения. +-- +-- ── ЦЕНА, КОТОРАЯ УЖЕ УПЛАЧЕНА ─────────────────────────────────────────────── +-- Миграция 206 прочитала статус буквально — как «площадка распознаёт наш паттерн» — +-- и на этом основании замедлила avito_full_load_exhaustive более чем вдвое, переведя +-- на недельный такт. Основание было ложным. Возврат такта сюда НЕ входит: он вынесен +-- в #2687 на данные 9-10.08. +-- +-- ── ПОЧЕМУ КОЛОНКА, А НЕ НОВЫЙ СТАТУС ──────────────────────────────────────── +-- 1. У 'banned' есть побочная функция: в отличие от 'failed' он СОХРАНЯЕТ +-- done_buckets-чекпоинт пройденных бакетов. Она нужна обоим исходам — следующий +-- прогон не должен начинать с нуля ни при нашем отказе, ни при блокировке. +-- Оставив статус, получаем её даром. +-- 2. Новое значение статуса пришлось бы доучить пяти местам, каждое из которых +-- молча даёт неверный ответ, если про него забыть: этот CHECK, IN-списки обоих +-- сторожей в orchestration/runs.py, Literal-фильтр admin API и хардкод-список +-- статусов во фронте (RunsTable.tsx). Это ровно тот класс оборванной проводки, +-- из-за которого задача и появилась. +-- 3. Прогон в обоих случаях требует одного обращения (оборвать, сохранить частичное); +-- различается только ДИАГНОЗ — метаданное, не состояние. +-- +-- В рантайме значение несётся от МЕСТА ПОРОЖДЕНИЯ отказа (тип исключения +-- AvitoSidecarUnavailableError), а не разбирается из текста ошибки постфактум. +-- Разбор текста ниже — РАЗОВАЯ ретро-классификация уже накопленной истории; для +-- новых строк этот путь не используется. + +ALTER TABLE scrape_runs + ADD COLUMN IF NOT EXISTS ban_kind text; + +COMMENT ON COLUMN scrape_runs.ban_kind IS + 'Диагноз status=''banned'' (#2686): platform — площадка заблокировала ' + '(firewall/403/captcha); infra — не отдала НАША инфраструктура (браузерный ' + 'сайдкар/прокси). NULL для прогонов с другим статусом. Пишется из типа ' + 'исключения в момент отказа, не из текста ошибки.'; + +ALTER TABLE scrape_runs DROP CONSTRAINT IF EXISTS scrape_runs_ban_kind_check; +ALTER TABLE scrape_runs + ADD CONSTRAINT scrape_runs_ban_kind_check + CHECK (ban_kind IS NULL OR ban_kind IN ('platform', 'infra')); + +-- Ретро-классификация истории (разово, только там, где ещё NULL). Порядок веток +-- важен: маркер сайдкара проверяется первым, потому что текст полного обхода +-- оборачивает его в свой префикс («avito full load aborted: avito SERP +-- browser-sidecar error (page=1): browser unavailable (proxy may be down)»). +UPDATE scrape_runs +SET ban_kind = CASE + WHEN error ILIKE '%browser-sidecar error%' + OR error ILIKE '%browser unavailable%' + OR error ILIKE '%proxy may be down%' THEN 'infra' + ELSE 'platform' + END +WHERE status = 'banned' + AND ban_kind IS NULL; diff --git a/tradein-mvp/backend/tests/test_2686_ban_kind_split.py b/tradein-mvp/backend/tests/test_2686_ban_kind_split.py new file mode 100644 index 00000000..1b4e1121 --- /dev/null +++ b/tradein-mvp/backend/tests/test_2686_ban_kind_split.py @@ -0,0 +1,246 @@ +"""#2686: статус 'banned' перестаёт смешивать наш сбой с блокировкой площадкой. + +Замер на проде (2026-08-06): из 115 avito-прогонов со статусом «забанен» 92 (80%) — +отказ НАШЕГО браузерного сайдкара («browser unavailable (proxy may be down)»), и +только 10 — реальный firewall площадки. По этому статусу приняли решение замедлить +avito_full_load_exhaustive более чем вдвое (миграция 206) — основание было ложным. + +Различитель несётся от МЕСТА ПОРОЖДЕНИЯ отказа (тип исключения), а не разбирается +из текста ошибки постфактум. Проверяем три звена цепочки: + 1. serp.py поднимает AvitoSidecarUnavailableError там и только там, где виноваты мы; + 2. pipeline.ban_kind_of_exception переводит тип в диагноз; + 3. run_avito_full_load доносит диагноз до mark_banned — и при ЭТОМ сохраняет + done_buckets-чекпоинт (побочная функция 'banned', которую нельзя потерять). +""" + +from __future__ import annotations + +import os +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db") + +from scraper_kit.avito_exceptions import ( + AvitoBlockedError, + AvitoRateLimitedError, + AvitoSidecarUnavailableError, +) +from scraper_kit.orchestration import runs as kit_runs +from scraper_kit.orchestration.pipeline import ( + ban_kind_of_exception, + run_avito_full_load, +) +from scraper_kit.providers.avito import serp as serp_module +from scraper_kit.providers.avito.serp import AvitoScraper + +from app.services.scraper_adapters import RealScraperConfig + +PFX = "scraper_kit.orchestration.pipeline" + +# Настоящий firewall-интерстициал Авито: _is_firewall_page ловит его по тексту. +_FIREWALL_HTML = "Доступ ограничен: проблема с IP" + + +def _sidecar_503() -> httpx.HTTPStatusError: + """Ровно то, что прод писал 90 раз с 05.07 по 03.08.""" + request = httpx.Request("POST", "http://tradein-browser:3000/fetch") + response = httpx.Response( + 503, json={"error": "browser unavailable (proxy may be down)"}, request=request + ) + return httpx.HTTPStatusError("503", request=request, response=response) + + +# ── 1. Место порождения ────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_sidecar_failure_raises_own_infra_subtype() -> None: + """503 сайдкара → AvitoSidecarUnavailableError, и он ОСТАЁТСЯ AvitoRateLimitedError. + + Фальсификация: до правки поднимался голый AvitoRateLimitedError — первый assert + падает. Второй assert стережёт совместимость: все существующие + `except (AvitoBlockedError, AvitoRateLimitedError)` обязаны продолжать ловить + этот отказ, иначе прогон уйдёт в mark_failed и потеряет чекпоинт. + """ + scraper = AvitoScraper(RealScraperConfig()) + scraper._browser = AsyncMock() + scraper._browser.fetch = AsyncMock(side_effect=_sidecar_503()) + + with patch.object(serp_module.asyncio, "sleep", AsyncMock()): + with pytest.raises(AvitoSidecarUnavailableError) as excinfo: + await scraper._fetch_serp_html("https://www.avito.ru/ekaterinburg/kvartiry", page=1) + + assert isinstance(excinfo.value, AvitoRateLimitedError) + + +@pytest.mark.asyncio +async def test_platform_firewall_is_not_our_infra() -> None: + """Рендер firewall-страницы → AvitoBlockedError, и это НЕ наш инфра-подтип.""" + scraper = AvitoScraper(RealScraperConfig()) + scraper._browser = AsyncMock() + scraper._browser.fetch = AsyncMock(return_value=_FIREWALL_HTML) + scraper._browser.report_ban = MagicMock() # sync-хук, не корутина + + with pytest.raises(AvitoBlockedError) as excinfo: + await scraper._fetch_serp_html("https://www.avito.ru/ekaterinburg/kvartiry", page=1) + + assert not isinstance(excinfo.value, AvitoSidecarUnavailableError) + + +# ── 2. Перевод типа в диагноз ──────────────────────────────────────────────── + + +@pytest.mark.parametrize( + ("exc", "expected"), + [ + (AvitoSidecarUnavailableError("sidecar down"), kit_runs.BAN_KIND_INFRA), + (AvitoBlockedError("firewall"), kit_runs.BAN_KIND_PLATFORM), + (AvitoRateLimitedError("HTTP 429"), kit_runs.BAN_KIND_PLATFORM), + (RuntimeError("что угодно ещё"), kit_runs.BAN_KIND_PLATFORM), + ], +) +def test_ban_kind_of_exception(exc: BaseException, expected: str) -> None: + """Честный 429 самой площадки остаётся 'platform' — инфра-диагноз даёт только подтип.""" + assert ban_kind_of_exception(exc) == expected + + +# ── 3. Диагноз доезжает до строки прогона, чекпоинт цел ────────────────────── + + +class _BanRecorder: + """Минимальный двойник scrape_runs: интересуют ban_kind и counters у mark_banned.""" + + def __init__(self) -> None: + self.banned: list[tuple[str, dict[str, Any]]] = [] + self.other: list[str] = [] + + 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: + pass + + def mark_done(self, db: Any, run_id: int, counters: dict[str, Any]) -> None: + self.other.append("mark_done") + + def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: + self.other.append("mark_failed") + + def mark_banned( + self, + db: Any, + run_id: int, + error: str, + counters: dict[str, Any], + *, + ban_kind: str = kit_runs.BAN_KIND_PLATFORM, + ) -> None: + self.banned.append((ban_kind, dict(counters))) + + +def _config() -> SimpleNamespace: + return SimpleNamespace( + scraper_fetch_mode="curl_cffi", + browser_http_endpoint="http://browser.test/fetch", + scraper_proxy_url=None, + avito_proxy_max_rotations=0, + avito_serp_ok_not_banned=True, + avito_proxy_rotate_settle_s=0.0, + proxy_rotate_attempts=1, + proxy_rotate_attempt_timeout_s=1.0, + avito_full_load_price_cap=1400, + avito_full_load_max_pages=100, + avito_full_load_concurrency=1, + request_delay_sec=0.0, + scraper_skip_seen_today=False, + ) + + +def _scraper_that_saves_one_bucket_then(exc: Exception) -> MagicMock: + """Fake AvitoScraper: один бакет успешно отдан в on_bucket, затем отказ.""" + + async def _fetch(*_a: Any, on_bucket: Any = None, **_k: Any) -> None: + on_bucket("2к:0-5m", [MagicMock(source_id="a1")]) + raise exc + + m = MagicMock() + m.__aenter__ = AsyncMock(return_value=m) + m.__aexit__ = AsyncMock(return_value=None) + m.fetch_all_secondary = _fetch + m._browser = None + return m + + +async def _drive_full_load(exc: Exception) -> _BanRecorder: + recorder = _BanRecorder() + with ( + patch(f"{PFX}.AvitoScraper", return_value=_scraper_that_saves_one_bucket_then(exc)), + patch(f"{PFX}.save_listings", MagicMock(return_value=(1, 0))), + patch(f"{PFX}.runs", recorder), + ): + await run_avito_full_load(MagicMock(), run_id=1, config=_config(), matcher=MagicMock()) + return recorder + + +@pytest.mark.asyncio +async def test_full_load_sidecar_ban_is_infra_and_keeps_checkpoint() -> None: + """Отказ нашего сайдкара → ban_kind='infra', done_buckets сохранён. + + Фальсификация: до правки mark_banned не принимал ban_kind вовсе — вызов падал + бы на TypeError; с ban_kind, но без проброса из pipeline, диагноз был бы + 'platform' (дефолт), то есть ровно то враньё, которое стоило миграции 206. + """ + recorder = await _drive_full_load( + AvitoSidecarUnavailableError("avito SERP browser-sidecar error (page=1)") + ) + + assert len(recorder.banned) == 1 + ban_kind, counters = recorder.banned[0] + assert ban_kind == kit_runs.BAN_KIND_INFRA + assert counters["done_buckets"] == ["2к:0-5m"] + + +@pytest.mark.asyncio +async def test_full_load_platform_ban_keeps_the_same_checkpoint() -> None: + """Блокировка площадкой → ban_kind='platform', чекпоинт сохраняется ТАК ЖЕ. + + Второй исход не должен потерять побочную функцию 'banned' при разведении. + """ + recorder = await _drive_full_load(AvitoBlockedError("Avito SERP firewall — IP banned")) + + assert len(recorder.banned) == 1 + ban_kind, counters = recorder.banned[0] + assert ban_kind == kit_runs.BAN_KIND_PLATFORM + assert counters["done_buckets"] == ["2к:0-5m"] + + +# ── 4. Диагноз доходит до SQL ──────────────────────────────────────────────── + + +@pytest.mark.parametrize("module_name", ["kit", "app"]) +@pytest.mark.parametrize( + ("kwargs", "expected"), + [({}, "platform"), ({"ban_kind": "infra"}, "infra")], +) +def test_mark_banned_writes_ban_kind( + module_name: str, kwargs: dict[str, str], expected: str +) -> None: + """Обе копии runs-модуля кладут ban_kind в UPDATE (default — историческое 'platform').""" + if module_name == "kit": + mod: Any = kit_runs + else: + from app.services import scrape_runs as app_runs + + mod = app_runs + + db = MagicMock() + mod.mark_banned(db, 1, "err", {"lots_fetched": 0}, **kwargs) + + sql, params = db.execute.call_args_list[0][0] # первый UPDATE, не alert-запрос после + assert "ban_kind = :ban_kind" in str(sql) + assert params["ban_kind"] == expected diff --git a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py index 7da8983b..f0615237 100644 --- a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py +++ b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity.py @@ -43,6 +43,7 @@ class _RunsRecorder: def __init__(self) -> None: self.calls: list[tuple[str, dict[str, Any]]] = [] + self.ban_kinds: list[str] = [] def is_cancelled(self, db: Any, run_id: int) -> bool: return False @@ -53,7 +54,16 @@ class _RunsRecorder: 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: + def mark_banned( + self, + db: Any, + run_id: int, + error: str, + counters: dict[str, Any], + *, + ban_kind: str = "platform", + ) -> None: + self.ban_kinds.append(ban_kind) # #2686: диагноз, не статус self.calls.append(("mark_banned", dict(counters))) def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: diff --git a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py index 248ebe81..37a1278b 100644 --- a/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py +++ b/tradein-mvp/backend/tests/test_scraper_kit_pipeline_parity2.py @@ -52,6 +52,7 @@ class _RunsRecorder: def __init__(self) -> None: self.calls: list[tuple[str, dict[str, Any]]] = [] + self.ban_kinds: list[str] = [] def is_cancelled(self, db: Any, run_id: int) -> bool: return False @@ -62,7 +63,16 @@ class _RunsRecorder: 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: + def mark_banned( + self, + db: Any, + run_id: int, + error: str, + counters: dict[str, Any], + *, + ban_kind: str = "platform", + ) -> None: + self.ban_kinds.append(ban_kind) # #2686: диагноз, не статус self.calls.append(("mark_banned", dict(counters))) def mark_failed(self, db: Any, run_id: int, error: str, counters: dict[str, Any]) -> None: diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py index 98557d30..18e5d5bf 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/avito_exceptions.py @@ -22,6 +22,25 @@ class AvitoRateLimitedError(AvitoError): """HTTP 429 от Avito — rate limit triggered.""" +class AvitoSidecarUnavailableError(AvitoRateLimitedError): + """НАШ браузерный сайдкар не отдал страницу — площадка ни при чём (#2686). + + Поднимается только там, где отказ ПОРОЖДЁН нашей инфраструктурой: sidecar + вернул 503 «browser unavailable», упал внутри (Page.goto Timeout, браузер не + поднялся) либо соединение до него оборвалось (ReadTimeout/ConnectError). Это + отличается от AvitoBlockedError (площадка показала firewall) и от честного + HTTP 429 самой площадки, хотя все три раньше давали ОДИН статус 'banned' — + 92 прогона из 114 (81%) за 45 суток были нашим сайдкаром, и по этому статусу + приняли решение замедлить сбор вдвое (миграция 206). + + Наследует AvitoRateLimitedError НАМЕРЕННО: всё поведение прогона обязано + остаться прежним (`except (AvitoBlockedError, AvitoRateLimitedError)` в + pipeline/full-load/backfill ловят его без изменений → mark_banned → + done_buckets-чекпоинт сохраняется). Разводятся не пути, а ДИАГНОЗ: + pipeline пишет его в scrape_runs.ban_kind ('infra' vs 'platform'). + """ + + class AvitoListingGoneError(AvitoError): """Объявление удалено / вернуло 404 (страница «Ошибка 404» вместо item-view). 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 index c571f75d..6c35bc08 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -46,10 +46,17 @@ 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.avito_exceptions import ( + AvitoBlockedError, + AvitoRateLimitedError, + AvitoSidecarUnavailableError, +) from scraper_kit.base import ScrapedLot, save_listings from scraper_kit.browser_fetcher import BrowserFetcher from scraper_kit.orchestration import runs +# Константы диагноза берём напрямую, а не через `runs.` — helper ниже обязан +# работать и когда тесты подменяют весь модуль runs двойником (#2686). +from scraper_kit.orchestration.runs import BAN_KIND_INFRA, BAN_KIND_PLATFORM from scraper_kit.providers.avito.detail import fetch_detail, save_detail_enrichment from scraper_kit.providers.avito.houses import ( fetch_house_catalog, @@ -184,6 +191,20 @@ _AVITO_DETAIL_CONSECUTIVE_TIMEOUT_ABORT: int = 5 # прерывают enrichment-фазу. Listings из SEARCH+SAVE сохраняются всегда. _AVITO_PIPELINE_CONSECUTIVE_BLOCK_ABORT: int = 3 + +def ban_kind_of_exception(exc: BaseException) -> str: + """Диагноз для scrape_runs.ban_kind по ТИПУ исключения, не по тексту (#2686). + + AvitoSidecarUnavailableError поднимается ровно там, где отказ породила наша + инфраструктура (браузерный сайдкар не отдал страницу) — всё остальное, что + доводит прогон до 'banned' (firewall, 403/captcha, честный 429 площадки, N + подряд заблокированных якорей), внешнее. + + Один узел на все avito-сайты mark_banned: разводить исход в каждом из трёх + было бы тремя копиями одного условия. + """ + return BAN_KIND_INFRA if isinstance(exc, AvitoSidecarUnavailableError) else BAN_KIND_PLATFORM + # #2160: константы для расчёта watchdog-таймаута Cian city sweep. При # USE_PROXY_POOL_BROWSER=true каждый SERP-фетч идёт через camoufox с relaunch при смене # прокси (page.goto timeout 60s + overhead) = 13-45s/страница, а якорь cian = 4 room-buckets @@ -1544,7 +1565,13 @@ async def run_avito_city_sweep( {**counters.to_dict(), "enrichment_abort_note": _note}, # type: ignore[arg-type] ) else: - runs.mark_banned(db, run_id, str(e), counters.to_dict()) + runs.mark_banned( + db, + run_id, + str(e), + counters.to_dict(), + ban_kind=ban_kind_of_exception(e), + ) return counters except Exception: logger.exception("city-sweep run_id=%d: anchor %s failed", run_id, name) @@ -1752,7 +1779,9 @@ async def run_avito_newbuilding_sweep( ) except (AvitoBlockedError, AvitoRateLimitedError) as e: logger.error("nb-sweep run_id=%d: SERP BLOCKED — %s", run_id, e) - runs.mark_banned(db, run_id, str(e), counters.to_dict()) + runs.mark_banned( + db, run_id, str(e), counters.to_dict(), ban_kind=ban_kind_of_exception(e) + ) return counters counters.lots_fetched += len(lots) @@ -3616,7 +3645,11 @@ async def run_avito_full_load( db, run_id, f"avito full load aborted: {exc}", + # done_buckets — чекпоинт пройденных бакетов; он сохраняется ОДИНАКОВО + # при обоих ban_kind (#2686 треб. 2): следующий прогон не начинает с нуля + # ни при нашем отказе сайдкара, ни при блокировке площадкой. {**counters.to_dict(), "done_buckets": sorted(done)}, + ban_kind=ban_kind_of_exception(exc), ) return counters 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 index fd5fa55f..4aa54fa2 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/runs.py @@ -40,6 +40,22 @@ CONSECUTIVE_FAILURE_ALERT_THRESHOLD = 3 # невидим для CONSECUTIVE_FAILURE_ALERT_THRESHOLD (тот считает только failed/banned). CONSECUTIVE_ZERO_RESULT_ALERT_THRESHOLD = 3 +# #2686: диагноз оборванного прогона. Пишется в scrape_runs.ban_kind (миграция 218) +# РЯДОМ со status='banned', а не ВМЕСТО него — сознательный выбор между «новый +# статус» и «явное поле причины»: +# 1. Побочная функция 'banned' — сохранение done_buckets-чекпоинта (mark_failed +# его теряет) — нужна ОБОИМ исходам. Оставив статус, получаем её даром; расщепив +# статус, пришлось бы дублировать её в каждом потребителе. +# 2. Новое значение статуса пришлось бы доучить пяти местам, каждое из которых +# молча даёт неверный ответ, если про него забыть: CHECK-констрейнт схемы, +# IN-списки обоих сторожей (_alert_if_consecutive_failures / _zero_results), +# Literal-фильтр admin API и хардкод-список статусов во фронте. Это ровно тот +# класс оборванной проводки, из-за которого задача и появилась. +# 3. Прогон в обоих случаях требует одного и того же обращения (оборвать, сохранить +# частичное); различается только ДИАГНОЗ — то есть метаданное, не состояние. +BAN_KIND_PLATFORM = "platform" # площадка показала firewall/403/captcha — внешнее +BAN_KIND_INFRA = "infra" # наш сайдкар/прокси не отдал страницу — внутреннее + def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]: """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters. @@ -436,11 +452,29 @@ def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) _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). +def mark_banned( + db: Session, + run_id: int, + error: str, + counters: dict[str, int], + *, + ban_kind: str = BAN_KIND_PLATFORM, +) -> None: + """Финализация run: status='banned' + диагноз ban_kind (#2686). Per migration 015 — 'banned' задокументирован как 'Avito вернул 403/captcha'. - Отличается от 'failed': это external constraint, не наш bug. Cooldown 2-4 часа. + Отличается от 'failed': прогон оборван внешним/блокирующим условием, а не нашим + багом, и — важно — СОХРАНЯЕТ done_buckets-чекпоинт в counters (mark_failed его + теряет). Cooldown 2-4 часа. + + `ban_kind` разводит два исхода, которые раньше схлопывались в один статус: + - BAN_KIND_PLATFORM — площадка нас заблокировала (firewall/403/captcha); + - BAN_KIND_INFRA — упала НАША инфраструктура (браузерный сайдкар/прокси). + Значение приходит от места ПОРОЖДЕНИЯ отказа (тип исключения), а не из разбора + текста ошибки. Default 'platform' = историческая семантика статуса, поэтому + вызывающие, которым разводить нечего, не меняются. + + Оба исхода одинаково сохраняют чекпоинт — они отличаются только диагнозом. Defensive rollback: если до этого вызова в той же транзакции был ошибочный UPDATE, он мог оставить сессию в error state — rollback сбрасывает состояние. @@ -456,6 +490,7 @@ def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) UPDATE scrape_runs SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(), error = :error, counters = CAST(:counters AS jsonb), + ban_kind = :ban_kind, 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' @@ -466,6 +501,7 @@ def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) "run_id": run_id, "error": error[:1000], "counters": json.dumps(counters), + "ban_kind": ban_kind, "total_seen": total_seen, "new_count": new_count, }, @@ -552,7 +588,7 @@ def list_all( text( f""" SELECT id AS run_id, source, status, params, counters, - total_seen, new_count, started_at, finished_at, + ban_kind, total_seen, new_count, started_at, finished_at, heartbeat_at, error AS error_text FROM scrape_runs WHERE {where_sql} diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py index 2b05b055..484403c3 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/avito/serp.py @@ -42,6 +42,7 @@ from scraper_kit.avito_exceptions import ( AvitoBlockedError, AvitoContentBlockedError, AvitoRateLimitedError, + AvitoSidecarUnavailableError, ) from scraper_kit.base import BaseScraper, ScrapedLot from scraper_kit.browser_fetcher import BrowserFetcher @@ -586,7 +587,11 @@ class AvitoScraper(BaseScraper): url, error_text[:200], ) - raise AvitoRateLimitedError( + # #2686: подтип AvitoRateLimitedError — поведение прогона прежнее + # (mark_banned + done_buckets-чекпоинт), но диагноз честный: это НАШ + # сайдкар, а не площадка. Признак несётся отсюда, с места порождения, + # а не разбирается из текста ошибки постфактум. + raise AvitoSidecarUnavailableError( f"avito SERP browser-sidecar error (page={page}): {error_text[:200]}" ) from exc # Пустой рендер ("" / None) = таймаут рендера sidecar'а, НЕ firewall: