From bf6368383391dbd061a92f4a96b6886b3d085945 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 23:04:35 +0300 Subject: [PATCH 1/2] 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 From 1f7018d41548af1844de79c5ab089e9899f0a4af Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 23:10:18 +0300 Subject: [PATCH 2/2] feat(scheduler): per-source interval_days cadence (weekly exhaustive avito) scrape_schedules had no interval column -> every source ran daily. Add interval_days from default_params (default 1, backward-compatible) so compute_next_run_at can schedule N days ahead. Set avito_full_load_exhaustive to interval_days=7 (weekly full pass for last_seen/delisting + silent price edits); avito_full_load stays daily incremental. --- tradein-mvp/backend/app/services/scheduler.py | 44 +++++++++++++------ .../129_avito_full_load_incremental_split.sql | 34 +++++++++----- tradein-mvp/backend/tests/test_scheduler.py | 36 +++++++++++++++ 3 files changed, 91 insertions(+), 23 deletions(-) diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index d8715b02..480d4caf 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -4,7 +4,8 @@ Логика: 1. Каждые 60 sec query scrape_schedules WHERE enabled=true AND next_run_at <= NOW(). 2. Для каждого: проверить нет ли уже running run этого source — если нет, trigger. -3. После trigger — recompute next_run_at в следующих суток. +3. После trigger — recompute next_run_at через interval_days суток (из default_params, + default 1 = daily; например avito_full_load_exhaustive = 7 → weekly). 4. На old runs (orphaned status='running' > 6h без heartbeat) — пометить как 'zombie'. Sources: @@ -18,8 +19,9 @@ Sources: ПО УМОЛЧАНИЮ инкрементальный: 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 + ловит тихие правки цены) + exhaustive обход; weekly cadence через interval_days=7 в + default_params — освежает 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) @@ -131,22 +133,32 @@ ZOMBIE_THRESHOLD_HOURS = 6 def compute_next_run_at( - window_start_hour: int, window_end_hour: int, *, now: datetime | None = None + window_start_hour: int, + window_end_hour: int, + *, + now: datetime | None = None, + interval_days: int = 1, ) -> datetime: - """Pick random datetime в window [start, end) UTC, для СЛЕДУЮЩИХ суток после now. + """Pick random datetime в window [start, end) UTC, через interval_days суток после now. + + interval_days задаёт каденс источника: 1 (default) = daily (back-compat), 7 = weekly. + Берётся из schedule.default_params["interval_days"] вызывающим кодом (_claim_run / + _defer_next_run_at); отсутствие ключа → 1 → прежнее ежедневное поведение. Если window_end_hour <= window_start_hour → cross-midnight window (например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59). """ now = now or datetime.now(tz=UTC) - tomorrow = (now + timedelta(days=1)).date() + interval_days = max(1, int(interval_days)) + # Целевая дата = now + interval_days суток (interval_days=1 → завтра, как раньше). + target = (now + timedelta(days=interval_days)).date() if window_end_hour > window_start_hour: # Обычное окно (например 2..5 → 02:00-04:59) start_seconds = window_start_hour * 3600 end_seconds = window_end_hour * 3600 rand_seconds = random.randint(start_seconds, end_seconds - 1) - return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds) + return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds) else: # Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59) # Длина окна = (24-start) + end часов @@ -155,15 +167,17 @@ def compute_next_run_at( # Если rand попадает в первую часть (start..24) first_half = (24 - window_start_hour) * 3600 if rand_seconds < first_half: - # Текущая дата (если ещё не наступило окно) или next day - base_date = now.date() if now.hour < window_start_hour else tomorrow + # interval_days=1: текущая дата (если окно ещё не наступило сегодня) или next day. + # interval_days>1: всегда целевая дата (стаггер на N суток вперёд). + today_ok = interval_days == 1 and now.hour < window_start_hour + base_date = now.date() if today_ok else target return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta( seconds=window_start_hour * 3600 + rand_seconds ) else: - # Во второй части (0..end), следующего дня + # Во второй части (0..end), целевого дня offset = rand_seconds - first_half - return datetime.combine(tomorrow, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset) + return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset) def has_running_run(db: Session, source: str) -> bool: @@ -243,6 +257,7 @@ def _claim_run(db: Session, schedule_row: dict[str, Any]) -> int | None: next_at = compute_next_run_at( schedule_row["window_start_hour"], schedule_row["window_end_hour"], + interval_days=int(params.get("interval_days", 1)), ) db.execute( text( @@ -274,9 +289,11 @@ def _defer_next_run_at(db: Session, schedule_row: dict[str, Any]) -> None: раз в минуту круглосуточно. Сдвиг откладывает следующую попытку до завтрашнего окна. """ source = schedule_row["source"] + params = schedule_row.get("default_params") or {} next_at = compute_next_run_at( schedule_row["window_start_hour"], schedule_row["window_end_hour"], + interval_days=int(params.get("interval_days", 1)), ) db.execute( text( @@ -594,8 +611,9 @@ async def trigger_avito_full_load_exhaustive_run( Зеркало trigger_avito_full_load_run, но incremental_days форсится в None → полный exhaustive обход (root-бисекция всего ценового диапазона на комнатность). - Запускается реже (weekly schedule) чтобы освежить last_seen (10-дневный - delisting TTL) и поймать тихие правки цены, которые shallow-инкремент пропускает. + Запускается реже (weekly через interval_days=7 в default_params → compute_next_run_at + откладывает next_run_at на 7 суток) чтобы освежить last_seen (10-дневный delisting TTL) + и поймать тихие правки цены, которые shallow-инкремент пропускает. Все прочие params (concurrency, price_cap_per_bucket, ...) читаются из default_params как у avito_full_load. Returns run_id (или None если skip). """ 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 index 30148c12..268b86c4 100644 --- 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 @@ -10,13 +10,14 @@ -- ЗАВИСИМОСТИ: 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. +-- CADENCE: в scrape_schedules нет колонки interval_minutes/cron — каденс задаётся +-- ключом "interval_days" в default_params (jsonb). scheduler.compute_next_run_at читает +-- его через _claim_run/_defer_next_run_at и откладывает next_run_at на N суток вперёд +-- (отсутствие ключа → 1 = daily, back-compat). avito_full_load_exhaustive получает +-- interval_days=7 → ИСТИННАЯ weekly-периодичность (не эмуляция через single-shot +-- next_run_at). next_run_at тут задаёт лишь ПЕРВЫЙ запуск (стаггер +3 дня от daily- +-- инкремента), дальше строка сама рекуррентит каждые 7 суток. avito_full_load остаётся +-- без interval_days → daily. Колонки строго по схеме 052/127. BEGIN; @@ -30,10 +31,13 @@ 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) ───── +-- ── 2. avito_full_load_exhaustive → полный обход, weekly (interval_days=7) ──── -- default_params БЕЗ incremental_days → run_avito_full_load(incremental_days=None) -- → exhaustive bisection. concurrency=1 (агрессивный анти-бан для глубокой пагинации). --- Окно 13:00-15:00 UTC то же что у daily (одна и та же apw-прокси-секция), но +-- interval_days=7 → compute_next_run_at откладывает next_run_at на 7 суток после каждого +-- fire = ИСТИННАЯ weekly-периодичность (освежает last_seen 10-дн delisting TTL + ловит +-- тихие правки цены без daily deep-pagination нагрузки). +-- Окно 13:00-15:00 UTC то же что у daily (одна и та же apw-прокси-секция), а первый -- next_run_at стартует через 3 дня чтобы не пересечься с инкрементальным прогоном. -- Идемпотентно: ON CONFLICT (source) DO NOTHING — повторный прогон не трогает -- операторские правки enabled/next_run_at/window. @@ -53,10 +57,20 @@ VALUES 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}' + '{"price_cap_per_bucket": 1400, "concurrency": 1, "request_delay_sec": 7.0, "secondary_only": true, "interval_days": 7}' AS jsonb ) ) ON CONFLICT (source) DO NOTHING; +-- Idempotent backfill для interval_days на случай если строка exhaustive уже была +-- засеяна предыдущей версией 129 (без interval_days) до этого деплоя — ON CONFLICT +-- DO NOTHING выше её бы не тронул. Мерджим только ключ interval_days, прочие +-- операторские правки default_params сохраняются (|| перезаписывает лишь этот ключ). +UPDATE scrape_schedules +SET default_params = default_params || CAST('{"interval_days": 7}' AS jsonb), + updated_at = NOW() +WHERE source = 'avito_full_load_exhaustive' + AND NOT (default_params ? 'interval_days'); + COMMIT; diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py index 2e515688..5e70b4e6 100644 --- a/tradein-mvp/backend/tests/test_scheduler.py +++ b/tradein-mvp/backend/tests/test_scheduler.py @@ -54,6 +54,42 @@ def test_compute_next_run_timezone_aware() -> None: assert next_at.tzinfo is not None +# ── per-source interval_days cadence (weekly exhaustive avito) ────────────── + + +def test_compute_next_run_default_interval_is_daily() -> None: + """Без interval_days → +1 сутки (back-compat, как раньше).""" + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) + next_at = compute_next_run_at(13, 15, now=now) + assert next_at.date() == datetime(2026, 5, 24).date() + assert 13 <= next_at.hour <= 14 # window [13, 15) UTC + + +def test_compute_next_run_interval_days_1_explicit() -> None: + """interval_days=1 явно → +1 сутки (идентично default).""" + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) + next_at = compute_next_run_at(13, 15, now=now, interval_days=1) + assert next_at.date() == datetime(2026, 5, 24).date() + assert 13 <= next_at.hour <= 14 + + +def test_compute_next_run_interval_days_7_weekly() -> None: + """interval_days=7 → ~7 суток вперёд, всё ещё в окне.""" + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) + next_at = compute_next_run_at(13, 15, now=now, interval_days=7) + assert next_at.date() == datetime(2026, 5, 30).date() # 23 + 7 + assert 13 <= next_at.hour <= 14 # всё ещё внутри окна [13, 15) UTC + assert next_at > now + + +def test_compute_next_run_interval_days_cross_midnight_weekly() -> None: + """interval_days с cross-midnight окном → целевой день +N, час в окне.""" + now = datetime(2026, 5, 23, 12, 0, tzinfo=UTC) + next_at = compute_next_run_at(22, 3, now=now, interval_days=7) + assert next_at > now + assert next_at.hour >= 22 or next_at.hour < 3 + + # ── #750: атомарный _claim_run (advisory lock — anti-double-sweep) ────────── from collections import deque # noqa: E402