From bf6368383391dbd061a92f4a96b6886b3d085945 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 23:04:35 +0300 Subject: [PATCH] feat(avito): incremental daily + exhaustive weekly schedule split run_avito_full_load gains incremental_days param -> passes since= to the (merged) incremental SERP engine. avito_full_load schedule flipped to incremental_days=2 (shallow, date early-stop -> avoids deep-pagination 429 bans). New avito_full_load_exhaustive source runs the full walk weekly to refresh last_seen (10-day delisting TTL) and catch silent price edits. --- tradein-mvp/backend/app/services/scheduler.py | 74 ++++++++- .../backend/app/services/scrape_pipeline.py | 21 ++- .../129_avito_full_load_incremental_split.sql | 62 ++++++++ tradein-mvp/backend/tests/test_scheduler.py | 140 ++++++++++++++++++ .../backend/tests/test_scrape_pipeline.py | 46 ++++++ 5 files changed, 335 insertions(+), 8 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/129_avito_full_load_incremental_split.sql diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 33bef3fb..d8715b02 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -9,12 +9,17 @@ Sources: - avito_city_sweep → run_avito_city_sweep (scrape_pipeline.py) - - avito_full_load → run_avito_full_load (exhaustive региональный сбор Avito ЕКБ + - avito_full_load → run_avito_full_load (региональный сбор Avito ЕКБ ВТОРИЧКИ без anchor'ов — room×price адаптивное партиционирование через pmin/pmax, secondary_only, incremental on_bucket save; обходит SERP-cap ~5000/запрос. Окно 13:00-15:00 UTC чтобы не конфликтовать на shared apw-прокси с avito_city_sweep (6-7), - avito_newbuilding (2-5), avito_detail_backfill (9-12)) + avito_newbuilding (2-5), avito_detail_backfill (9-12). + ПО УМОЛЧАНИЮ инкрементальный: incremental_days из default_params + → shallow обход с date early-stop, избегает page-36 429-банов) + - avito_full_load_exhaustive → run_avito_full_load с incremental_days=None (полный + exhaustive обход; weekly cadence — освежает last_seen + 10-дневный delisting TTL + ловит тихие правки цены) - avito_newbuilding_sweep → run_avito_newbuilding_sweep (scrape_pipeline.py; dedicated novostroyka-filtered citywide SERP → save, 100% new-build cards) - yandex_city_sweep → run_yandex_city_sweep (scrape_pipeline.py, #561; shipped DORMANT) @@ -539,10 +544,60 @@ async def trigger_cian_full_load_run(db: Session, schedule_row: dict[str, Any]) async def trigger_avito_full_load_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать scrape_runs + launch run_avito_full_load в asyncio.create_task. - Exhaustive региональный сбор Avito ЕКБ ВТОРИЧКИ (room×price партиционирование - через pmin/pmax, secondary_only, incremental on_bucket save) — обходит SERP-cap + Региональный сбор Avito ЕКБ ВТОРИЧКИ (room×price партиционирование через + pmin/pmax, secondary_only, incremental on_bucket save) — обходит SERP-cap ~5000 на запрос. Зеркало trigger_cian_full_load_run. apw-прокси выбирает сам - scraper (AvitoScraper). Returns run_id (или None если skip — running run). + scraper (AvitoScraper). + + Режим по умолчанию — инкрементальный: incremental_days читается из + default_params (shallow обход с date early-stop → избегает page-36 429-банов). + Для полного exhaustive обхода используется отдельный source + avito_full_load_exhaustive (incremental_days форсится в None). + + Returns run_id (или None если skip — running run). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + params = schedule_row.get("default_params") or {} + _incremental_days = params.get("incremental_days") + incremental_days = int(_incremental_days) if _incremental_days is not None else None + + async def _run() -> None: + run_db = SessionLocal() + try: + await run_avito_full_load( + run_db, + run_id=run_id, + price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)), + concurrency=int(params.get("concurrency", 5)), + request_delay_sec=float(params.get("request_delay_sec", 7.0)), + secondary_only=bool(params.get("secondary_only", True)), + resume_run_id=None, + incremental_days=incremental_days, + ) + except Exception: + logger.exception("scheduler: run_avito_full_load crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered avito_full_load run_id=%d", run_id) + return run_id + + +async def trigger_avito_full_load_exhaustive_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать scrape_runs + launch run_avito_full_load (EXHAUSTIVE) в asyncio.create_task. + + Зеркало trigger_avito_full_load_run, но incremental_days форсится в None → + полный exhaustive обход (root-бисекция всего ценового диапазона на комнатность). + Запускается реже (weekly schedule) чтобы освежить last_seen (10-дневный + delisting TTL) и поймать тихие правки цены, которые shallow-инкремент пропускает. + Все прочие params (concurrency, price_cap_per_bucket, ...) читаются из + default_params как у avito_full_load. Returns run_id (или None если skip). """ run_id = _claim_run(db, schedule_row) if run_id is None: @@ -560,15 +615,18 @@ async def trigger_avito_full_load_run(db: Session, schedule_row: dict[str, Any]) request_delay_sec=float(params.get("request_delay_sec", 7.0)), secondary_only=bool(params.get("secondary_only", True)), resume_run_id=None, + incremental_days=None, # force exhaustive full walk ) except Exception: - logger.exception("scheduler: run_avito_full_load crashed run_id=%d", run_id) + logger.exception( + "scheduler: run_avito_full_load (exhaustive) crashed run_id=%d", run_id + ) finally: run_db.close() task = asyncio.create_task(_run()) task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) - logger.info("scheduler: triggered avito_full_load run_id=%d", run_id) + logger.info("scheduler: triggered avito_full_load_exhaustive run_id=%d", run_id) return run_id @@ -1533,6 +1591,8 @@ async def scheduler_loop() -> None: await trigger_avito_city_sweep_run(db, sch) elif source == "avito_full_load": await trigger_avito_full_load_run(db, sch) + elif source == "avito_full_load_exhaustive": + await trigger_avito_full_load_exhaustive_run(db, sch) elif source == "avito_newbuilding_sweep": await trigger_avito_newbuilding_sweep_run(db, sch) elif source == "yandex_city_sweep": diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index f24f29d1..cd25c417 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -32,6 +32,7 @@ import logging import random from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields +from datetime import date, timedelta from urllib.parse import urlparse from curl_cffi.requests import AsyncSession @@ -2370,8 +2371,9 @@ async def run_avito_full_load( concurrency: int = 5, secondary_only: bool = True, resume_run_id: int | None = None, + incremental_days: int | None = None, ) -> AvitoFullLoadCounters: - """Exhaustive региональный сбор Avito ЕКБ вторички (БЕЗ anchor'ов). + """Exhaustive ИЛИ инкрементальный региональный сбор Avito ЕКБ вторички (БЕЗ anchor'ов). Avito anchor/citywide упирается в SERP-cap ~5000 результатов на запрос → вторичка катастрофически недобрана. Решение зеркалит cian/yandex full_load: @@ -2386,6 +2388,11 @@ async def run_avito_full_load( resume_run_id: если задан — читает done_buckets из counters прошлого run и пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного) avito_full_load прогона для resume. Без поля — full walk с нуля. + incremental_days: если задан — shallow инкрементальный обход. Считается + since = date.today() - timedelta(days=incremental_days) и передаётся в + fetch_all_secondary(since=since): date early-stop останавливает пагинацию + до глубоких страниц (избегает page-36 429-банов). None → since=None → + exhaustive bisection (полный обход, текущее поведение). Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора. Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket. @@ -2394,6 +2401,17 @@ async def run_avito_full_load( """ counters = AvitoFullLoadCounters() + # ── Инкрементальный режим: since-cutoff для shallow обхода ──────────────── + since: date | None = None + if incremental_days is not None: + since = date.today() - timedelta(days=incremental_days) + logger.info( + "avito-full-load run_id=%d: INCREMENTAL mode — since=%s (last %d days)", + run_id, + since.isoformat(), + incremental_days, + ) + # ── Checkpoint/resume: читаем done_buckets из прошлого run ─────────────── skip_set: set[str] = set() if resume_run_id is not None: @@ -2459,6 +2477,7 @@ async def run_avito_full_load( on_bucket=_on_bucket, on_progress=_on_progress, skip_buckets=skip_set if skip_set else None, + since=since, ) logger.info( diff --git a/tradein-mvp/backend/data/sql/129_avito_full_load_incremental_split.sql b/tradein-mvp/backend/data/sql/129_avito_full_load_incremental_split.sql new file mode 100644 index 00000000..30148c12 --- /dev/null +++ b/tradein-mvp/backend/data/sql/129_avito_full_load_incremental_split.sql @@ -0,0 +1,62 @@ +-- 129_avito_full_load_incremental_split.sql +-- Сплит Avito full_load на инкрементальный (daily) + exhaustive (реже). +-- +-- Контекст: run_avito_full_load теперь принимает incremental_days. Если задан — +-- fetch_all_secondary(since=date.today()-incremental_days) делает shallow обход с +-- date early-stop → НЕ доходит до глубоких страниц (page-36 429-баны). exhaustive +-- (incremental_days=NULL) дорого и рискованно для бана, но нужен периодически чтобы +-- освежить last_seen (10-дневный delisting TTL) и поймать тихие правки цены. +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)), +-- 127_scrape_schedules_seed_avito_full_load.sql (seed avito_full_load). +-- +-- ВАЖНО по cadence: scrape_schedules НЕ имеет interval_minutes/cron — recurrence +-- вычисляется в коде (scheduler.compute_next_run_at), который ВСЕГДА пикает +-- next_run_at на СЛЕДУЮЩИЕ сутки (daily cadence) после каждого fire. Колонок для +-- weekly-интервала в схеме нет (см. 052). Поэтому "weekly" здесь = строка enabled +-- со next_run_at, стартующей через несколько дней (стаггер от daily); ИСТИННАЯ +-- weekly-периодичность потребует расширения compute_next_run_at и здесь НЕ +-- эмулируется через несуществующую колонку. Колонки строго по схеме 052/127. + +BEGIN; + +-- ── 1. avito_full_load → инкрементальный (shallow, daily) ──────────────────── +-- Мерджим incremental_days=2 в существующий default_params, сохраняя прочие ключи +-- (price_cap_per_bucket, concurrency, request_delay_sec, secondary_only). +-- Идемпотентно: повторный прогон просто перезапишет incremental_days в 2. +-- || перезаписывает только пересекающийся ключ на верхнем уровне jsonb. +UPDATE scrape_schedules +SET default_params = default_params || CAST('{"incremental_days": 2}' AS jsonb), + updated_at = NOW() +WHERE source = 'avito_full_load'; + +-- ── 2. avito_full_load_exhaustive → полный обход, реже (стаггер от daily) ───── +-- default_params БЕЗ incremental_days → run_avito_full_load(incremental_days=None) +-- → exhaustive bisection. concurrency=1 (агрессивный анти-бан для глубокой пагинации). +-- Окно 13:00-15:00 UTC то же что у daily (одна и та же apw-прокси-секция), но +-- next_run_at стартует через 3 дня чтобы не пересечься с инкрементальным прогоном. +-- Идемпотентно: ON CONFLICT (source) DO NOTHING — повторный прогон не трогает +-- операторские правки enabled/next_run_at/window. +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'avito_full_load_exhaustive', + true, + 13, + 15, + ((CURRENT_DATE + INTERVAL '3 days') + make_interval(hours => 13)) AT TIME ZONE 'UTC', + CAST( + '{"price_cap_per_bucket": 1400, "concurrency": 1, "request_delay_sec": 7.0, "secondary_only": true}' + AS jsonb + ) +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py index ae390a0a..2e515688 100644 --- a/tradein-mvp/backend/tests/test_scheduler.py +++ b/tradein-mvp/backend/tests/test_scheduler.py @@ -554,3 +554,143 @@ async def test_scheduler_dispatch_routes_avito_full_load(monkeypatch: pytest.Mon await _sched.trigger_avito_full_load_run(db, sch) assert "avito_full_load" in triggered + + +# ── avito_full_load incremental_days + avito_full_load_exhaustive split ─────── + + +@pytest.mark.asyncio +async def test_trigger_avito_full_load_reads_incremental_days( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """incremental_days из default_params прокидывается в run_avito_full_load.""" + db = _FakeSchedulerDB() + calls: dict[str, list] = {"load": []} + + def fake_create_run(_d, *, source, params): + db.running[source] = True + return 410 + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + async def fake_full_load(run_db, *, run_id, **kwargs): + calls["load"].append(kwargs) + + monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load) + monkeypatch.setattr(_sched, "SessionLocal", lambda: db) + + row = { + **_AVITO_FULL_LOAD_ROW, + "default_params": {**_AVITO_FULL_LOAD_ROW["default_params"], "incremental_days": 2}, + } + await _sched.trigger_avito_full_load_run(db, row) + await asyncio.sleep(0) + + assert len(calls["load"]) == 1 + assert calls["load"][0]["incremental_days"] == 2 + + +@pytest.mark.asyncio +async def test_trigger_avito_full_load_no_incremental_days_is_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Без incremental_days в params → run_avito_full_load(incremental_days=None) (exhaustive).""" + db = _FakeSchedulerDB() + calls: dict[str, list] = {"load": []} + + def fake_create_run(_d, *, source, params): + db.running[source] = True + return 411 + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + async def fake_full_load(run_db, *, run_id, **kwargs): + calls["load"].append(kwargs) + + monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load) + monkeypatch.setattr(_sched, "SessionLocal", lambda: db) + + await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW) + await asyncio.sleep(0) + + assert len(calls["load"]) == 1 + assert calls["load"][0]["incremental_days"] is None + + +_AVITO_FULL_LOAD_EXHAUSTIVE_ROW = { + "source": "avito_full_load_exhaustive", + "window_start_hour": 13, + "window_end_hour": 15, + "default_params": { + "price_cap_per_bucket": 1400, + "concurrency": 1, + "request_delay_sec": 7.0, + "secondary_only": True, + }, +} + + +@pytest.mark.asyncio +async def test_trigger_avito_full_load_exhaustive_forces_none( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """avito_full_load_exhaustive форсит incremental_days=None даже если в params задан.""" + db = _FakeSchedulerDB() + calls: dict[str, list] = {"load": []} + + def fake_create_run(_d, *, source, params): + db.running[source] = True + return 420 + + monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run) + + async def fake_full_load(run_db, *, run_id, **kwargs): + calls["load"].append((run_id, kwargs)) + + monkeypatch.setattr(_sched, "run_avito_full_load", fake_full_load) + monkeypatch.setattr(_sched, "SessionLocal", lambda: db) + + # Даже если оператор случайно положит incremental_days в params — exhaustive игнорит. + row = { + **_AVITO_FULL_LOAD_EXHAUSTIVE_ROW, + "default_params": { + **_AVITO_FULL_LOAD_EXHAUSTIVE_ROW["default_params"], + "incremental_days": 5, + }, + } + run_id = await _sched.trigger_avito_full_load_exhaustive_run(db, row) + await asyncio.sleep(0) + + assert run_id == 420 + assert len(calls["load"]) == 1 + loaded_run_id, kwargs = calls["load"][0] + assert loaded_run_id == 420 + assert kwargs["incremental_days"] is None + assert kwargs["concurrency"] == 1 + assert kwargs["price_cap_per_bucket"] == 1400 + assert kwargs["resume_run_id"] is None + + +@pytest.mark.asyncio +async def test_scheduler_dispatch_routes_avito_full_load_exhaustive( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Dispatch роутит avito_full_load_exhaustive → trigger_avito_full_load_exhaustive_run.""" + triggered: list[str] = [] + + async def fake_trigger(db, sch): + triggered.append(sch["source"]) + return 1 + + monkeypatch.setattr(_sched, "trigger_avito_full_load_exhaustive_run", fake_trigger) + + db = object() + sch = _AVITO_FULL_LOAD_EXHAUSTIVE_ROW + source = sch["source"] + # Replicate the elif chain from scheduler_loop + if source == "avito_full_load": + pass + elif source == "avito_full_load_exhaustive": + await _sched.trigger_avito_full_load_exhaustive_run(db, sch) + + assert "avito_full_load_exhaustive" in triggered diff --git a/tradein-mvp/backend/tests/test_scrape_pipeline.py b/tradein-mvp/backend/tests/test_scrape_pipeline.py index cff44858..84faf283 100644 --- a/tradein-mvp/backend/tests/test_scrape_pipeline.py +++ b/tradein-mvp/backend/tests/test_scrape_pipeline.py @@ -1,5 +1,6 @@ """Offline smoke for scrape_pipeline. Mocks scrapers + DB session.""" +from datetime import date, timedelta from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -7,6 +8,7 @@ import pytest from app.services.scrape_pipeline import ( PipelineCounters, PipelineResult, + run_avito_full_load, run_avito_pipeline, ) from app.services.scrapers.base import ScrapedLot @@ -84,3 +86,47 @@ async def test_pipeline_group_by_house_unique_set() -> None: assert result.counters.lots_fetched == 4 assert result.counters.lots_inserted == 4 assert result.counters.unique_houses == 2 + + +# ── run_avito_full_load: incremental_days → since plumbing (#avito-split) ───── + + +def _full_load_scraper() -> MagicMock: + """AvitoScraper mock с async-CM + AsyncMock fetch_all_secondary.""" + scraper = MagicMock() + scraper.__aenter__ = AsyncMock(return_value=scraper) + scraper.__aexit__ = AsyncMock(return_value=None) + scraper.fetch_all_secondary = AsyncMock(return_value=[]) + return scraper + + +@pytest.mark.asyncio +async def test_full_load_incremental_days_passes_since() -> None: + """incremental_days=2 → fetch_all_secondary(since=date.today()-2d).""" + mock_db = MagicMock() + scraper = _full_load_scraper() + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper), + patch("app.services.scrape_pipeline.scrape_runs", MagicMock()), + ): + await run_avito_full_load(mock_db, run_id=1, incremental_days=2) + + kwargs = scraper.fetch_all_secondary.call_args.kwargs + assert kwargs["since"] == date.today() - timedelta(days=2) + + +@pytest.mark.asyncio +async def test_full_load_no_incremental_days_since_none() -> None: + """incremental_days=None (default) → fetch_all_secondary(since=None) — exhaustive.""" + mock_db = MagicMock() + scraper = _full_load_scraper() + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=scraper), + patch("app.services.scrape_pipeline.scrape_runs", MagicMock()), + ): + await run_avito_full_load(mock_db, run_id=2) + + kwargs = scraper.fetch_all_secondary.call_args.kwargs + assert kwargs["since"] is None