From 1ddbf867f880620016a8a04471037d9864edbab0 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 13:55:18 +0300 Subject: [PATCH] =?UTF-8?q?feat(tradein):=20avito=20exhaustive=20full-load?= =?UTF-8?q?=20(room=C3=97price=20bisection)=20=E2=80=94=20mirror=20cian?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tradein-mvp/backend/app/services/scheduler.py | 45 ++ .../backend/app/services/scrape_pipeline.py | 172 +++++++ .../backend/app/services/scrapers/avito.py | 414 +++++++++++++++- ..._scrape_schedules_seed_avito_full_load.sql | 54 +++ .../tests/scrapers/test_avito_exhaustive.py | 443 ++++++++++++++++++ tradein-mvp/backend/tests/test_scheduler.py | 95 ++++ 6 files changed, 1221 insertions(+), 2 deletions(-) create mode 100644 tradein-mvp/backend/data/sql/127_scrape_schedules_seed_avito_full_load.sql create mode 100644 tradein-mvp/backend/tests/scrapers/test_avito_exhaustive.py 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 -- 2.45.3