diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py
index 92f0abed..33bef3fb 100644
--- a/tradein-mvp/backend/app/services/scheduler.py
+++ b/tradein-mvp/backend/app/services/scheduler.py
@@ -9,6 +9,12 @@
Sources:
- avito_city_sweep → run_avito_city_sweep (scrape_pipeline.py)
+ - avito_full_load → run_avito_full_load (exhaustive региональный сбор 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_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)
@@ -104,6 +110,7 @@ from app.core.db import SessionLocal
from app.services import scrape_runs as runs_mod
from app.services.scrape_pipeline import (
run_avito_city_sweep,
+ run_avito_full_load,
run_avito_newbuilding_sweep,
run_cian_city_sweep,
run_cian_full_load,
@@ -529,6 +536,42 @@ async def trigger_cian_full_load_run(db: Session, schedule_row: dict[str, Any])
return run_id
+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
+ ~5000 на запрос. Зеркало trigger_cian_full_load_run. apw-прокси выбирает сам
+ scraper (AvitoScraper). 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 {}
+
+ 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,
+ )
+ 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_domclick_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
"""Создать scrape_runs + launch run_domclick_city_sweep в asyncio.create_task.
@@ -1488,6 +1531,8 @@ async def scheduler_loop() -> None:
source = sch["source"]
if source == "avito_city_sweep":
await trigger_avito_city_sweep_run(db, sch)
+ elif source == "avito_full_load":
+ await trigger_avito_full_load_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 d9fd695f..18070bb7 100644
--- a/tradein-mvp/backend/app/services/scrape_pipeline.py
+++ b/tradein-mvp/backend/app/services/scrape_pipeline.py
@@ -2310,6 +2310,178 @@ async def run_yandex_full_load(
raise
+# ── Avito exhaustive full load (citywide, без anchor'ов) ──────────────────────
+
+
+@dataclass
+class AvitoFullLoadCounters:
+ """Aggregate counters для run_avito_full_load.
+
+ Зеркало YandexFullLoadCounters — citywide room×price bisection без anchor/detail.
+ """
+
+ unique_fetched: int = 0
+ saved_inserted: int = 0
+ saved_updated: int = 0
+ errors_count: int = 0
+
+ def to_dict(self) -> dict[str, int]:
+ return {f.name: getattr(self, f.name) for f in fields(self)}
+
+
+async def run_avito_full_load(
+ db: Session,
+ *,
+ run_id: int,
+ price_cap_per_bucket: int = 1400,
+ request_delay_sec: float = 7.0,
+ concurrency: int = 5,
+ secondary_only: bool = True,
+ resume_run_id: int | None = None,
+) -> AvitoFullLoadCounters:
+ """Exhaustive региональный сбор Avito ЕКБ вторички (БЕЗ anchor'ов).
+
+ Avito anchor/citywide упирается в SERP-cap ~5000 результатов на запрос →
+ вторичка катастрофически недобрана. Решение зеркалит cian/yandex full_load:
+ один проход с партиционированием по КОМНАТНОСТИ × ЦЕНЕ (адаптивное бинарное
+ деление ценового диапазона через `pmin`/`pmax`) охватывает весь ЕКБ.
+
+ Параметры:
+ price_cap_per_bucket: максимум результатов на price-бакет (< Avito SERP-cap ~5000).
+ request_delay_sec: задержка между SERP-запросами (перезаписывает scraper default).
+ concurrency: параллельных page-фетчей в leaf-бакете.
+ secondary_only: отбрасывать новостройки (listing_segment=="novostroyki") до save.
+ resume_run_id: если задан — читает done_buckets из counters прошлого run и
+ пропускает уже завершённые бакеты. Передать run_id предыдущего (прерванного)
+ avito_full_load прогона для resume. Без поля — full walk с нуля.
+
+ Инкрементальный save: on_bucket коммитит каждый leaf-бакет в БД сразу после сбора.
+ Cooperative cancel: scrape_runs.is_cancelled проверяется per-bucket.
+ AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned'), как в
+ run_avito_city_sweep — partial results уже сохранены через on_bucket.
+ """
+ counters = AvitoFullLoadCounters()
+
+ # ── Checkpoint/resume: читаем done_buckets из прошлого run ───────────────
+ skip_set: set[str] = set()
+ if resume_run_id is not None:
+ _prev_row = db.execute(
+ text("SELECT counters FROM scrape_runs WHERE id = CAST(:rid AS bigint)"),
+ {"rid": resume_run_id},
+ ).fetchone()
+ if _prev_row is not None and _prev_row.counters:
+ _prev_counters: dict = (
+ _prev_row.counters if isinstance(_prev_row.counters, dict) else {}
+ )
+ skip_set = set(_prev_counters.get("done_buckets", []))
+ logger.info(
+ "avito-full-load run_id=%d: resuming from run %d — %d buckets already done",
+ run_id,
+ resume_run_id,
+ len(skip_set),
+ )
+ done: set[str] = set(skip_set)
+
+ def _on_bucket(bucket_key: str, lots: list) -> None: # type: ignore[type-arg]
+ """Инкрементальный save после каждого leaf-бакета."""
+ nonlocal done
+ if scrape_runs.is_cancelled(db, run_id):
+ logger.info("avito-full-load run_id=%d: cancel detected in on_bucket", run_id)
+ raise RuntimeError("cancelled")
+ if not lots:
+ done.add(bucket_key)
+ scrape_runs.update_heartbeat(
+ db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
+ )
+ return
+ inserted, updated = save_listings(db, lots, run_id=run_id)
+ # save_listings вызывает db.commit() внутри — данные в БД сразу
+ counters.saved_inserted += inserted
+ counters.saved_updated += updated
+ counters.unique_fetched += len(lots)
+ done.add(bucket_key)
+ scrape_runs.update_heartbeat(
+ db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)}
+ )
+ logger.info(
+ "avito-full-load run_id=%d: bucket=%s saved ins=%d upd=%d total_unique=%d",
+ run_id,
+ bucket_key,
+ inserted,
+ updated,
+ counters.unique_fetched,
+ )
+
+ def _on_progress(unique_count: int) -> None:
+ """Heartbeat per room-bucket (на случай пустых бакетов без on_bucket вызовов)."""
+ scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+
+ try:
+ async with AvitoScraper() as scraper:
+ scraper.request_delay_sec = request_delay_sec
+
+ await scraper.fetch_all_secondary(
+ price_cap_per_bucket=price_cap_per_bucket,
+ concurrency=concurrency,
+ secondary_only=secondary_only,
+ on_bucket=_on_bucket,
+ on_progress=_on_progress,
+ skip_buckets=skip_set if skip_set else None,
+ )
+
+ logger.info(
+ "avito-full-load run_id=%d: fetch done — unique=%d ins=%d upd=%d",
+ run_id,
+ counters.unique_fetched,
+ counters.saved_inserted,
+ counters.saved_updated,
+ )
+ scrape_runs.update_heartbeat(db, run_id, counters.to_dict())
+ scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
+ logger.info(
+ "avito-full-load run_id=%d done: unique=%d ins=%d upd=%d errors=%d",
+ run_id,
+ counters.unique_fetched,
+ counters.saved_inserted,
+ counters.saved_updated,
+ counters.errors_count,
+ )
+ return counters
+
+ except (AvitoBlockedError, AvitoRateLimitedError) as exc:
+ # IP-бан/rate-limit: partial results уже сохранены через on_bucket.
+ logger.error("avito-full-load run_id=%d: blocked/rate-limited — %s", run_id, exc)
+ counters.errors_count += 1
+ scrape_runs.mark_banned(
+ db,
+ run_id,
+ f"avito full load aborted: {exc}",
+ {**counters.to_dict(), "done_buckets": sorted(done)},
+ )
+ return counters
+
+ except RuntimeError as exc:
+ # on_bucket кидает RuntimeError("cancelled") при cooperative cancel
+ if str(exc) == "cancelled":
+ logger.info(
+ "avito-full-load run_id=%d: cancelled — partial results unique=%d ins=%d upd=%d",
+ run_id,
+ counters.unique_fetched,
+ counters.saved_inserted,
+ counters.saved_updated,
+ )
+ scrape_runs.mark_done(db, run_id, {**counters.to_dict(), "done_buckets": sorted(done)})
+ return counters
+ logger.exception("avito-full-load run_id=%d: fatal error", run_id)
+ scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+ raise
+
+ except Exception as exc:
+ logger.exception("avito-full-load run_id=%d: fatal error", run_id)
+ scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict())
+ raise
+
+
# ── DomClick citywide sweep (без anchor'ов) ────────────────────────────────
diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py
index d2304fb7..3f081fd8 100644
--- a/tradein-mvp/backend/app/services/scrapers/avito.py
+++ b/tradein-mvp/backend/app/services/scrapers/avito.py
@@ -24,7 +24,9 @@ URL patterns (EKB):
from __future__ import annotations
import asyncio
+import inspect
import logging
+import math
import re
from collections.abc import Callable
from datetime import date, datetime, timedelta, timezone
@@ -223,6 +225,24 @@ _FIREWALL_MARKERS = ("доступ ограничен", "проблема с ip"
# search-URL: /ekaterinburg/kvartiry/prodam/novostroyka-ASgBAgICAkSSA8YQ5geOUg
NOVOSTROYKA_SLUG = "novostroyka-ASgBAgICAkSSA8YQ5geOUg"
+# ── Exhaustive full-load (room×price bisection) ───────────────────────────────
+# Верхняя граница цены при первом рекурсивном делении (нет явного hi).
+# 200 млн ₽ заведомо выше ТОПа ЕКБ — дальние бакеты дадут 0 результатов.
+_AVITO_MAX_PRICE = 200_000_000
+# Минимальный ценовой диапазон для рекурсии; при hi - lo < MIN_BRACKET прекращаем
+# деление и пагинируем как есть (принимаем возможный «хвост»).
+_AVITO_MIN_BRACKET = 50_000
+# Avito SERP показывает ~50 карточек на страницу.
+_AVITO_OFFERS_PER_PAGE = 50
+
+# Счётчик результатов в шапке SERP: 1 178.
+# Текст может содержать NBSP (\xa0), -entity или обычные пробелы как разделители
+# разрядов — чистятся в _extract_total_count. [^<]* нежадно ловит весь текст до .
+_TOTAL_COUNT_RE = re.compile(
+ r'data-marker="page-title/count"[^>]*>(?P[^<]*)<',
+ flags=re.I,
+)
+
def _is_firewall_page(html: str) -> bool:
"""True если Avito вернул firewall-страницу IP-блока (на HTTP 200)."""
@@ -517,15 +537,377 @@ class AvitoScraper(BaseScraper):
path = f"/ekaterinburg/kvartiry/prodam/{NOVOSTROYKA_SLUG}"
return f"{self.base_url}{path}?{urlencode(params)}"
- def _build_rooms_url(self, room_slug: str, page: int = 1) -> str:
+ def _build_rooms_url(
+ self,
+ room_slug: str,
+ page: int = 1,
+ min_price: int | None = None,
+ max_price: int | None = None,
+ ) -> str:
"""T6: URL с фильтром по комнатности для всего ЕКБ (no geo).
Дробление по комнатности позволяет обойти cap ~5000 на category.
room_slug — берётся из ROOM_SLUGS (кодирует ASgB-параметры выборки).
+
+ min_price/max_price — опциональные ценовые границы (Avito `pmin`/`pmax`),
+ добавляются в query ТОЛЬКО если заданы (backward-compat: вызовы без цены
+ не меняют URL). Используются exhaustive full-load для room×price бисекции.
"""
- params = {"s": 104, "p": page}
+ params: dict[str, Any] = {"s": 104, "p": page}
+ if min_price is not None:
+ params["pmin"] = min_price
+ if max_price is not None:
+ params["pmax"] = max_price
return f"{self.base_url}/ekaterinburg/kvartiry/prodam/{room_slug}?{urlencode(params)}"
+ def _extract_total_count(self, html: str) -> int | None:
+ """Извлечь общее число результатов из Avito SERP (`page-title/count`).
+
+ Avito рендерит счётчик в 1 178.
+ Это аналог cian totalOffers — gate для бисекции (total > cap → split).
+ Парсинг: текст span → убрать NBSP (\xa0)/пробелы/` ` → int.
+
+ Returns:
+ int при успехе (включая 0), либо None при captcha/firewall/missing span.
+ """
+ m = _TOTAL_COUNT_RE.search(html)
+ if m is None:
+ return None
+ raw = m.group("count")
+ # Убрать NBSP (\xa0), -entity и обычные пробелы-разделители разрядов.
+ cleaned = raw.replace("\xa0", "").replace(" ", "").replace(" ", "")
+ if not cleaned:
+ return None
+ try:
+ return int(cleaned)
+ except (TypeError, ValueError):
+ return None
+
+ async def _fetch_rooms_page_html(
+ self,
+ room_slug: str,
+ page: int,
+ min_price: int | None,
+ max_price: int | None,
+ ) -> str | None:
+ """GET одной room×price SERP-страницы через anti-block pipeline.
+
+ Делегирует _fetch_serp_html (firewall-detect, IP-rotation, транзиентные
+ ретраи). AvitoBlockedError/AvitoRateLimitedError пробрасываются наверх.
+ """
+ url = self._build_rooms_url(room_slug, page, min_price, max_price)
+ return await self._fetch_serp_html(url, page)
+
+ async def fetch_all_secondary(
+ self,
+ *,
+ rooms_buckets: list[tuple[str, str]] | None = None,
+ price_cap_per_bucket: int = 1400,
+ max_pages_per_bucket: int = 100,
+ concurrency: int = 5,
+ secondary_only: bool = True,
+ on_bucket: Callable[..., Any] | None = None,
+ on_progress: Callable[[int], None] | None = None,
+ skip_buckets: set[str] | None = None,
+ ) -> list[ScrapedLot]:
+ """Exhaustive-загрузка Avito ЕКБ вторички через КОМНАТНОСТЬ × ЦЕНА.
+
+ Зеркало CianScraper.fetch_all_secondary. Обходит Avito SERP-cap (~5000
+ результатов на запрос): внутри каждой комнатности адаптивно бьёт ценовой
+ диапазон на бакеты так, чтобы в каждом total < price_cap_per_bucket →
+ пагинирует бакет полностью. Страницы leaf-бакета — параллельно (Semaphore).
+ Дедуп по source_id (dict seen).
+
+ Параметры:
+ rooms_buckets: список (name, slug) комнатностей (default: все ROOM_SLUGS,
+ вкл. студии и своб.планировку).
+ price_cap_per_bucket: максимум результатов в бакете перед делением (< 5000).
+ max_pages_per_bucket: верхний предел страниц на leaf-бакет (default=100).
+ concurrency: максимум параллельных page-фетчей в leaf-бакете (default=5).
+ secondary_only: если True (default) — отбрасывает новостройки
+ (listing_segment=="novostroyki") после парсинга, до save/on_bucket.
+ Avito тегирует сегмент по DOM-маркеру item-development-name в _parse_html.
+ on_bucket: опциональный callback(bucket_key, list[ScrapedLot]) после каждого
+ leaf-бакета. Может быть async или sync. Исключение прерывает прогон.
+ on_progress: опциональный callback(unique_count) для heartbeat (per room-bucket).
+ skip_buckets: множество ключей «room_label:lo:hi» уже завершённых бакетов —
+ пагинация и on_bucket для них пропускаются. Probe-запросы выполняются.
+
+ Возвращает list[ScrapedLot] уникальных лотов (дедуп по source_id/source_url).
+ """
+ _buckets = rooms_buckets if rooms_buckets is not None else _AVITO_DEFAULT_ROOMS
+ seen: dict[str, ScrapedLot] = {}
+
+ for name, slug in _buckets:
+ room_label = _avito_room_label(name)
+ logger.info(
+ "avito exhaustive: starting %s (price_cap=%d concurrency=%d secondary_only=%s)",
+ room_label,
+ price_cap_per_bucket,
+ concurrency,
+ secondary_only,
+ )
+ before = len(seen)
+ await self._walk_price_range(
+ room_slug=slug,
+ room_label=room_label,
+ lo=0,
+ hi=_AVITO_MAX_PRICE,
+ seen=seen,
+ price_cap_per_bucket=price_cap_per_bucket,
+ max_pages_per_bucket=max_pages_per_bucket,
+ concurrency=concurrency,
+ secondary_only=secondary_only,
+ on_bucket=on_bucket,
+ skip_buckets=skip_buckets,
+ )
+ room_collected = len(seen) - before
+ logger.info(
+ "avito exhaustive: %s done — collected %d (total unique=%d)",
+ room_label,
+ room_collected,
+ len(seen),
+ )
+ if on_progress is not None:
+ on_progress(len(seen))
+
+ logger.info("avito exhaustive: DONE — total unique=%d lots", len(seen))
+ return list(seen.values())
+
+ async def _walk_price_range(
+ self,
+ *,
+ room_slug: str,
+ room_label: str,
+ lo: int,
+ hi: int,
+ seen: dict[str, ScrapedLot],
+ price_cap_per_bucket: int,
+ max_pages_per_bucket: int,
+ concurrency: int = 5,
+ secondary_only: bool = True,
+ on_bucket: Callable[..., Any] | None = None,
+ skip_buckets: set[str] | None = None,
+ _depth: int = 0,
+ ) -> None:
+ """Рекурсивное адаптивное бинарное партиционирование ценового диапазона [lo, hi].
+
+ Зеркало CianScraper._walk_price_range. Алгоритм:
+ 1. Запросить page=1 с pmin=lo, pmax=hi → total (page-title/count).
+ 2. Если total <= cap → пагинировать leaf-бакет параллельно (concurrency).
+ 3. Если total > cap → разбить бакет пополам (рекурсия).
+ Guard: hi - lo < _AVITO_MIN_BRACKET → пагинировать как есть (WARNING).
+
+ После пагинации leaf-бакета вызывает on_bucket(bucket_key, bucket_lots).
+ bucket_key = "room_label:lo:hi" (checkpoint-ключ для resume).
+ on_bucket может быть async или sync. Исключение в on_bucket прерывает прогон.
+ skip_buckets: если bucket_key в skip_buckets — пагинация и on_bucket пропускаются.
+ secondary_only: новостройки (listing_segment=="novostroyki") отбрасываются после
+ сбора bucket_lots, до дедупа в seen и вызова on_bucket.
+ AvitoBlockedError/AvitoRateLimitedError из _fetch_serp_html пробрасываются наверх.
+ """
+ _lo_param = lo if lo > 0 else None
+
+ # ── Шаг 1: probe page 1 ────────────────────────────────────────────────
+ html = await self._fetch_rooms_page_html(room_slug, 1, _lo_param, hi)
+
+ total: int | None = None
+ if html is not None:
+ total = self._extract_total_count(html)
+
+ # Ретрай на captcha/missing-count: rotate IP + 1 retry.
+ if total is None:
+ logger.warning(
+ "avito: total=None for %s [%d, %d] depth=%d — rotating IP + retry",
+ room_label,
+ lo,
+ hi,
+ _depth,
+ )
+ rotated = await self._rotate_ip()
+ if rotated:
+ html = await self._fetch_rooms_page_html(room_slug, 1, _lo_param, hi)
+ if html is not None:
+ total = self._extract_total_count(html)
+
+ if total is None:
+ logger.error(
+ "avito: skipping bucket %s [%d, %d] — total unavailable after retry",
+ room_label,
+ lo,
+ hi,
+ )
+ return
+
+ logger.info(
+ "avito: %s [%d, %d] total=%d depth=%d",
+ room_label,
+ lo,
+ hi,
+ total,
+ _depth,
+ )
+
+ if total == 0:
+ return
+
+ # ── Шаг 2: деление или пагинация ──────────────────────────────────────
+ bracket_size = hi - lo
+ need_split = total > price_cap_per_bucket
+ too_narrow = bracket_size < _AVITO_MIN_BRACKET
+
+ if need_split and too_narrow:
+ logger.warning(
+ "avito: %s [%d, %d] total=%d > cap=%d but bracket=%d < MIN_BRACKET=%d "
+ "— paginating as-is (tail loss ~%d)",
+ room_label,
+ lo,
+ hi,
+ total,
+ price_cap_per_bucket,
+ bracket_size,
+ _AVITO_MIN_BRACKET,
+ max(0, total - price_cap_per_bucket),
+ )
+ need_split = False # принудительно пагинируем
+
+ if need_split:
+ mid = (lo + hi) // 2
+ # [lo, mid]
+ await self._walk_price_range(
+ room_slug=room_slug,
+ room_label=room_label,
+ lo=lo,
+ hi=mid,
+ seen=seen,
+ price_cap_per_bucket=price_cap_per_bucket,
+ max_pages_per_bucket=max_pages_per_bucket,
+ concurrency=concurrency,
+ secondary_only=secondary_only,
+ on_bucket=on_bucket,
+ skip_buckets=skip_buckets,
+ _depth=_depth + 1,
+ )
+ # [mid+1, hi]
+ await self._walk_price_range(
+ room_slug=room_slug,
+ room_label=room_label,
+ lo=mid + 1,
+ hi=hi,
+ seen=seen,
+ price_cap_per_bucket=price_cap_per_bucket,
+ max_pages_per_bucket=max_pages_per_bucket,
+ concurrency=concurrency,
+ secondary_only=secondary_only,
+ on_bucket=on_bucket,
+ skip_buckets=skip_buckets,
+ _depth=_depth + 1,
+ )
+ return
+
+ # ── Checkpoint-ключ leaf-бакета ──────────────────────────────────────
+ bucket_key = f"{room_label}:{lo}:{hi}"
+
+ # Если бакет уже завершён в предыдущем запуске — пропускаем пагинацию и on_bucket.
+ if skip_buckets and bucket_key in skip_buckets:
+ logger.info("avito: skip bucket %s — already done (resume)", bucket_key)
+ return
+
+ # ── Параллельная пагинация leaf-бакета ────────────────────────────────
+ pages_needed = math.ceil(total / _AVITO_OFFERS_PER_PAGE)
+ max_pages = min(pages_needed, max_pages_per_bucket)
+
+ if pages_needed > max_pages_per_bucket:
+ tail_loss = total - max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE
+ logger.warning(
+ "avito: %s [%d, %d] total=%d exceeds page cap "
+ "(max_pages=%d × %d=%d offers) — tail loss ~%d offers",
+ room_label,
+ lo,
+ hi,
+ total,
+ max_pages_per_bucket,
+ _AVITO_OFFERS_PER_PAGE,
+ max_pages_per_bucket * _AVITO_OFFERS_PER_PAGE,
+ tail_loss,
+ )
+
+ # Страница 1 уже есть (html из probe выше); остальные — параллельно.
+ sem = asyncio.Semaphore(concurrency)
+ first_url = self._build_rooms_url(room_slug, 1, _lo_param, hi)
+
+ async def _one_page(p: int) -> list[ScrapedLot]:
+ if p == 1:
+ return self._parse_html(html, source_url_base=first_url) if html else []
+ async with sem:
+ page_html = await self._fetch_rooms_page_html(room_slug, p, _lo_param, hi)
+ await asyncio.sleep(self.request_delay_sec)
+ if page_html is None:
+ logger.warning(
+ "avito: page_html=None %s [%d, %d] page=%d — skipping page",
+ room_label,
+ lo,
+ hi,
+ p,
+ )
+ return []
+ page_url = self._build_rooms_url(room_slug, p, _lo_param, hi)
+ return self._parse_html(page_html, source_url_base=page_url)
+
+ page_results = await asyncio.gather(
+ *[_one_page(p) for p in range(1, max_pages + 1)],
+ return_exceptions=True,
+ )
+
+ bucket_lots: list[ScrapedLot] = []
+ for p_idx, res in enumerate(page_results, start=1):
+ if isinstance(res, BaseException):
+ # Блокировки пробрасываем наверх (mark_banned в pipeline-обёртке).
+ if isinstance(res, AvitoBlockedError | AvitoRateLimitedError):
+ raise res
+ logger.warning(
+ "avito: page exception %s [%d, %d] page=%d — %r",
+ room_label,
+ lo,
+ hi,
+ p_idx,
+ res,
+ )
+ else:
+ bucket_lots.extend(res)
+
+ collected_this_bucket = len(bucket_lots)
+
+ # ── Фильтр новостроек (secondary_only) ────────────────────────────────
+ dropped_nb = 0
+ if secondary_only:
+ filtered = [lot for lot in bucket_lots if lot.listing_segment != "novostroyki"]
+ dropped_nb = collected_this_bucket - len(filtered)
+ bucket_lots = filtered
+
+ # Дедуп в общий seen
+ for lot in bucket_lots:
+ key = lot.source_id or lot.source_url
+ if key:
+ seen[key] = lot
+
+ logger.info(
+ "avito: %s [%d, %d] paginated=%d pages collected=%d dropped_nb=%d unique_total=%d",
+ room_label,
+ lo,
+ hi,
+ max_pages,
+ collected_this_bucket,
+ dropped_nb,
+ len(seen),
+ )
+
+ # ── on_bucket callback: инкрементальный save ──────────────────────────
+ if on_bucket is not None and bucket_lots:
+ res_cb = on_bucket(bucket_key, bucket_lots)
+ if inspect.isawaitable(res_cb):
+ await res_cb
+
# ── T6: citywide / byrooms fetch methods ──────────────────────────────────
async def fetch_city_wide(
@@ -909,6 +1291,34 @@ ROOM_SLUGS: list[tuple[str, str]] = [
("своб.планир.", "svobodnaya_planirovka-ASgBAgICAkSSA8YQygj8zzI"),
]
+# Дефолтные комнатности для exhaustive full-load — все ROOM_SLUGS (вкл. студии и
+# своб.планировку), т.к. каждая категория < SERP-cap после room×price бисекции.
+_AVITO_DEFAULT_ROOMS: list[tuple[str, str]] = list(ROOM_SLUGS)
+
+_NON_SLUG_CHARS_RE = re.compile(r"[^a-z0-9]+")
+
+
+def _avito_room_label(name: str) -> str:
+ """Стабильный bucket_key-префикс из человекочитаемого имени комнатности.
+
+ «1-комн.» → «room_1_komn», «студии» → «room_studii», «5+комн.» → «room_5_komn».
+ Транслитерация не нужна — берём латиницу/цифры из slug-friendly формы; для
+ кириллицы используем порядковый индекс категории как fallback стабильности.
+ """
+ # Цифры из имени (комнатность) + транслит ключевых маркеров для читабельности логов.
+ digits = "".join(ch for ch in name if ch.isdigit())
+ base = name.lower()
+ if "студ" in base:
+ return "room_studii"
+ if "планир" in base:
+ return "room_svobodnaya"
+ if digits:
+ return f"room_{digits}_komn"
+ # Fallback: нормализуем что есть (на случай новых категорий).
+ norm = _NON_SLUG_CHARS_RE.sub("_", base).strip("_")
+ return f"room_{norm or 'unknown'}"
+
+
_CSS_NOISE_RE = re.compile(r"\.?css-[a-z0-9_-]+\s*\{[^}]*\}", flags=re.I)
_NOT_ADDRESS_TAIL_RE = re.compile(
r"\s*(Площадь \d|от \d+\s?мин\.|css-[a-z0-9_-]+)",
diff --git a/tradein-mvp/backend/data/sql/127_scrape_schedules_seed_avito_full_load.sql b/tradein-mvp/backend/data/sql/127_scrape_schedules_seed_avito_full_load.sql
new file mode 100644
index 00000000..b1b9f097
--- /dev/null
+++ b/tradein-mvp/backend/data/sql/127_scrape_schedules_seed_avito_full_load.sql
@@ -0,0 +1,54 @@
+-- 127_scrape_schedules_seed_avito_full_load.sql
+-- Seed row для avito_full_load — exhaustive региональный сбор Avito ЕКБ ВТОРИЧКИ
+-- (run_avito_full_load, scrape_pipeline.py). Авторитетный источник вторички:
+-- room×price адаптивное партиционирование через pmin/pmax (БЕЗ anchor'ов),
+-- secondary_only, incremental on_bucket save. Обходит Avito SERP-cap ~5000/запрос,
+-- из-за которого anchor/citywide-обход добирал лишь ~4% активной вторички.
+--
+-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)).
+-- Idempotent: ON CONFLICT (source) DO UPDATE обновляет window/default_params при
+-- переустановке миграции, НО НЕ затирает enabled — оператор может выключить schedule
+-- вручную, и повторный прогон миграции это сохранит (ON CONFLICT не трогает enabled).
+--
+-- Окно 13:00-15:00 UTC (16:00-18:00 МСК) — намеренно отделено от прочих avito-задач
+-- на shared apw-прокси: avito_newbuilding (02-05), deactivate_stale_avito +
+-- avito_city_sweep (06-07), avito_detail_backfill (09-12). Так full_load не
+-- конкурирует за мобильный IP с другими avito-прогонами.
+--
+-- default_params соответствуют сигнатуре run_avito_full_load():
+-- price_cap_per_bucket — максимум результатов на price-бакет (< Avito SERP-cap ~5000).
+-- concurrency — параллельных page-фетчей в leaf-бакете.
+-- request_delay_sec — задержка между SERP-запросами (Avito агрессивен → 7.0s).
+-- secondary_only — отбрасывать новостройки (listing_segment="novostroyki") до save.
+
+BEGIN;
+
+-- next_run_at = следующее наступление day-window (tomorrow + window_start_hour),
+-- иначе при NULL get_due_schedules() выстрелит сразу после деплоя (вне окна).
+-- window_*_hour документированы в UTC (см. 052) → AT TIME ZONE 'UTC' фиксирует
+-- wall-clock час в UTC независимо от session TimeZone GUC.
+INSERT INTO scrape_schedules (
+ source,
+ enabled,
+ window_start_hour,
+ window_end_hour,
+ next_run_at,
+ default_params
+)
+VALUES
+(
+ 'avito_full_load',
+ true,
+ 13,
+ 15,
+ ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 13)) AT TIME ZONE 'UTC',
+ '{"price_cap_per_bucket": 1400, "concurrency": 5, "request_delay_sec": 7.0, "secondary_only": true}'::jsonb
+)
+ON CONFLICT (source) DO UPDATE SET
+ window_start_hour = EXCLUDED.window_start_hour,
+ window_end_hour = EXCLUDED.window_end_hour,
+ default_params = EXCLUDED.default_params,
+ updated_at = NOW();
+-- enabled НЕ в SET — сохраняем операторское значение (ON CONFLICT не трогает enabled).
+
+COMMIT;
diff --git a/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py b/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py
new file mode 100644
index 00000000..062af5b6
--- /dev/null
+++ b/tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py
@@ -0,0 +1,443 @@
+"""Тесты exhaustive-загрузки Avito: fetch_all_secondary + _walk_price_range.
+
+Зеркало tests/scrapers/test_cian_exhaustive.py. Мокаем сетевые вызовы
+(_fetch_rooms_page_html, _extract_total_count, _parse_html) для детерминированного
+контроля бисекции без реального HTTP.
+
+Дополнительно: _build_rooms_url с pmin/pmax (backward-compat) и _extract_total_count
+на реальном span-фрагменте `data-marker="page-title/count"`.
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+# Settings requires DATABASE_URL at import time — set dummy DSN before any app import.
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import pytest
+
+from app.services.scrapers.avito import (
+ _AVITO_MAX_PRICE,
+ _AVITO_MIN_BRACKET,
+ _AVITO_OFFERS_PER_PAGE,
+ AvitoScraper,
+)
+from app.services.scrapers.base import ScrapedLot
+
+
+def _make_lot(source_id: str, price: int = 3_000_000, segment: str = "vtorichka") -> ScrapedLot:
+ """Минимальный ScrapedLot для теста."""
+ return ScrapedLot(
+ source="avito",
+ source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/{source_id}",
+ source_id=source_id,
+ address="Екатеринбург",
+ price_rub=price,
+ listing_segment=segment,
+ )
+
+
+@pytest.fixture
+def scraper() -> AvitoScraper:
+ """AvitoScraper без реального __aenter__ (без HTTP/IP-rotation)."""
+ s = AvitoScraper.__new__(AvitoScraper)
+ s.name = "avito"
+ s.base_url = "https://www.avito.ru"
+ s.request_delay_sec = 0.0
+ s._cffi = MagicMock() # не None
+ s._browser = None
+ s.parse_failures = 0
+ return s
+
+
+# ── _build_rooms_url с pmin/pmax ──────────────────────────────────────────────
+
+
+def test_build_rooms_url_both_prices(scraper: AvitoScraper) -> None:
+ """min_price + max_price → pmin/pmax оба в query."""
+ url = scraper._build_rooms_url(
+ "1-komnatnye-X", page=2, min_price=3_000_000, max_price=4_000_000
+ )
+ assert "pmin=3000000" in url
+ assert "pmax=4000000" in url
+ assert "p=2" in url
+ assert "s=104" in url
+ assert "/kvartiry/prodam/1-komnatnye-X?" in url
+
+
+def test_build_rooms_url_min_only(scraper: AvitoScraper) -> None:
+ """Только min_price → pmin в query, pmax отсутствует."""
+ url = scraper._build_rooms_url("1-komnatnye-X", min_price=3_000_000)
+ assert "pmin=3000000" in url
+ assert "pmax" not in url
+
+
+def test_build_rooms_url_no_prices_backward_compat(scraper: AvitoScraper) -> None:
+ """Без цен URL не меняется (backward-compat для fetch_by_rooms)."""
+ url = scraper._build_rooms_url("1-komnatnye-X", page=1)
+ assert "pmin" not in url
+ assert "pmax" not in url
+ assert url.endswith("/kvartiry/prodam/1-komnatnye-X?s=104&p=1")
+
+
+# ── _extract_total_count ──────────────────────────────────────────────────────
+
+# Реальный span-фрагмент из живого Avito SERP (NBSP-разделитель разрядов).
+_REAL_COUNT_SPAN = '1\xa0178'
+
+
+def test_extract_total_count_real_span(scraper: AvitoScraper) -> None:
+ """Реальный span с NBSP `1\\xa0178` → 1178."""
+ html = f"Квартиры
{_REAL_COUNT_SPAN}"
+ assert scraper._extract_total_count(html) == 1178
+
+
+def test_extract_total_count_nbsp_entity(scraper: AvitoScraper) -> None:
+ """Вариант с ` `-entity → корректно очищается."""
+ html = '12 345'
+ assert scraper._extract_total_count(html) == 12345
+
+
+def test_extract_total_count_zero(scraper: AvitoScraper) -> None:
+ """0 результатов → 0 (не None)."""
+ html = '0'
+ assert scraper._extract_total_count(html) == 0
+
+
+def test_extract_total_count_missing_returns_none(scraper: AvitoScraper) -> None:
+ """Captcha/firewall (нет span) → None."""
+ html = "Доступ ограничен"
+ assert scraper._extract_total_count(html) is None
+
+
+# ── Бисекция: split / paginate / dedup ────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_splits_on_cap(scraper: AvitoScraper) -> None:
+ """total > cap → бакет делится; оба суб-бакета пагинируются."""
+ mid = _AVITO_MAX_PRICE // 2
+ call_log: list[tuple] = []
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ call_log.append((page, min_price, max_price))
+ return f"page={page} min={min_price} max={max_price}"
+
+ def fake_total(html: str) -> int | None:
+ import re
+
+ min_m = re.search(r"min=(\w+)", html)
+ max_m = re.search(r"max=(\w+)", html)
+ min_p = int(min_m.group(1)) if min_m and min_m.group(1) != "None" else 0
+ max_p = int(max_m.group(1)) if max_m and max_m.group(1) != "None" else _AVITO_MAX_PRICE
+ if max_p == mid:
+ return 800 # левая половина — пагинировать
+ if min_p == mid + 1:
+ return 700 # правая половина — пагинировать
+ return 3000 # весь диапазон — split
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ import re
+
+ page_m = re.search(r"page=(\d+)", html)
+ page = int(page_m.group(1)) if page_m else 1
+ min_m = re.search(r"min=(\w+)", html)
+ max_m = re.search(r"max=(\w+)", html)
+ min_p = min_m.group(1) if min_m else "0"
+ max_p = max_m.group(1) if max_m else "MAX"
+ if page == 1:
+ return [
+ _make_lot(f"lot_{min_p}_{max_p}_1"),
+ _make_lot(f"lot_{min_p}_{max_p}_2"),
+ ]
+ return []
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ lots = await scraper.fetch_all_secondary(
+ rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ )
+
+ assert len(lots) > 0
+ # Должны быть пробы суб-бакетов (после split весь диапазон не пагинируется)
+ probed_max = [mx for (p, _mn, mx) in call_log if p == 1]
+ assert any(mx == mid for mx in probed_max), f"Ожидался probe левой половины, got {probed_max}"
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_paginates_when_under_cap(scraper: AvitoScraper) -> None:
+ """total <= cap → пагинирует бакет полностью."""
+ pages_fetched: list[int] = []
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ pages_fetched.append(page)
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 100 # ceil(100/50) = 2 страницы
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ import re
+
+ page_m = re.search(r"page=(\d+)", html)
+ page = int(page_m.group(1)) if page_m else 1
+ if page <= 2:
+ return [_make_lot(f"lot_p{page}_{i}") for i in range(_AVITO_OFFERS_PER_PAGE)]
+ return []
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ lots = await scraper.fetch_all_secondary(
+ rooms_buckets=[("2-комн.", "2-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ )
+
+ assert len(lots) == 100, f"Ожидалось 100 уникальных лотов, получено {len(lots)}"
+ assert 2 in pages_fetched, "Должна быть запрошена страница 2"
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_min_bracket_guard(scraper: AvitoScraper) -> None:
+ """hi - lo < MIN_BRACKET → пагинируем как есть (нет бесконечной рекурсии)."""
+ call_count = 0
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ nonlocal call_count
+ call_count += 1
+ if call_count > 200:
+ raise RuntimeError("Бесконечная рекурсия — MIN_BRACKET guard не сработал")
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 5000 # всегда > cap → без guard делилось бы бесконечно
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ return []
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ await scraper._walk_price_range(
+ room_slug="1-komnatnye-X",
+ room_label="room_1_komn",
+ lo=1_000_000,
+ hi=1_000_000 + _AVITO_MIN_BRACKET - 1,
+ seen={},
+ price_cap_per_bucket=1400,
+ max_pages_per_bucket=100,
+ )
+
+ # Guard сработал: рекурсии нет, только один проход пагинации (probe + до cap страниц).
+ assert call_count <= 101, f"Слишком много вызовов ({call_count}) — guard не сработал"
+
+
+@pytest.mark.asyncio
+async def test_fetch_all_secondary_concurrent_pages_deduped(scraper: AvitoScraper) -> None:
+ """Параллельная пагинация не ломает дедуп по source_id."""
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 150 # ceil(150/50) = 3 страницы
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ import re
+
+ page_m = re.search(r"page=(\d+)", html)
+ page = int(page_m.group(1)) if page_m else 1
+ if page == 3:
+ return [_make_lot(f"lot_p1_{i}") for i in range(10)] # дубликаты стр 1
+ return [_make_lot(f"lot_p{page}_{i}") for i in range(10)]
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ lots = await scraper.fetch_all_secondary(
+ rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ concurrency=5,
+ )
+
+ # Страницы 1+2 = 20 уникальных, страница 3 дублирует → 20
+ assert len(lots) == 20, f"Ожидалось 20 уникальных лотов (дедуп), получено {len(lots)}"
+
+
+@pytest.mark.asyncio
+async def test_on_bucket_receives_key_and_lots(scraper: AvitoScraper) -> None:
+ """on_bucket получает (bucket_key, lots) — bucket_key = room_label:lo:hi."""
+ received: list[tuple[str, int]] = []
+
+ def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
+ received.append((bucket_key, len(lots)))
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 5 # 1 страница
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ import re
+
+ page_m = re.search(r"page=(\d+)", html)
+ page = int(page_m.group(1)) if page_m else 1
+ return [_make_lot(f"lot_p{page}_{i}") for i in range(5)]
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ await scraper.fetch_all_secondary(
+ rooms_buckets=[("2-комн.", "2-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ on_bucket=fake_on_bucket,
+ )
+
+ assert len(received) == 1, f"Ожидался 1 вызов on_bucket, получено {len(received)}"
+ key, count = received[0]
+ assert key == f"room_2_komn:0:{_AVITO_MAX_PRICE}", f"Неверный bucket_key: {key!r}"
+ assert count == 5
+
+
+@pytest.mark.asyncio
+async def test_skip_buckets_skips_pagination(scraper: AvitoScraper) -> None:
+ """skip_buckets: probe выполняется, но пагинация и on_bucket пропускаются."""
+ fetch_calls: list[int] = []
+ on_bucket_calls: list[str] = []
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ fetch_calls.append(page)
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 5
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ return [_make_lot("lot_1")]
+
+ def fake_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
+ on_bucket_calls.append(bucket_key)
+
+ skip_set = {f"room_1_komn:0:{_AVITO_MAX_PRICE}"}
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ lots = await scraper.fetch_all_secondary(
+ rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ on_bucket=fake_on_bucket,
+ skip_buckets=skip_set,
+ )
+
+ assert fetch_calls == [1], f"Ожидался ровно 1 вызов (probe), got {fetch_calls}"
+ assert on_bucket_calls == [], "on_bucket не должен вызываться для skip-бакета"
+ assert lots == []
+
+
+@pytest.mark.asyncio
+async def test_secondary_only_drops_novostroyki(scraper: AvitoScraper) -> None:
+ """secondary_only=True: новостройки (listing_segment=='novostroyki') отброшены."""
+ mixed = [
+ _make_lot("nb_1", segment="novostroyki"),
+ _make_lot("vt_1", segment="vtorichka"),
+ _make_lot("vt_2", segment="vtorichka"),
+ ]
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 3
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ import re
+
+ page_m = re.search(r"page=(\d+)", html)
+ page = int(page_m.group(1)) if page_m else 1
+ return mixed if page == 1 else []
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ lots = await scraper.fetch_all_secondary(
+ rooms_buckets=[("1-комн.", "1-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ secondary_only=True,
+ )
+
+ ids = {lot.source_id for lot in lots}
+ assert "nb_1" not in ids
+ assert ids == {"vt_1", "vt_2"}
+
+
+@pytest.mark.asyncio
+async def test_on_bucket_cancel_stops_run(scraper: AvitoScraper) -> None:
+ """on_bucket кидает RuntimeError('cancelled') → прогон прерывается."""
+
+ def cancel_on_bucket(bucket_key: str, lots: list[ScrapedLot]) -> None:
+ raise RuntimeError("cancelled")
+
+ async def fake_fetch(
+ room_slug: str, page: int, min_price: int | None, max_price: int | None
+ ) -> str:
+ return f"page={page}"
+
+ def fake_total(html: str) -> int | None:
+ return 5
+
+ def fake_parse(html: str, source_url_base: str) -> list[ScrapedLot]:
+ return [_make_lot("lot_1")]
+
+ with (
+ patch.object(scraper, "_fetch_rooms_page_html", side_effect=fake_fetch),
+ patch.object(scraper, "_extract_total_count", side_effect=fake_total),
+ patch.object(scraper, "_parse_html", side_effect=fake_parse),
+ patch.object(scraper, "_rotate_ip", return_value=False),
+ ):
+ with pytest.raises(RuntimeError, match="cancelled"):
+ await scraper.fetch_all_secondary(
+ rooms_buckets=[("1-комн.", "1-komnatnye-X"), ("2-комн.", "2-komnatnye-X")],
+ price_cap_per_bucket=1400,
+ on_bucket=cancel_on_bucket,
+ )
diff --git a/tradein-mvp/backend/tests/test_scheduler.py b/tradein-mvp/backend/tests/test_scheduler.py
index fa4b59cf..ae390a0a 100644
--- a/tradein-mvp/backend/tests/test_scheduler.py
+++ b/tradein-mvp/backend/tests/test_scheduler.py
@@ -459,3 +459,98 @@ async def test_scheduler_dispatch_routes_cian_full_load(monkeypatch: pytest.Monk
await _sched.trigger_cian_full_load_run(db, sch)
assert "cian_full_load" in triggered
+
+
+# ── avito_full_load: trigger_avito_full_load_run + dispatch ──────────────────
+
+
+_AVITO_FULL_LOAD_ROW = {
+ "source": "avito_full_load",
+ "window_start_hour": 13,
+ "window_end_hour": 15,
+ "default_params": {
+ "price_cap_per_bucket": 1400,
+ "concurrency": 5,
+ "request_delay_sec": 7.0,
+ "secondary_only": True,
+ },
+}
+
+
+@pytest.mark.asyncio
+async def test_trigger_avito_full_load_creates_run(monkeypatch: pytest.MonkeyPatch) -> None:
+ """trigger_avito_full_load_run claims run + launches run_avito_full_load via create_task."""
+ db = _FakeSchedulerDB()
+ calls: dict[str, list] = {"create_run": [], "load": []}
+
+ def fake_create_run(_d, *, source, params):
+ calls["create_run"].append(source)
+ db.running[source] = True
+ return 401
+
+ 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)
+
+ run_id = await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)
+
+ assert run_id == 401
+ assert calls["create_run"] == ["avito_full_load"]
+ # Drain event-loop so the asyncio.create_task runs
+ await asyncio.sleep(0)
+ assert len(calls["load"]) == 1
+ loaded_run_id, kwargs = calls["load"][0]
+ assert loaded_run_id == 401
+ # Params propagated from default_params; resume_run_id forced to None
+ assert kwargs["price_cap_per_bucket"] == 1400
+ assert kwargs["concurrency"] == 5
+ assert kwargs["request_delay_sec"] == 7.0
+ assert kwargs["secondary_only"] is True
+ assert kwargs["resume_run_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_trigger_avito_full_load_skips_when_running(monkeypatch: pytest.MonkeyPatch) -> None:
+ """trigger_avito_full_load_run returns None when source already running."""
+ db = _FakeSchedulerDB()
+ db.running["avito_full_load"] = True
+
+ calls: dict[str, int] = {"n": 0}
+
+ def fake_create_run(_d, *, source, params):
+ calls["n"] += 1
+ return 999
+
+ monkeypatch.setattr(_sched.runs_mod, "create_run", fake_create_run)
+
+ result = await _sched.trigger_avito_full_load_run(db, _AVITO_FULL_LOAD_ROW)
+
+ assert result is None
+ assert calls["n"] == 0
+
+
+@pytest.mark.asyncio
+async def test_scheduler_dispatch_routes_avito_full_load(monkeypatch: pytest.MonkeyPatch) -> None:
+ """Dispatch block в scheduler_loop роутит avito_full_load → trigger_avito_full_load_run."""
+ triggered: list[str] = []
+
+ async def fake_trigger(db, sch):
+ triggered.append(sch["source"])
+ return 1
+
+ monkeypatch.setattr(_sched, "trigger_avito_full_load_run", fake_trigger)
+
+ db = object()
+ sch = _AVITO_FULL_LOAD_ROW
+ source = sch["source"]
+ # Replicate the elif chain from scheduler_loop for avito_full_load
+ if source == "avito_city_sweep":
+ pass
+ elif source == "avito_full_load":
+ await _sched.trigger_avito_full_load_run(db, sch)
+
+ assert "avito_full_load" in triggered