feat(avito): incremental daily + exhaustive weekly schedule split (+ scheduler interval_days) #1770
5 changed files with 412 additions and 17 deletions
|
|
@ -4,17 +4,24 @@
|
|||
Логика:
|
||||
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:
|
||||
- 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 через 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)
|
||||
|
|
@ -126,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 часов
|
||||
|
|
@ -150,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:
|
||||
|
|
@ -238,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(
|
||||
|
|
@ -269,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(
|
||||
|
|
@ -539,10 +561,61 @@ 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 через 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).
|
||||
"""
|
||||
run_id = _claim_run(db, schedule_row)
|
||||
if run_id is None:
|
||||
|
|
@ -560,15 +633,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 +1609,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":
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -0,0 +1,76 @@
|
|||
-- 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 — каденс задаётся
|
||||
-- ключом "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;
|
||||
|
||||
-- ── 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 → полный обход, weekly (interval_days=7) ────
|
||||
-- default_params БЕЗ incremental_days → run_avito_full_load(incremental_days=None)
|
||||
-- → exhaustive bisection. concurrency=1 (агрессивный анти-бан для глубокой пагинации).
|
||||
-- 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.
|
||||
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, "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;
|
||||
|
|
@ -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
|
||||
|
|
@ -554,3 +590,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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue