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