diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py new file mode 100644 index 00000000..0353513d --- /dev/null +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -0,0 +1,229 @@ +"""scrape_pipeline.py — Avito full orchestrator (Stage 2e). + +Координирует 5 шагов одного pipeline run: + 1. SEARCH — AvitoScraper().fetch_around(lat, lon, radius_m) + 2. SAVE — save_listings(db, lots) → (inserted, updated) + 3. GROUP — dedupe house_url из lots → unique set + 4. ENRICH_HOUSES — fetch_house_catalog + save_house_catalog_enrichment per unique house + 5. ENRICH_DETAIL — top-N listings (detail_enriched_at IS NULL) → fetch_detail + save + +Graceful degradation на каждом step: exception в одной house/listing +не валит весь pipeline. + +IMV ОТСУТСТВУЕТ — он on-demand (Stage 3 estimator integration), не cron. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from urllib.parse import urlparse + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services.scrapers.avito import AvitoScraper +from app.services.scrapers.avito_detail import fetch_detail, save_detail_enrichment +from app.services.scrapers.avito_houses import fetch_house_catalog, save_house_catalog_enrichment +from app.services.scrapers.base import ScrapedLot, save_listings + +logger = logging.getLogger(__name__) + + +# ── Result dataclasses ────────────────────────────────────────────────────── + + +@dataclass +class PipelineCounters: + """Counters одного pipeline run для логов/админки.""" + + lots_fetched: int = 0 + lots_inserted: int = 0 + lots_updated: int = 0 + unique_houses: int = 0 + houses_enriched: int = 0 + houses_failed: int = 0 + detail_attempted: int = 0 + detail_enriched: int = 0 + detail_failed: int = 0 + errors: list[str] = field(default_factory=list) + + +@dataclass +class PipelineResult: + """Полный результат full Avito pipeline.""" + + anchor_lat: float + anchor_lon: float + radius_m: int + counters: PipelineCounters + enrich_houses: bool + enrich_detail_top_n: int + + +# ── Main orchestrator ─────────────────────────────────────────────────────── + + +async def run_avito_pipeline( + db: Session, + lat: float, + lon: float, + radius_m: int = 1500, + *, + enrich_houses: bool = True, + enrich_detail_top_n: int = 10, +) -> PipelineResult: + """Full Avito search → houses → detail enrichment pipeline. + + Steps: + 1. SEARCH: AvitoScraper().fetch_around(lat, lon, radius_m) → list[ScrapedLot] + 2. SAVE: save_listings(db, lots) → (inserted, updated) + 3. GROUP: dedupe house_url из новых lots → unique house path set + 4. ENRICH_HOUSES (если enrich_houses=True): для каждого unique house path + → fetch_house_catalog + save_house_catalog_enrichment (try/except per house) + 5. ENRICH_DETAIL (если enrich_detail_top_n > 0): select top-N priority listings + (приоритет — detail_enriched_at IS NULL + price > 0) + → fetch_detail + save_detail_enrichment (try/except per listing) + 6. Return PipelineResult с counters + + Graceful degradation на каждом step — exception на одной house/listing + не должна провалить весь pipeline. + """ + counters = PipelineCounters() + lots: list[ScrapedLot] = [] + + # ── Step 1: search ────────────────────────────────────────────────────── + async with AvitoScraper() as scraper: + try: + lots = await scraper.fetch_around(lat, lon, radius_m) + counters.lots_fetched = len(lots) + logger.info( + "pipeline:search anchor=(%.4f,%.4f) r=%dm fetched=%d", + lat, + lon, + radius_m, + len(lots), + ) + except Exception as e: + logger.exception("pipeline:search failed") + counters.errors.append(f"search: {e}") + return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n) + + # ── Step 2: save listings ─────────────────────────────────────────────── + if lots: + try: + counters.lots_inserted, counters.lots_updated = save_listings(db, lots) + except Exception as e: + logger.exception("pipeline:save_listings failed") + counters.errors.append(f"save_listings: {e}") + + # ── Step 3: group by house ────────────────────────────────────────────── + # Dedupe house path (без host) для consistency; fetch_house_catalog принимает path + unique_house_paths: set[str] = set() + for lot in lots: + if lot.house_url: + try: + parsed = urlparse(lot.house_url) + path = parsed.path if parsed.path else lot.house_url + unique_house_paths.add(path) + except Exception: + continue + + counters.unique_houses = len(unique_house_paths) + logger.info("pipeline:group_by_house unique=%d", counters.unique_houses) + + # ── Step 4: enrich houses ─────────────────────────────────────────────── + if enrich_houses and unique_house_paths: + for house_path in unique_house_paths: + try: + enrichment = await fetch_house_catalog(house_path) + save_house_catalog_enrichment(db, enrichment) + counters.houses_enriched += 1 + except Exception as e: + logger.warning( + "pipeline:house_enrich failed for %s: %s", house_path, e + ) + counters.houses_failed += 1 + counters.errors.append(f"house {house_path}: {e}") + + # ── Step 5: enrich detail для top-N priority listings ────────────────── + if enrich_detail_top_n > 0: + # Priority: листинги ещё не enriched (detail_enriched_at IS NULL) + # в радиусе 2× от anchor ИЛИ только что scraped (последние 2 часа). + # OR обёрнут в скобки — приоритет AND выше OR. + priority_rows = db.execute( + text(""" + SELECT source_url + FROM listings + WHERE source = 'avito' + AND source_url IS NOT NULL + AND ( + ( + detail_enriched_at IS NULL + AND price_rub > 0 + AND ST_DWithin( + geom::geography, + ST_MakePoint(:lon, :lat)::geography, + :radius + ) + ) + OR ( + detail_enriched_at IS NULL + AND scraped_at > NOW() - INTERVAL '2 hours' + ) + ) + ORDER BY scraped_at DESC NULLS LAST + LIMIT :limit + """), + { + "lat": lat, + "lon": lon, + "radius": radius_m * 2, + "limit": enrich_detail_top_n, + }, + ).mappings().all() + + for row in priority_rows: + source_url: str = row["source_url"] + counters.detail_attempted += 1 + try: + # fetch_detail принимает absolute URL или relative path + item_url = ( + urlparse(source_url).path + if source_url.startswith("http") + else source_url + ) + enrichment_detail = await fetch_detail(item_url) + if save_detail_enrichment(db, enrichment_detail): + counters.detail_enriched += 1 + except Exception as e: + logger.warning( + "pipeline:detail_enrich failed for %s: %s", source_url, e + ) + counters.detail_failed += 1 + counters.errors.append(f"detail {source_url}: {e}") + + logger.info( + "pipeline:done anchor=(%.4f,%.4f) lots=%d (ins=%d/upd=%d) " + "houses=%d/%d detail=%d/%d errors=%d", + lat, + lon, + counters.lots_fetched, + counters.lots_inserted, + counters.lots_updated, + counters.houses_enriched, + counters.unique_houses, + counters.detail_enriched, + counters.detail_attempted, + len(counters.errors), + ) + + return PipelineResult(lat, lon, radius_m, counters, enrich_houses, enrich_detail_top_n) + + +# ── Public re-exports ─────────────────────────────────────────────────────── +__all__ = [ + "PipelineCounters", + "PipelineResult", + "run_avito_pipeline", +] diff --git a/tradein-mvp/backend/pyproject.toml b/tradein-mvp/backend/pyproject.toml index cc988bb7..b65afed4 100644 --- a/tradein-mvp/backend/pyproject.toml +++ b/tradein-mvp/backend/pyproject.toml @@ -25,9 +25,13 @@ dependencies = [ [dependency-groups] dev = [ "pytest>=8.0.0", + "pytest-asyncio>=0.24.0", "ruff>=0.5.0", ] +[tool.pytest.ini_options] +asyncio_mode = "auto" + [tool.setuptools.packages.find] include = ["app*"] diff --git a/tradein-mvp/backend/tests/test_scrape_pipeline.py b/tradein-mvp/backend/tests/test_scrape_pipeline.py new file mode 100644 index 00000000..cff44858 --- /dev/null +++ b/tradein-mvp/backend/tests/test_scrape_pipeline.py @@ -0,0 +1,86 @@ +"""Offline smoke for scrape_pipeline. Mocks scrapers + DB session.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from app.services.scrape_pipeline import ( + PipelineCounters, + PipelineResult, + run_avito_pipeline, +) +from app.services.scrapers.base import ScrapedLot + + +def _make_lot( + item_id: str, + house_url: str | None = None, + price: int = 5_000_000, +) -> ScrapedLot: + return ScrapedLot( + source="avito", + source_url=f"https://www.avito.ru/ekaterinburg/kvartiry/test_{item_id}", + source_id=item_id, + price_rub=price, + address="Test address", + house_source="avito" if house_url else None, + house_ext_id=house_url.rstrip("/").split("/")[-1] if house_url else None, + house_url=house_url, + listing_segment="vtorichka", + ) + + +def test_counters_dataclass_default() -> None: + c = PipelineCounters() + assert c.lots_fetched == 0 + assert c.unique_houses == 0 + assert c.errors == [] + + +@pytest.mark.asyncio +async def test_pipeline_search_failure_graceful() -> None: + """Если AvitoScraper.fetch_around raises — pipeline возвращает PipelineResult + с errors, не падает наружу.""" + mock_db = MagicMock() + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper.fetch_around = AsyncMock(side_effect=RuntimeError("test net error")) + + with patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper): + result = await run_avito_pipeline( + mock_db, 56.8, 60.6, 1000, enrich_houses=False, enrich_detail_top_n=0 + ) + + assert isinstance(result, PipelineResult) + assert result.counters.lots_fetched == 0 + assert len(result.counters.errors) == 1 + assert "search:" in result.counters.errors[0] + + +@pytest.mark.asyncio +async def test_pipeline_group_by_house_unique_set() -> None: + """3 lots с house_url'ами (2 одинаковых, 1 уникальный) → unique_houses=2.""" + mock_db = MagicMock() + lots = [ + _make_lot("1", "https://www.avito.ru/catalog/houses/ekb/test/100"), + _make_lot("2", "https://www.avito.ru/catalog/houses/ekb/test/100"), # дубль + _make_lot("3", "https://www.avito.ru/catalog/houses/ekb/other/200"), + _make_lot("4", None), # без дома + ] + mock_scraper = MagicMock() + mock_scraper.__aenter__ = AsyncMock(return_value=mock_scraper) + mock_scraper.__aexit__ = AsyncMock(return_value=None) + mock_scraper.fetch_around = AsyncMock(return_value=lots) + + with ( + patch("app.services.scrape_pipeline.AvitoScraper", return_value=mock_scraper), + patch("app.services.scrape_pipeline.save_listings", return_value=(4, 0)), + ): + result = await run_avito_pipeline( + mock_db, 56.8, 60.6, 1000, enrich_houses=False, enrich_detail_top_n=0 + ) + + assert result.counters.lots_fetched == 4 + assert result.counters.lots_inserted == 4 + assert result.counters.unique_houses == 2