feat(tradein): avito exhaustive full-load (room×price bisection) — mirror cian

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":

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'ов) ────────────────────────────────
 
 

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: <span data-marker="page-title/count">1&nbsp;178</span>.
+# Текст может содержать NBSP (\xa0), &nbsp;-entity или обычные пробелы как разделители
+# разрядов — чистятся в _extract_total_count. [^<]* нежадно ловит весь текст до </span>.
+_TOTAL_COUNT_RE = re.compile(
+    r'data-marker="page-title/count"[^>]*>(?P<count>[^<]*)<',
+    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 рендерит счётчик в <span data-marker="page-title/count">1&nbsp;178</span>.
+        Это аналог cian totalOffers — gate для бисекции (total > cap → split).
+        Парсинг: текст span → убрать NBSP (\xa0)/пробелы/`&nbsp;` → 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), &nbsp;-entity и обычные пробелы-разделители разрядов.
+        cleaned = raw.replace("\xa0", "").replace("&nbsp;", "").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_-]+)",

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;

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 = '<span class="some-css" data-marker="page-title/count">1\xa0178</span>'
+
+
+def test_extract_total_count_real_span(scraper: AvitoScraper) -> None:
+    """Реальный span с NBSP `1\\xa0178` → 1178."""
+    html = f"<div><h1>Квартиры</h1>{_REAL_COUNT_SPAN}</div>"
+    assert scraper._extract_total_count(html) == 1178
+
+
+def test_extract_total_count_nbsp_entity(scraper: AvitoScraper) -> None:
+    """Вариант с `&nbsp;`-entity → корректно очищается."""
+    html = '<span data-marker="page-title/count">12&nbsp;345</span>'
+    assert scraper._extract_total_count(html) == 12345
+
+
+def test_extract_total_count_zero(scraper: AvitoScraper) -> None:
+    """0 результатов → 0 (не None)."""
+    html = '<span data-marker="page-title/count">0</span>'
+    assert scraper._extract_total_count(html) == 0
+
+
+def test_extract_total_count_missing_returns_none(scraper: AvitoScraper) -> None:
+    """Captcha/firewall (нет span) → None."""
+    html = "<html><head><title>Доступ ограничен</title></head></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"<html>page={page} min={min_price} max={max_price}</html>"
+
+    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"<html>page={page}</html>"
+
+    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"<html>page={page}</html>"
+
+    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"<html>page={page}</html>"
+
+    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"<html>page={page}</html>"
+
+    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"<html>page={page}</html>"
+
+    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"<html>page={page}</html>"
+
+    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"<html>page={page}</html>"
+
+    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,
+            )

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