diff --git a/tradein-mvp/backend/data/sql/215_avito_full_load_window_matches_cadence.sql b/tradein-mvp/backend/data/sql/215_avito_full_load_window_matches_cadence.sql new file mode 100644 index 00000000..03b78b11 --- /dev/null +++ b/tradein-mvp/backend/data/sql/215_avito_full_load_window_matches_cadence.sql @@ -0,0 +1,85 @@ +-- 215_avito_full_load_window_matches_cadence.sql +-- Окно ретроспективы avito_full_load приводится в согласие с тактом (#2674). +-- +-- КОНТЕКСТ. Каденс источника задаётся ключом default_params."interval_days" +-- (scheduler.compute_next_run_at), глубина обхода — ключом "incremental_days" +-- (run_avito_full_load: since = today - incremental_days, дальше пагинация +-- newest-first с ранней остановкой на первой странице без карточек >= since). +-- Это ДВА независимых литерала, которые обязаны совпадать — и они разъехались: +-- 129_avito_full_load_incremental_split.sql поставил incremental_days=2 при +-- тогдашнем ежедневном такте (2 >= 1 — окно перекрывало такт, дыр не было); +-- 206_scrape_schedules_cut_wasteful_load.sql перевёл источник на interval_days=7, +-- но incremental_days не тронул. +-- Итог на проде (scrape_schedules id=138, проверено read-only 2026-08-06): +-- {"interval_days": 7, "incremental_days": 2, ...} +-- Прогон видит listing_date в [D-2, D] = 3 календарных дня из 7; следующий прогон +-- начинает с [D+5, D+7]. Дни D+1..D+4 не попадают НИ в один прогон — 4 суток из 7 +-- (57%) структурно вне поля зрения источника. +-- +-- ЧИСЛА, обосновывающие новое значение (прод, tradein): +-- * Замер по ОДНОМУ прогону (run 2990, exhaustive 2026-08-02, 2317 строк +-- снапшота / 2184 с датой) — сколько лотов достаёт окно шириной W суток: +-- W=2 -> 147 W=6 -> 446 W=7 -> 1275 W=12 -> 1278 +-- Три вывода. Окно 2 теряет в 8.7 раза (147 против 1275). Шестёрка теряет +-- две трети семёрки (446 против 1275) — это обрыв, а не экономия. И 7..12 — +-- ПЛАТО: +3 лота на пять дополнительных суток окна, то есть семёрка стоит +-- на самой дешёвой точке плато, а не является компромиссом. +-- * Обрыв на 7 и плато за ним — свойство НЕ Avito, а КВАНТОВАНИЯ нашего же +-- парсера относительных дат (_parse_relative_date, providers/avito/serp.py): +-- «неделю назад» -> ровно today-7, «две недели назад» -> ровно today-14. +-- Возрасты 8..13 по этому пути недостижимы — на проде их 3 лота из 2184 +-- (это и есть плато). Значит бакет «возраст 7» — не «поднятые ровно неделю +-- назад», а ВСЁ, чему реально от 7 до 13 суток. W=6 режет не по пику +-- распределения, а по ГРАНИЦЕ КВАНТОВАНИЯ и теряет бакет целиком. +-- * Тот же бакет в общей выборке: (last_seen_at::date - listing_date) по +-- avito за 20 суток — возраст 0-2 = 594, возраст ровно 7 = 1212 из 3714 +-- датированных наблюдений (32.6%). +-- * Полосы соседних прогонов [D-7, D] и [D, D+7] смыкаются с суточным +-- перехлёстом, который покрывает дрейф расписания (замер: last_run +-- 2026-08-03 13:37 -> next_run_at 2026-08-10 14:16 = +7 суток 39 минут). +-- +-- ЦЕНА ПО ЗАПРОСАМ растёт НЕ пропорционально лотам. Стоимость бакета — +-- ceil(свежих / 50) страниц с полом в 1-2 страницы; при окне 7 на бакет +-- приходится ~15-20 свежих (1275 лотов на 77 бакетов = 7 комнатностей x 11 +-- ценовых seed-брекетов) — МЕНЬШЕ одной страницы. Большинство бакетов как +-- стояло на 1-2 страницах, так и останется, глубже пойдут только плотные. +-- Верхняя граница честная и продом уже пережитая: полный обход без отсечки +-- вообще — 6 ч 59 мин (run 295) и 2 ч 34 мин (run 2990); окно 13-15 UTC +-- ограничивает только СТАРТ прогона, не длительность. +-- +-- ВОЗВРАЩАЕТЕ ЕЖЕДНЕВНЫЙ ТАКТ — ВЕРНИТЕ И ОКНО. Ни эта миграция (GREATEST +-- только расширяет), ни планировщик (расширяет до такта, не сужает) окно НЕ +-- сузят. interval_days 7 -> 1 при incremental_days = 7 даст ежедневный прогон +-- с семисуточной глубиной, то есть восьмикратный охват КАЖДЫЙ день. Такт и +-- окно менять одной правкой. +-- +-- Значение НЕ хардкодим числом 7, а выводим из фактического interval_days строки: +-- если такт когда-нибудь поменяют снова, повторный прогон файла (или ручной +-- вызов) снова сведёт окно с тактом. Ту же инвариантность страхует код — +-- scheduler._job_avito_full_load расширяет окно до такта и пишет warning. +-- GREATEST с текущим значением: миграция никогда не СУЖАЕТ окно, если оператор +-- выставил его шире такта осознанно. +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)), +-- 129_avito_full_load_incremental_split.sql (incremental_days), +-- 206_scrape_schedules_cut_wasteful_load.sql (interval_days=7). +-- ТОЛЬКО данные (UPDATE default_params), DDL нет — idempotent by construction. +-- source сравнивается на ТОЧНОЕ равенство: 'avito_full_load_exhaustive' — соседняя +-- джоба, у неё incremental_days нет вовсе (полный обход) и трогать её нельзя. + +BEGIN; + +UPDATE scrape_schedules +SET default_params = default_params + || jsonb_build_object( + 'incremental_days', + GREATEST( + COALESCE(CAST(default_params ->> 'interval_days' AS integer), 1), + COALESCE(CAST(default_params ->> 'incremental_days' AS integer), 1) + ) + ), + updated_at = NOW() +WHERE source = 'avito_full_load' + AND default_params ? 'incremental_days'; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_2674_avito_full_load_coverage.py b/tradein-mvp/backend/tests/test_2674_avito_full_load_coverage.py new file mode 100644 index 00000000..e8f8bcf2 --- /dev/null +++ b/tradein-mvp/backend/tests/test_2674_avito_full_load_coverage.py @@ -0,0 +1,225 @@ +"""#2674, Avito full load: окно ретроспективы под недельный такт + 503 сайдкара. + +Две находки эпика, обе — продолжающаяся потеря данных, а не история. + +1. Окно разошлось с тактом. `interval_days` (каденс, миграция 206) и + `incremental_days` (глубина обхода, миграция 129) — два независимых литерала, + которые обязаны совпадать. На проде было {interval_days: 7, incremental_days: 2}: + прогон видит listing_date в [D-2, D] = 3 суток из 7, дни D+1..D+4 не попадают + ни в один прогон. Чиним в двух местах — данными (миграция 215) и кодом + (scheduler расширяет окно до такта, чтобы расхождение не вернулось). + +2. 503 сайдкара не ретраился ни разу. `_fetch_serp_html_browser` классифицировал + "browser unavailable (proxy may be down)" как soft-ban и отправлял в бюджет + IP-ротации, а ротация снята в #2616 (max_rot=0, `_rotate_ip()` всегда False) → + условие `rot_done < max_rot` ложно всегда. Бюджет коротких backoff-retry стоял + в `else:` и для soft-ban был структурно недостижим. Итог на проде: один блип + сайдкара съедал бакет, четыре подряд — весь прогон + (`_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED = 4`). +""" + +from __future__ import annotations + +import re +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from scraper_kit.avito_exceptions import AvitoRateLimitedError +from scraper_kit.orchestration import scheduler as kit_sched +from scraper_kit.orchestration.scheduler import ( + SchedulerContext, + _job_avito_full_load, + _job_avito_full_load_exhaustive, +) +from scraper_kit.providers.avito import serp as serp_module +from scraper_kit.providers.avito.serp import _AVITO_SIDECAR_TRANSIENT_RETRIES, AvitoScraper + +from app.services.scraper_adapters import RealScraperConfig + +_MIGRATION_215 = ( + Path(__file__).resolve().parents[1] + / "data" + / "sql" + / "215_avito_full_load_window_matches_cadence.sql" +) + + +def _ctx() -> SchedulerContext: + return SchedulerContext( + config=MagicMock(), + matcher=MagicMock(), + enrichment=MagicMock(), + session_factory=MagicMock(), + runs=MagicMock(), + ) + + +async def _incremental_days_passed(params: dict) -> object: + """Прогнать _job_avito_full_load с params и вернуть переданный incremental_days.""" + with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run: + await _job_avito_full_load(MagicMock(), 1, params, _ctx()) + mock_run.assert_awaited_once() + _args, kwargs = mock_run.call_args + return kwargs.get("incremental_days") + + +# ── 1. Окно ретроспективы не уже такта ──────────────────────────────────────── + + +async def test_window_widened_to_cadence() -> None: + """Прод-конфиг на момент находки: такт 7 суток, окно 2 → окно расширяется до 7. + + Фальсификация: без правки в scheduler._job_avito_full_load сюда приезжает 2 — + ровно те «двое суток из семи», о которых говорит эпик. + """ + assert await _incremental_days_passed({"interval_days": 7, "incremental_days": 2}) == 7 + + +async def test_window_wider_than_cadence_left_alone() -> None: + """Окно ШИРЕ такта — осознанный запас оператора, не сужаем.""" + assert await _incremental_days_passed({"interval_days": 3, "incremental_days": 10}) == 10 + + +async def test_daily_cadence_keeps_window() -> None: + """Back-compat: без interval_days такт = 1 сутки, окно 2 уже перекрывает его.""" + assert await _incremental_days_passed({"incremental_days": 2}) == 2 + + +async def test_null_interval_days_does_not_crash() -> None: + """`"interval_days": null` в jsonb приезжает сюда как None, а int(None) — TypeError.""" + assert await _incremental_days_passed({"interval_days": None, "incremental_days": 2}) == 2 + + +async def test_no_window_stays_exhaustive() -> None: + """Строка без incremental_days = полный обход; такт не должен её «инкрементализировать».""" + assert await _incremental_days_passed({"interval_days": 7}) is None + + +async def test_exhaustive_job_ignores_window_params() -> None: + """Соседняя джоба всегда идёт полным обходом, что бы ни лежало в её params.""" + with patch.object(kit_sched, "run_avito_full_load", AsyncMock()) as mock_run: + await _job_avito_full_load_exhaustive( + MagicMock(), 1, {"interval_days": 7, "incremental_days": 2}, _ctx() + ) + _args, kwargs = mock_run.call_args + assert kwargs.get("incremental_days") is None + + +# ── 2. 503 сайдкара получает backoff-retry, а не мгновенный отказ ───────────── + + +def _sidecar_503() -> httpx.HTTPStatusError: + 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) + + +@pytest.mark.asyncio +async def test_sidecar_503_retried_before_giving_up() -> None: + """503 «browser unavailable» — это упавший launch камуфокса, а не бан площадки. + + Сайдкар на таком отказе сам поднимает фоновый retry launch'а, поэтому повтор + через пару секунд обычно проходит. Ждём 1 попытку + весь бюджет backoff-retry. + + Фальсификация: без правки soft-ban уходит в ветку ротации (max_rot=0 → условие + ложно всегда), бюджет backoff недостижим → ровно 1 вызов fetch и немедленный + AvitoRateLimitedError. + """ + scraper = AvitoScraper(RealScraperConfig()) + assert scraper._cffi is None + scraper._browser = AsyncMock() + scraper._browser.fetch = AsyncMock(side_effect=_sidecar_503()) + + with patch.object(serp_module.asyncio, "sleep", AsyncMock()): + with pytest.raises(AvitoRateLimitedError): + await scraper._fetch_serp_html( + "https://www.avito.ru/ekaterinburg/kvartiry/prodam", page=1 + ) + + assert scraper._browser.fetch.await_count == 1 + _AVITO_SIDECAR_TRANSIENT_RETRIES + + +@pytest.mark.asyncio +async def test_sidecar_503_recovers_without_aborting_bucket() -> None: + """Один блип сайдкара больше не стоит бакета: вторая попытка отдаёт HTML.""" + scraper = AvitoScraper(RealScraperConfig()) + scraper._browser = AsyncMock() + scraper._browser.fetch = AsyncMock( + side_effect=[_sidecar_503(), "serp"] + ) + + with patch.object(serp_module.asyncio, "sleep", AsyncMock()): + html = await scraper._fetch_serp_html( + "https://www.avito.ru/ekaterinburg/kvartiry/prodam", page=1 + ) + + assert html == "serp" + assert scraper._browser.fetch.await_count == 2 + + +# ── 3. Миграция 215: статические инварианты ────────────────────────────────── +# Живой БД в тестах нет (см. conftest.py) — эффект UPDATE'а проверяется на проде, +# здесь фиксируем форму файла: транзакционность, отсутствие DDL, точное совпадение +# source (соседняя джоба с похожим именем ловится подстрокой), вывод значения из +# фактического interval_days вместо литерала. + + +def _sql() -> str: + return _MIGRATION_215.read_text(encoding="utf-8") + + +def _executable_sql() -> str: + """SQL без построчных `--`-комментариев — только исполняемый код.""" + lines = [] + for raw in _sql().splitlines(): + code = raw.split("--", 1)[0] + if code.strip(): + lines.append(code) + return "\n".join(lines) + + +def _flat() -> str: + return re.sub(r"\s+", " ", _executable_sql()).strip().lower() + + +def test_migration_215_is_transactional_and_data_only() -> None: + flat = _flat() + assert flat.startswith("begin;") + assert flat.endswith("commit;") + for ddl in ("create ", "alter ", "drop ", "truncate "): + assert ddl not in flat, f"миграция только про данные, найден DDL: {ddl!r}" + + +def test_migration_215_targets_only_avito_full_load() -> None: + """Точное равенство source, НЕ подстрока — 'avito_full_load_exhaustive' рядом.""" + flat = _flat() + assert "where source = 'avito_full_load'" in flat + assert "avito_full_load_exhaustive" not in _executable_sql() + assert " like " not in flat + + +def test_migration_215_derives_window_from_cadence() -> None: + """Окно выводится из interval_days строки, а не хардкодится числом.""" + flat = _flat() + assert "interval_days" in flat + assert "greatest(" in flat + assert re.search(r"'incremental_days',\s*greatest", flat) is not None + + +def test_migration_215_merges_params_not_overwrites() -> None: + """`||` мерджит ключ в default_params — соседние ключи обязаны выжить.""" + flat = _flat() + assert "default_params ||" in flat or "default_params\n||" in _executable_sql().lower() + assert "set default_params = jsonb_build_object" not in flat + + +def test_migration_215_has_no_psycopg_cast_trap() -> None: + """Repo-конвенция: CAST(x AS type), не x::type.""" + assert "::" not in _executable_sql() + assert "cast(" in _flat() diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py index ccef889a..e012353d 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/scheduler.py @@ -519,6 +519,28 @@ async def _job_avito_full_load( ) -> None: _incremental_days = params.get("incremental_days") incremental_days = int(_incremental_days) if _incremental_days is not None else None + # Окно ретроспективы не может быть уже такта (#2674). incremental_days задаёт + # since=today-N, а прогон повторяется раз в interval_days суток → при N < interval_days + # объявления, поднятые в дни (D, D+interval_days-N), не попадают НИ в один прогон. + # Ровно это и случилось: миграция 206 перевела источник на interval_days=7, оставив + # incremental_days=2 из миграции 129 → 3 календарных дня из 7 в поле зрения. + # Два независимых литерала, которые обязаны совпадать, однажды уже разъехались — + # поэтому расхождение чиним здесь, а не только данными. + # None-safe так же, как incremental_days выше: `"interval_days": null` в jsonb + # приезжает сюда как None, а int(None) — TypeError. + _interval_days = params.get("interval_days") + interval_days = max(1, int(_interval_days)) if _interval_days is not None else 1 + if incremental_days is not None and incremental_days < interval_days: + logger.warning( + "avito_full_load: окно ретроспективы incremental_days=%d уже такта " + "interval_days=%d — расширяю до %d, иначе %d суток объявлений не видит " + "ни один прогон", + incremental_days, + interval_days, + interval_days, + interval_days - incremental_days, + ) + incremental_days = interval_days await run_avito_full_load( db, run_id=run_id, 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 c52c2d8d..2b05b055 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 @@ -505,9 +505,9 @@ class AvitoScraper(BaseScraper): # условным на avito_proxy_rotate_url, который больше не существует). max_rot = 0 rot_done = 0 - # Независимый бюджет коротких retry на transient sidecar-ошибки (timeout/ - # crash) — НЕ тратит бюджет ротации. Каждая ветка ниже (ротация / transient) - # уменьшает свой счётчик → цикл всегда движется к условию исчерпания. + # Бюджет коротких backoff-retry на ЛЮБУЮ sidecar-ошибку (timeout/crash И + # soft-ban/503) — НЕ тратит бюджет ротации. Каждая ветка ниже (ротация / + # backoff) уменьшает свой счётчик → цикл всегда движется к исчерпанию. transient_left = _AVITO_SIDECAR_TRANSIENT_RETRIES while True: try: @@ -541,36 +541,40 @@ class AvitoScraper(BaseScraper): or "browser unavailable" in lowered or "proxy may be down" in lowered ) - if is_soft_ban: + if is_soft_ban and rot_done < max_rot and await self._rotate_ip(): # Soft proxy rate-limit / прокси недоступен → свежий exit-IP. - if rot_done < max_rot and await self._rotate_ip(): - rot_done += 1 - logger.info( - "avito page=%d sidecar soft-ban (status=%d) — " - "retry after rotation #%d: %s", - page, - status, - rot_done, - error_text[:200], - ) - continue - else: - # Transient (Page.goto Timeout / Target closed / прочие 5xx) → - # короткий backoff-retry того же url БЕЗ ротации. - if transient_left > 0: - transient_left -= 1 - backoff = _AVITO_SIDECAR_TRANSIENT_BACKOFF_SEC + random.uniform(0, 0.5) - logger.warning( - "avito page=%d sidecar transient (status=%d) — " - "retry in %.1fs (left=%d): %s", - page, - status, - backoff, - transient_left, - error_text[:200], - ) - await asyncio.sleep(backoff) - continue + rot_done += 1 + logger.info( + "avito page=%d sidecar soft-ban (status=%d) — retry after rotation #%d: %s", + page, + status, + rot_done, + error_text[:200], + ) + continue + # Backoff-retry того же url БЕЗ ротации. Раньше эта ветка стояла в + # `else:` к soft-ban и была для него НЕДОСТИЖИМА, а бюджет ротации с + # #2616 шага 2 равен нулю (max_rot=0, _rotate_ip() всегда False) — + # то есть 503 «browser unavailable (proxy may be down)» от сайдкара + # не ретраился ВООБЩЕ, ни разу (#2674). Сайдкар на таком 503 сам + # поднимает фоновый retry launch'а браузера, так что повтор через + # пару секунд обычно проходит; вместо этого один блип сайдкара + # съедал бакет, четыре подряд (_AVITO_SWEEP_MAX_CONSECUTIVE_BLOCKED) + # — весь прогон. Теперь бюджет transient общий для обеих причин. + if transient_left > 0: + transient_left -= 1 + backoff = _AVITO_SIDECAR_TRANSIENT_BACKOFF_SEC + random.uniform(0, 0.5) + logger.warning( + "avito page=%d sidecar %s (status=%d) — retry in %.1fs (left=%d): %s", + page, + "soft-ban" if is_soft_ban else "transient", + status, + backoff, + transient_left, + error_text[:200], + ) + await asyncio.sleep(backoff) + continue # Оба бюджета (ротация + transient) исчерпаны → graceful soft-ban, # чтобы run_avito_full_load/city_sweep ушли в mark_banned (partial # preserved), а НЕ в generic except → mark_failed. Намеренно