diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 831623a7..e97f4e5c 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -9,6 +9,8 @@ Sources: - avito_city_sweep → run_avito_city_sweep (scrape_pipeline.py) + - 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) - cian_history_backfill → run_cian_backfill (this module, #560) - rosreestr_dkp_import → import_rosreestr_dkp (this module, #563) @@ -85,6 +87,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_newbuilding_sweep, run_cian_city_sweep, run_domclick_city_sweep, run_yandex_city_sweep, @@ -299,6 +302,44 @@ async def trigger_avito_city_sweep_run(db: Session, schedule_row: dict[str, Any] return run_id +async def trigger_avito_newbuilding_sweep_run( + db: Session, schedule_row: dict[str, Any] +) -> int | None: + """Создать новый scrape_runs + launch run_avito_newbuilding_sweep в asyncio.create_task. + + Зеркало trigger_avito_city_sweep_run, но citywide novostroyka-обход (без anchor/ + houses/detail-параметров): dedicated novostroyka-SERP → save. + + 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 {} + + # Spawn asyncio task — pass NEW session (avoid sharing with this scheduler tick) + async def _run() -> None: + run_db = SessionLocal() + try: + await run_avito_newbuilding_sweep( + run_db, + run_id=run_id, + pages=int(params.get("pages", 20)), + request_delay_sec=float(params.get("request_delay_sec", 7.0)), + ) + except Exception: + logger.exception("scheduler: run_avito_newbuilding_sweep crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + # Keep reference to avoid GC before task completes (RUF006) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered newbuilding_sweep run_id=%d", run_id) + return run_id + + async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any]) -> int | None: """Создать новый scrape_runs + launch run_yandex_city_sweep в asyncio.create_task. @@ -1350,6 +1391,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_newbuilding_sweep": + await trigger_avito_newbuilding_sweep_run(db, sch) elif source == "yandex_city_sweep": await trigger_yandex_city_sweep_run(db, sch) elif source == "cian_history_backfill": diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index c8b2531d..29c5f99a 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -990,6 +990,114 @@ async def run_avito_city_sweep( raise +# ── Avito newbuilding (novostroyka) citywide sweep ──────────────── + + +@dataclass +class NewbuildingSweepCounters: + """Aggregate counters для Avito novostroyka citywide sweep run.""" + + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_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_newbuilding_sweep( + db: Session, + *, + run_id: int, + pages: int = 20, + request_delay_sec: float = 7.0, +) -> NewbuildingSweepCounters: + """Citywide-обход ЕКБ-выборки только новостроек (novostroyka-filter) → save. + + В отличие от run_avito_city_sweep (anchor-based fetch_around + houses/detail/IMV), + это простой citywide paginated-обход через AvitoScraper.fetch_newbuildings: + dedicated novostroyka-SERP отдаёт 100% new-build карточки (listing_segment= + "novostroyki", newbuilding_id/newbuilding_url заполнены парсером). Сохраняем через + стандартный save_listings, ведём SERP-счётчики через scrape_runs. + + - Single shared AsyncSession/BrowserFetcher на весь sweep (один TLS fingerprint) + - AvitoBlockedError/AvitoRateLimitedError → mark_banned (status='banned') + - is_cancelled-чек до старта SERP-фазы + """ + counters = NewbuildingSweepCounters() + + browser_mode = settings.scraper_fetch_mode == "browser" + async with AsyncExitStack() as stack: + session: AsyncSession | None = None + shared_bf: BrowserFetcher | None = None + if browser_mode: + shared_bf = await stack.enter_async_context(BrowserFetcher(source="avito")) + else: + session = await stack.enter_async_context( + AsyncSession( + impersonate="chrome120", + timeout=25, + headers=_CHROME_HEADERS, + proxies=_avito_proxies(), + ) + ) + try: + if scrape_runs.is_cancelled(db, run_id): + logger.info("nb-sweep run_id=%d: cancelled before SERP phase", run_id) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + return counters + + scraper = AvitoScraper() + if browser_mode: + scraper._browser = shared_bf + else: + scraper._cffi = session + + try: + lots: list[ScrapedLot] = await scraper.fetch_newbuildings( + pages=pages, + delay_override_sec=request_delay_sec, + ) + except (AvitoBlockedError, AvitoRateLimitedError) as e: + logger.error("nb-sweep run_id=%d: SERP BLOCKED — %s", run_id, e) + scrape_runs.mark_banned(db, run_id, str(e), counters.to_dict()) + return counters + + counters.lots_fetched += len(lots) + if lots: + try: + ins, upd = save_listings(db, lots) + counters.lots_inserted += ins + counters.lots_updated += upd + except Exception as save_exc: + logger.exception( + "nb-sweep run_id=%d: save_listings failed: %s", run_id, save_exc + ) + counters.errors_count += 1 + try: + db.rollback() + except Exception: + pass + + scrape_runs.update_heartbeat(db, run_id, counters.to_dict()) + scrape_runs.mark_done(db, run_id, counters.to_dict()) + logger.info( + "nb-sweep run_id=%d done: lots=%d (ins=%d/upd=%d) errors=%d", + run_id, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.errors_count, + ) + return counters + + except Exception as exc: + logger.exception("nb-sweep run_id=%d: fatal error", run_id) + scrape_runs.mark_failed(db, run_id, str(exc), counters.to_dict()) + raise + + # ── Yandex city sweep ─────────────────────────────────────────── @@ -2266,8 +2374,7 @@ async def run_domclick_city_sweep( return counters logger.info( - "domclick-sweep run_id=%d: citywide SERP city_id=%d rooms=%s pages=%d " - "(watchdog %ds)", + "domclick-sweep run_id=%d: citywide SERP city_id=%d rooms=%s pages=%d (watchdog %ds)", run_id, city_id, _rooms, @@ -2333,11 +2440,13 @@ __all__ = [ "CianFullLoadCounters", "CitySweepCounters", "DomClickCitySweepCounters", + "NewbuildingSweepCounters", "PipelineCounters", "PipelineResult", "YandexCitySweepCounters", "YandexFullLoadCounters", "run_avito_city_sweep", + "run_avito_newbuilding_sweep", "run_avito_pipeline", "run_cian_city_sweep", "run_cian_full_load", diff --git a/tradein-mvp/backend/app/services/scrapers/avito.py b/tradein-mvp/backend/app/services/scrapers/avito.py index d5002826..d2304fb7 100644 --- a/tradein-mvp/backend/app/services/scrapers/avito.py +++ b/tradein-mvp/backend/app/services/scrapers/avito.py @@ -26,6 +26,7 @@ from __future__ import annotations import asyncio import logging import re +from collections.abc import Callable from datetime import date, datetime, timedelta, timezone from typing import Any from urllib.parse import urlencode, urljoin, urlparse, urlunparse @@ -216,6 +217,12 @@ def _build_sort_timestamp_map(html: str) -> dict[str, date]: # маркерам в начале документа (firewall-страница объёмна, не сканируем целиком). _FIREWALL_MARKERS = ("доступ ограничен", "проблема с ip", "firewall-container") +# EKB-фильтр «Новостройка» (slug категории kvartiry/prodam). Кодирует +# ASgB-параметры выборки только новостроек — SERP возвращает 100% new-build +# карточки (с data-marker="item-development-name"). Извлечён из реального +# search-URL: /ekaterinburg/kvartiry/prodam/novostroyka-ASgBAgICAkSSA8YQ5geOUg +NOVOSTROYKA_SLUG = "novostroyka-ASgBAgICAkSSA8YQ5geOUg" + def _is_firewall_page(html: str) -> bool: """True если Avito вернул firewall-страницу IP-блока (на HTTP 200).""" @@ -500,6 +507,16 @@ class AvitoScraper(BaseScraper): params = {"s": 104, "p": page} return f"{self.base_url}/ekaterinburg/kvartiry/prodam-ASgBAgICAUSSA8YQ?{urlencode(params)}" + def _build_newbuilding_url(self, page: int = 1) -> str: + """URL ЕКБ-выборки только новостроек (novostroyka-filter), сортировка по дате. + + Тот же shape параметров, что _build_citywide_url (s=104, p=page), но slug + категории NOVOSTROYKA_SLUG отдаёт исключительно new-build карточки. + """ + params = {"s": 104, "p": page} + 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: """T6: URL с фильтром по комнатности для всего ЕКБ (no geo). @@ -532,6 +549,30 @@ class AvitoScraper(BaseScraper): Returns: Список ScrapedLot (все страницы, дедуп по source_id). """ + all_lots = await self._paginate_sweep( + pages, + self._build_citywide_url, + label="citywide", + delay_override_sec=delay_override_sec, + ) + logger.info("avito fetch_city_wide pages=%d total_lots=%d", pages, len(all_lots)) + return all_lots + + async def _paginate_sweep( + self, + pages: int, + url_builder: Callable[[int], str], + *, + label: str, + delay_override_sec: float | None = None, + ) -> list[ScrapedLot]: + """Общий paginated-обход ЕКБ (citywide / novostroyka) с break-on-empty. + + url_builder(page) — функция построения URL страницы (citywide или + novostroyka). Сохраняет anti-block pipeline (_fetch_serp_html: firewall, + IP rotation, ретраи), дедуп по source_id, break-on-empty. label — только + для логов. Не меняет наблюдаемое поведение fetch_city_wide. + """ if delay_override_sec is not None: self.request_delay_sec = delay_override_sec @@ -539,21 +580,21 @@ class AvitoScraper(BaseScraper): seen_ids: set[str] = set() for page in range(1, pages + 1): - url = self._build_citywide_url(page) + url = url_builder(page) try: html = await self._fetch_serp_html(url, page) except (AvitoBlockedError, AvitoRateLimitedError): raise except Exception: - logger.exception("avito citywide page=%d fetch failed url=%s", page, url) + logger.exception("avito %s page=%d fetch failed url=%s", label, page, url) break if html is None: - logger.info("avito citywide page=%d: non-200 — end of pagination", page) + logger.info("avito %s page=%d: non-200 — end of pagination", label, page) break lots = self._parse_html(html, source_url_base=url) if not lots: - logger.info("avito citywide page=%d: 0 lots — end of pagination", page) + logger.info("avito %s page=%d: 0 lots — end of pagination", label, page) break new_lots = [lot for lot in lots if lot.source_id not in seen_ids] @@ -562,7 +603,8 @@ class AvitoScraper(BaseScraper): seen_ids.add(lot.source_id) all_lots.extend(new_lots) logger.info( - "avito citywide page=%d: %d lots (%d new, %d dedup'd, total=%d)", + "avito %s page=%d: %d lots (%d new, %d dedup'd, total=%d)", + label, page, len(lots), len(new_lots), @@ -571,8 +613,37 @@ class AvitoScraper(BaseScraper): ) if page < pages: await self.sleep_between_requests() + return all_lots - logger.info("avito fetch_city_wide pages=%d total_lots=%d", pages, len(all_lots)) + async def fetch_newbuildings( + self, + pages: int = 30, + *, + delay_override_sec: float | None = None, + ) -> list[ScrapedLot]: + """Обход ЕКБ-выборки только новостроек (novostroyka-filter), paginated. + + Avito отдаёт dedicated SERP с фильтром «Новостройка» — 100% new-build + карточек (с маркером застройщика → listing_segment="novostroyki", + newbuilding_id/newbuilding_url заполнены). Citywide (без geo/anchor). + Возвращает дедуплицированный список (по source_id), break-on-empty. + Сохраняет весь anti-block pipeline. Не трогает fetch_city_wide/fetch_around. + + Args: + pages: максимальное число страниц (default 30). + delay_override_sec: если задан — переопределяет request_delay_sec для + этого вызова. + + Returns: + Список ScrapedLot новостроек (все страницы, дедуп по source_id). + """ + all_lots = await self._paginate_sweep( + pages, + self._build_newbuilding_url, + label="newbuilding", + delay_override_sec=delay_override_sec, + ) + logger.info("avito fetch_newbuildings pages=%d total_lots=%d", pages, len(all_lots)) return all_lots async def fetch_by_rooms( diff --git a/tradein-mvp/backend/data/sql/123_avito_newbuilding_sweep_schedule.sql b/tradein-mvp/backend/data/sql/123_avito_newbuilding_sweep_schedule.sql new file mode 100644 index 00000000..de012f0c --- /dev/null +++ b/tradein-mvp/backend/data/sql/123_avito_newbuilding_sweep_schedule.sql @@ -0,0 +1,41 @@ +-- 123_avito_newbuilding_sweep_schedule.sql +-- Seed row для Avito newbuilding (novostroyka) citywide sweep. +-- +-- Avito отдаёт dedicated novostroyka-filtered SERP (slug +-- novostroyka-ASgBAgICAkSSA8YQ5geOUg) — 100% new-build карточки с маркером +-- застройщика (listing_segment='novostroyki', newbuilding_id заполнен парсером). +-- Текущий general sweep ловит новостройки лишь инцидентно (~20 строк); этот +-- dedicated обход даёт сотни. Citywide (без anchor/geo), save через стандартный путь. +-- +-- default_params: +-- pages — макс. число страниц SERP за прогон. +-- request_delay_sec — пауза между страницами (anti-block). +-- +-- next_run_at = завтра 02:00 UTC — чтобы НЕ выстрелить мгновенно на деплое (enabled +-- row с NULL next_run_at трактуется due немедленно); первый прогон в окне 02:00-05:00. +-- +-- ЗАВИСИМОСТИ: 052_scrape_schedules.sql (таблица + UNIQUE(source)). +-- Idempotent: ON CONFLICT (source) DO NOTHING — безопасно запускать повторно. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'avito_newbuilding_sweep', + true, + 2, + 5, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 2)) AT TIME ZONE 'UTC', + '{"pages": 20, "request_delay_sec": 8}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/fixtures/avito_serp_novostroyka.html b/tradein-mvp/backend/tests/fixtures/avito_serp_novostroyka.html new file mode 100644 index 00000000..3d79c627 --- /dev/null +++ b/tradein-mvp/backend/tests/fixtures/avito_serp_novostroyka.html @@ -0,0 +1,35 @@ + + +Avito novostroyka SERP fixture + +
+
+ 1-к. квартира, 40 м², 5/10 эт. + +

ул. Меридиан, 1

+
ЖК «Меридиан»
+ ЖК «Меридиан» +

2 часа назад

+
+
+ 2-к. квартира, 62 м², 8/12 эт. + +

ул. Северная, 7

+
ЖК «Северный квартал»
+ ЖК «Северный квартал» +

3 часа назад

+
+
+ + diff --git a/tradein-mvp/backend/tests/test_avito_newbuilding_sweep.py b/tradein-mvp/backend/tests/test_avito_newbuilding_sweep.py new file mode 100644 index 00000000..f90208c9 --- /dev/null +++ b/tradein-mvp/backend/tests/test_avito_newbuilding_sweep.py @@ -0,0 +1,245 @@ +"""Avito newbuilding (novostroyka) citywide sweep. + +Тесты (без сети, без реального DB): +1. _build_newbuilding_url — novostroyka-slug в пути, s=104, page param, без geo. +2. _parse_html на novostroyka-SERP fixture → listing_segment='novostroyki' + + newbuilding_id заполнен (DOM-маркер застройщика + zhk-якорь). +3. fetch_newbuildings break-on-empty + dedup (mirror fetch_city_wide). +4. NewbuildingSweepCounters — defaults + to_dict. +5. run_avito_newbuilding_sweep wiring — мокаем fetch_newbuildings + save_listings, + assert save вызван и SERP-счётчики записаны (mark_done). +""" + +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import scrape_pipeline +from app.services.scrapers.avito import NOVOSTROYKA_SLUG, AvitoScraper +from app.services.scrapers.base import ScrapedLot + +FIXTURE = Path(__file__).parent / "fixtures" / "avito_serp_novostroyka.html" + + +def _make_lot(source_id: str) -> ScrapedLot: + return ScrapedLot( + source="avito", + source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/{source_id}", + source_id=source_id, + price_rub=6_000_000, + listing_segment="novostroyki", + ) + + +# ── 1. _build_newbuilding_url ─────────────────────────────────────────────── + + +def test_build_newbuilding_url_contains_slug_and_sort() -> None: + s = AvitoScraper() + url = s._build_newbuilding_url(page=1) + assert NOVOSTROYKA_SLUG in url + assert "novostroyka-" in url + assert "geoCoords" not in url + assert "radius" not in url + assert "s=104" in url + assert "p=1" in url + assert "ekaterinburg/kvartiry/prodam/" in url + + +def test_build_newbuilding_url_page_param() -> None: + s = AvitoScraper() + url = s._build_newbuilding_url(page=2) + assert "novostroyka-" in url + assert "p=2" in url + + +def test_build_newbuilding_url_differs_from_citywide() -> None: + s = AvitoScraper() + assert s._build_newbuilding_url(page=1) != s._build_citywide_url(page=1) + + +# ── 2. _parse_html classifies novostroyka cards ───────────────────────────── + + +def test_parse_novostroyka_fixture_segments_and_newbuilding_id() -> None: + html = FIXTURE.read_text(encoding="utf-8") + s = AvitoScraper() + lots = s._parse_html(html, source_url_base=s._build_newbuilding_url(page=1)) + + assert len(lots) == 2 + for lot in lots: + assert lot.listing_segment == "novostroyki" + assert lot.newbuilding_id is not None + assert lot.newbuilding_url is not None + + ids = {lot.newbuilding_id for lot in lots} + assert "meridian-ekaterinburg" in ids + assert "severnyy-kvartal-ekaterinburg" in ids + + +# ── 3. fetch_newbuildings break-on-empty + dedup ──────────────────────────── + + +@pytest.mark.asyncio +async def test_fetch_newbuildings_break_on_empty() -> None: + s = AvitoScraper() + pages_data: list[list[ScrapedLot]] = [[_make_lot("A"), _make_lot("B")], []] + call_n = 0 + + async def mock_html(url: str, page: int) -> str: + assert "novostroyka-" in url + return f"{page}" + + def mock_parse(html: str, source_url_base: str) -> list[ScrapedLot]: + nonlocal call_n + idx = call_n + call_n += 1 + return pages_data[idx] if idx < len(pages_data) else [] + + with patch.object(s, "_fetch_serp_html", side_effect=mock_html): + with patch.object(s, "_parse_html", side_effect=mock_parse): + with patch.object(s, "sleep_between_requests", new=AsyncMock()): + result = await s.fetch_newbuildings(pages=10, delay_override_sec=0) + + assert len(result) == 2 + assert call_n == 2 # page1 + page2(empty) → stop + + +@pytest.mark.asyncio +async def test_fetch_newbuildings_dedup_by_source_id() -> None: + s = AvitoScraper() + dup = _make_lot("SAME") + pages_data: list[list[ScrapedLot]] = [[dup], [dup], []] + call_n = 0 + + async def mock_html(url: str, page: int) -> str: + return f"{page}" + + def mock_parse(html: str, source_url_base: str) -> list[ScrapedLot]: + nonlocal call_n + idx = call_n + call_n += 1 + return pages_data[idx] if idx < len(pages_data) else [] + + with patch.object(s, "_fetch_serp_html", side_effect=mock_html): + with patch.object(s, "_parse_html", side_effect=mock_parse): + with patch.object(s, "sleep_between_requests", new=AsyncMock()): + result = await s.fetch_newbuildings(pages=10, delay_override_sec=0) + + assert len(result) == 1 + assert result[0].source_id == "SAME" + + +def test_fetch_city_wide_url_unchanged() -> None: + """fetch_city_wide остаётся citywide-URL — рефактор не сменил его поведение.""" + s = AvitoScraper() + assert "novostroyka-" not in s._build_citywide_url(page=1) + assert "prodam-ASgBAgICAUSSA8YQ" in s._build_citywide_url(page=1) + + +# ── 4. NewbuildingSweepCounters ───────────────────────────────────────────── + + +def test_newbuilding_counters_defaults_and_to_dict() -> None: + from app.services.scrape_pipeline import NewbuildingSweepCounters + + c = NewbuildingSweepCounters() + assert c.lots_fetched == 0 + assert c.lots_inserted == 0 + assert c.lots_updated == 0 + assert c.errors_count == 0 + d = c.to_dict() + assert set(d.keys()) == {"lots_fetched", "lots_inserted", "lots_updated", "errors_count"} + + +# ── 5. run_avito_newbuilding_sweep wiring ─────────────────────────────────── + + +class _FakeRuns: + def __init__(self) -> None: + self.heartbeats: list[dict[str, int]] = [] + self.done: dict[str, int] | None = None + self.failed: tuple[str, dict[str, int]] | None = None + self.banned: tuple[str, dict[str, int]] | None = None + + def is_cancelled(self, _db: Any, _run_id: int) -> bool: + return False + + def update_heartbeat(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.heartbeats.append(dict(counters)) + + def mark_done(self, _db: Any, _run_id: int, counters: dict[str, int]) -> None: + self.done = dict(counters) + + def mark_failed(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.failed = (error, dict(counters)) + + def mark_banned(self, _db: Any, _run_id: int, error: str, counters: dict[str, int]) -> None: + self.banned = (error, dict(counters)) + + +def _install_runs(monkeypatch: pytest.MonkeyPatch, fake: _FakeRuns) -> None: + monkeypatch.setattr(scrape_pipeline.scrape_runs, "is_cancelled", fake.is_cancelled) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "update_heartbeat", fake.update_heartbeat) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_done", fake.mark_done) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_failed", fake.mark_failed) + monkeypatch.setattr(scrape_pipeline.scrape_runs, "mark_banned", fake.mark_banned) + + +@pytest.mark.asyncio +async def test_run_avito_newbuilding_sweep_saves_and_counts( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fake_runs = _FakeRuns() + _install_runs(monkeypatch, fake_runs) + + fake_lots = [_make_lot("N1"), _make_lot("N2"), _make_lot("N3")] + + async def fake_fetch_newbuildings( + self: AvitoScraper, + pages: int = 30, + *, + delay_override_sec: float | None = None, + ) -> list[ScrapedLot]: + return list(fake_lots) + + monkeypatch.setattr(AvitoScraper, "fetch_newbuildings", fake_fetch_newbuildings) + + save_calls: list[int] = [] + + def fake_save_listings(_db: Any, lots: Any, **_kw: Any) -> tuple[int, int]: + save_calls.append(len(lots)) + return (len(lots), 0) + + monkeypatch.setattr(scrape_pipeline, "save_listings", fake_save_listings) + # Force cffi-mode (no browser) so we don't spin up BrowserFetcher. + monkeypatch.setattr(scrape_pipeline.settings, "scraper_fetch_mode", "cffi") + + # Stub shared AsyncSession (used by the sweep in cffi-mode) — no real session. + fake_session = AsyncMock() + fake_session.__aenter__ = AsyncMock(return_value=fake_session) + fake_session.__aexit__ = AsyncMock(return_value=None) + monkeypatch.setattr(scrape_pipeline, "AsyncSession", lambda **_kw: fake_session) + + counters = await scrape_pipeline.run_avito_newbuilding_sweep( + MagicMock(), + run_id=42, + pages=2, + request_delay_sec=0.0, + ) + + assert save_calls == [3] + assert counters.lots_fetched == 3 + assert counters.lots_inserted == 3 + assert counters.lots_updated == 0 + assert fake_runs.done is not None + assert fake_runs.done["lots_fetched"] == 3 + assert fake_runs.failed is None + assert fake_runs.banned is None