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