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

@@ -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).
     """

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;

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