diff --git a/tradein-mvp/backend/app/api/v1/admin.py b/tradein-mvp/backend/app/api/v1/admin.py index 07deca99..3acca303 100644 --- a/tradein-mvp/backend/app/api/v1/admin.py +++ b/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_inserted: 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 = 50, + 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__) diff --git a/tradein-mvp/backend/app/tasks/cian_history_backfill.py b/tradein-mvp/backend/app/tasks/cian_history_backfill.py new file mode 100644 index 00000000..84cf847a --- /dev/null +++ b/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_inserted: 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_inserted += 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_inserted, + result.houses_processed, + result.houses_total, + result.houses_succeeded, + result.houses_failed_fetch + result.houses_failed_save, + result.duration_sec, + ) + return result