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

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