feat(avito): incremental daily + exhaustive weekly schedule split (+ scheduler interval_days) #1770

Merged
lekss361 merged 2 commits from feat/avito-incremental-schedule-split into main 2026-06-18 20:25:39 +00:00
5 changed files with 412 additions and 17 deletions

View file

@ -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
(например 223 окно 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":

View file

@ -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(

View file

@ -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;

View file

@ -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

View file

@ -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