feat(tradein/cian): batch backfill task для offer_price_history + houses_price_dynamics (#535)

tradein-mvp/backend/app/api/v1/admin.py

@@ -1020,3 +1020,51 @@ async def scrape_cian_newbuilding(
         saved=saved,
         house_id=house_id,
     )
+
+
+class CianBackfillResp(BaseModel):
+    ok: bool
+    listings_total: int
+    listings_processed: int
+    listings_succeeded: int
+    listings_failed_fetch: int
+    listings_failed_save: int
+    price_changes_attempted: int
+    houses_total: int
+    houses_processed: int
+    houses_succeeded: int
+    houses_failed_fetch: int
+    houses_failed_save: int
+    duration_sec: float
+
+
+@router.post("/scrape/cian-backfill-history", response_model=CianBackfillResp)
+async def scrape_cian_backfill_history(
+    batch_size: int = Query(50, ge=1, le=200),
+    do_listings: bool = True,
+    do_houses: bool = True,
+    dry_run: bool = False,
+    db: Session = Depends(get_db),  # noqa: B008
+) -> CianBackfillResp:
+    """Batch backfill для Cian historical data.
+
+    Iterates Cian listings with missing offer_price_history, fetches Cian detail
+    pages, persists enrichment (price changes, views, agent, ceiling_height, etc.).
+
+    Houses block (houses_price_dynamics) is currently a no-op — requires
+    a cian_zhk_url column on houses (migration pending, see TODO in
+    app/tasks/cian_history_backfill.py).
+
+    Rate-limit: scraper_settings 'cian' delay (default 5s) between requests.
+    Idempotent: skips listings that already have offer_price_history rows.
+    """
+    from app.tasks.cian_history_backfill import CianBackfillResult, backfill_cian_history
+
+    result: CianBackfillResult = await backfill_cian_history(
+        db,
+        batch_size=batch_size,
+        do_listings=do_listings,
+        do_houses=do_houses,
+        dry_run=dry_run,
+    )
+    return CianBackfillResp(ok=True, **result.__dict__)

tradein-mvp/backend/app/tasks/cian_history_backfill.py

@@ -0,0 +1,171 @@
+"""Batch backfill для Cian historical data.
+
+Запускается:
+  - Manual через POST /admin/scrape/cian-backfill-history
+  - Future: in-app scheduler (после bootstrap'а Celery beat)
+
+Что делает:
+  1. SELECT listings WHERE source='cian' AND source_url IS NOT NULL
+     AND NOT EXISTS (SELECT 1 FROM offer_price_history WHERE listing_id=...)
+     LIMIT batch_size
+     → fetch_detail + save_detail_enrichment per listing
+  2. Houses block — TODO (PR #525): houses не имеют cian_url / cian_zhk_url колонки.
+     Колонка cian_internal_house_id (020_houses_alter_cian.sql) хранит numeric Cian ID,
+     но не ZHK URL, необходимый для fetch_newbuilding().
+     Backfill ЖК-динамики требует отдельной миграции, добавляющей cian_zhk_url к houses.
+     До тех пор do_houses=True — no-op (houses_total=0).
+
+Rate limit: scraper_settings.get_scraper_delay('cian') between requests.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass, field
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.services.scraper_settings import get_scraper_delay
+from app.services.scrapers.cian_detail import fetch_detail, save_detail_enrichment
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CianBackfillResult:
+    listings_total: int = 0
+    listings_processed: int = 0
+    listings_succeeded: int = 0
+    listings_failed_fetch: int = 0
+    listings_failed_save: int = 0
+    price_changes_attempted: int = 0
+    houses_total: int = 0
+    houses_processed: int = 0
+    houses_succeeded: int = 0
+    houses_failed_fetch: int = 0
+    houses_failed_save: int = 0
+    duration_sec: float = field(default=0.0)
+
+
+async def backfill_cian_history(
+    db: Session,
+    *,
+    batch_size: int = 50,
+    do_listings: bool = True,
+    do_houses: bool = True,
+    dry_run: bool = False,
+) -> CianBackfillResult:
+    """Iterate Cian listings + houses with missing history, fetch+save.
+
+    Args:
+        db: SQLAlchemy session (caller-owned; each entity commits internally via
+            save_detail_enrichment / save_newbuilding_enrichment).
+        batch_size: max entities to process per call (listings + houses counted separately).
+        do_listings: process listings batch (offer_price_history backfill).
+        do_houses: reserved for future houses_price_dynamics backfill (currently no-op —
+            see module docstring TODO).
+        dry_run: skip all fetch+save; only count and log pending rows.
+
+    Returns:
+        CianBackfillResult with per-domain counters + total wall-clock duration.
+    """
+    result = CianBackfillResult()
+    t0 = time.time()
+    delay = get_scraper_delay("cian")  # seconds; default 5.0
+
+    # ── 1. Listings: missing offer_price_history ──────────────────────────────
+    if do_listings:
+        rows = (
+            db.execute(
+                text("""
+                    SELECT l.id, l.source_url
+                    FROM listings l
+                    LEFT JOIN offer_price_history oph ON oph.listing_id = l.id
+                    WHERE l.source = 'cian'
+                      AND l.source_url IS NOT NULL
+                      AND oph.listing_id IS NULL
+                    LIMIT :lim
+                """),
+                {"lim": batch_size},
+            )
+            .mappings()
+            .all()
+        )
+        result.listings_total = len(rows)
+
+        if dry_run:
+            logger.info("dry_run: would process %d cian listings", result.listings_total)
+        else:
+            for row in rows:
+                listing_id: int = row["id"]
+                source_url: str = row["source_url"]
+                result.listings_processed += 1
+
+                enrichment = None
+                try:
+                    enrichment = await fetch_detail(source_url)
+                except Exception as exc:
+                    logger.warning(
+                        "cian_detail fetch failed for listing_id=%s url=%s: %s",
+                        listing_id,
+                        source_url,
+                        exc,
+                    )
+                    result.listings_failed_fetch += 1
+                    await asyncio.sleep(delay)
+                    continue
+
+                if enrichment is None:
+                    logger.warning(
+                        "cian_detail fetch returned None for listing_id=%s url=%s",
+                        listing_id,
+                        source_url,
+                    )
+                    result.listings_failed_fetch += 1
+                    await asyncio.sleep(delay)
+                    continue
+
+                try:
+                    save_detail_enrichment(db, listing_id, enrichment)
+                    result.listings_succeeded += 1
+                    result.price_changes_attempted += len(enrichment.price_changes or [])
+                except Exception as exc:
+                    logger.warning(
+                        "cian_detail save failed for listing_id=%s: %s", listing_id, exc
+                    )
+                    result.listings_failed_save += 1
+
+                await asyncio.sleep(delay)
+
+    # ── 2. Houses: missing houses_price_dynamics ──────────────────────────────
+    # TODO (PR #525): houses table has cian_internal_house_id (bigint) but no
+    # cian_zhk_url column. fetch_newbuilding() requires a ZHK URL
+    # (e.g. https://zhk-...-i.cian.ru/). Add migration `063_houses_add_cian_url.sql`
+    # with `ALTER TABLE houses ADD COLUMN IF NOT EXISTS cian_zhk_url text` and populate
+    # it during the existing SERP scrape (CianScraper → save_listings → house upsert).
+    # Until that column exists, do_houses is accepted but silently no-ops.
+    if do_houses:
+        logger.info(
+            "cian_backfill: houses block skipped — cian_zhk_url column not yet added "
+            "(see TODO in cian_history_backfill.py)"
+        )
+
+    result.duration_sec = time.time() - t0
+    logger.info(
+        "cian_backfill done: listings=%d/%d (ok=%d fetch_fail=%d save_fail=%d "
+        "price_rows=%d) houses=%d/%d (ok=%d fail=%d) %.1fs",
+        result.listings_processed,
+        result.listings_total,
+        result.listings_succeeded,
+        result.listings_failed_fetch,
+        result.listings_failed_save,
+        result.price_changes_attempted,
+        result.houses_processed,
+        result.houses_total,
+        result.houses_succeeded,
+        result.houses_failed_fetch + result.houses_failed_save,
+        result.duration_sec,
+    )
+    return result