feat(tradein): scrape_pipeline.py — Avito full orchestrator (search → houses → detail) (#451)

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",
+]

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*"]
 

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