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(browser_fetcher=bf) + save_detail_enrichment per listing
     Использует BrowserFetcher (camoufox) для получения JS-rendered HTML, потому что
     curl_cffi не рендерит priceChanges (они hidden в raw HTML).
  2. Houses block: SELECT houses WHERE cian_zhk_url IS NOT NULL
     AND NOT EXISTS (SELECT 1 FROM houses_price_dynamics WHERE house_id=...)
     LIMIT batch_size
     → fetch_newbuilding(cian_zhk_url) + save_newbuilding_enrichment per house
     Requires migration 071_houses_cian_zhk_url.sql (cian_zhk_url column).

Rate limit: scraper_settings.get_scraper_delay('cian') between requests.

Сигнал живости (#2725): батч дёргает `on_progress` на КАЖДОЙ сущности — caller
переливает это в scrape_runs.heartbeat_at. Пока колбэка не было, планировщик слал
heartbeat один раз ДО батча, а `reap_zombies` меряет ровно heartbeat с порогом 6 ч —
и добивал живые прогоны строго на 6-м часу (6 прод-прогонов, у пятерых внутри окна
писались строки offer_price_history, у одного — до 5.4 ч после старта). Ослаблять
критерий нельзя: пометка 'zombie' снимает running-блокировку источника
(`has_running_run`), без неё зависший прогон запер бы источник навсегда.
"""

from __future__ import annotations

import asyncio
import logging
import time
from collections import Counter
from collections.abc import Callable
from dataclasses import dataclass, field

from scraper_kit.browser_fetcher import BrowserFetcher, ban_kind_from_status
from scraper_kit.providers.cian.detail import fetch_detail, save_detail_enrichment
from scraper_kit.providers.cian.valuation import estimate_via_cian_valuation
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services.scraper_adapters import (
    RealMatcherAdapter,
    RealProxyProvider,
    RealScraperConfig,
)
from app.services.scraper_settings import get_scraper_delay

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
    valuations_total: int = 0
    valuations_processed: int = 0
    valuations_succeeded: int = 0
    valuations_failed: int = 0
    duration_sec: float = field(default=0.0)
    # #3196: отказы detail-фетча и их перепись (kind -> сколько раз). До этой правки
    # циановский прогон отдавал наверх только "не смогли обогатить", и scrape_runs.ban_kind
    # у него не проставлялся вовсе.
    listings_blocked: int = 0
    ban_kinds: Counter[str] = field(default_factory=Counter)

    @property
    def ban_kind(self) -> str:
        """Доминирующий диагноз отказов прогона (#3196), пригоден для scrape_runs.ban_kind.

        Считает тот же `_dominant_ban_kind`, что и остальные backfill-и: один вид →
        он; строгое большинство → оно; иначе (и при пустой переписи) — 'unknown'.
        Импорт локальный — таск не должен тянуть services на уровне модуля.
        """
        from app.services.scrape_runs import _dominant_ban_kind

        return _dominant_ban_kind(self.ban_kinds)


def _note_refusal(result: CianBackfillResult, status: int | None) -> str | None:
    """Записать отказ detail-фетча, если его природа диагностируема по HTTP-статусу (#3196).

    Инвариант: непустой `ban_kinds` ⟺ мы видели ответ с 403/429/5xx. `ban_kind_from_status`
    отвечает диагнозом только на такие статусы; на всё прочее (в т.ч. статуса нет — сайдкар
    не дошёл до навигации, или это HTTP 200 с промахом нашего парсера — дрейф разметки, а
    не отказ площадки) возвращает None, и тогда эта функция НЕ инкрементит ни
    `listings_blocked`, ни `ban_kinds` — только возвращает None вызывающему. Раньше
    недиагностируемые случаи писались как 'unknown' и это в scrape_runs.mark_banned
    (scheduler.py) превращало наши собственные сбои в фиктивный бан площадки.
    """
    kind = ban_kind_from_status(status)
    if kind is None:
        return None
    result.listings_blocked += 1
    result.ban_kinds[kind] += 1
    return kind


async def backfill_cian_history(
    db: Session,
    *,
    batch_size: int = 50,
    do_listings: bool = True,
    do_houses: bool = True,
    do_valuations: bool = False,
    dry_run: bool = False,
    on_progress: Callable[[CianBackfillResult], None] | None = None,
) -> 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 + valuations counted
            separately).
        do_listings: process listings batch (offer_price_history backfill).
        do_houses: process houses batch (houses_price_dynamics backfill via
            fetch_newbuilding). Requires migration 071_houses_cian_zhk_url.sql applied.
        do_valuations: process Cian Valuation Calculator batch (external_valuations backfill).
            Default False — opt-in because each call hits Cian auth-gated API.
        dry_run: skip all fetch+save; only count and log pending rows.
        on_progress: колбэк живости (#2725) — вызывается на каждой сущности ЛЮБОГО из
            трёх блоков, до её обработки, с текущим (мутируемым) result. Caller пишет
            heartbeat; исключения колбэка — на его совести (планировщик глушит их сам),
            здесь они прервали бы батч.

    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:
            # One BrowserFetcher instance shared across all listings in this batch.
            # priceChanges requires JS rendering — curl_cffi returns empty list (#1574).
            async with BrowserFetcher(source="cian", endpoint=settings.browser_http_endpoint) as bf:
                for row in rows:
                    listing_id: int = row["id"]
                    source_url: str = row["source_url"]
                    result.listings_processed += 1
                    if on_progress is not None:
                        on_progress(result)

                    enrichment = None
                    try:
                        enrichment = await fetch_detail(source_url, browser_fetcher=bf)
                    except Exception as exc:
                        kind = _note_refusal(result, bf.last_response_status)
                        logger.warning(
                            "cian_detail fetch failed for listing_id=%s url=%s: %s "
                            "(http=%s ban_kind=%s)",
                            listing_id,
                            source_url,
                            exc,
                            bf.last_response_status,
                            kind,
                        )
                        result.listings_failed_fetch += 1
                        await asyncio.sleep(delay)
                        continue

                    if enrichment is None:
                        kind = _note_refusal(result, bf.last_response_status)
                        logger.warning(
                            "cian_detail fetch returned None for listing_id=%s url=%s "
                            "(http=%s ban_kind=%s)",
                            listing_id,
                            source_url,
                            bf.last_response_status,
                            kind,
                        )
                        result.listings_failed_fetch += 1
                        await asyncio.sleep(delay)
                        continue

                    try:
                        # matcher (#2435): резолвит канонический дом из bti_data (address/geo
                        # листинга) — иначе BTI-снапшот распарсен, но выбрасывается.
                        save_detail_enrichment(
                            db, listing_id, enrichment, matcher=RealMatcherAdapter()
                        )
                        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
                        # Roll back to clean session state so next listing can proceed.
                        # save_detail_enrichment commits on success; on failure the
                        # transaction is left open/dirty — rollback to avoid session poison
                        # (same class of defect as the houses block below).
                        try:
                            db.rollback()
                        except Exception as rb_exc:
                            logger.warning(
                                "cian_detail rollback failed for listing_id=%s: %s",
                                listing_id,
                                rb_exc,
                            )

                    await asyncio.sleep(delay)

    # ── 2. Houses: missing houses_price_dynamics ──────────────────────────────
    if do_houses:
        # Kit's fetch_newbuilding() now accepts config= (issue #2322 fixed) — pass
        # RealScraperConfig() at the call site below so BrowserFetcher gets a real
        # endpoint instead of degrading to endpoint=None (#2397 Part D2).
        from scraper_kit.providers.cian.newbuilding import (
            fetch_newbuilding,
            save_newbuilding_enrichment,
        )

        house_rows = (
            db.execute(
                text("""
                    SELECT h.id, h.cian_zhk_url
                    FROM houses h
                    WHERE h.cian_zhk_url IS NOT NULL
                      AND NOT EXISTS (
                          SELECT 1 FROM houses_price_dynamics hpd
                          WHERE hpd.house_id = h.id
                      )
                    LIMIT :lim
                """),
                {"lim": batch_size},
            )
            .mappings()
            .all()
        )
        result.houses_total = len(house_rows)

        if dry_run:
            logger.info("dry_run: would process %d cian houses", result.houses_total)
        else:
            for hrow in house_rows:
                house_id: int = hrow["id"]
                zhk_url: str = hrow["cian_zhk_url"]
                result.houses_processed += 1
                if on_progress is not None:
                    on_progress(result)

                enrichment = None
                try:
                    # proxy_provider (#2767): тот же сожжённый env-узел бил и сюда —
                    # это второй вызывающий fetch_newbuilding, чинить надо оба.
                    enrichment = await fetch_newbuilding(
                        zhk_url,
                        config=RealScraperConfig(),
                        proxy_provider=RealProxyProvider(),
                    )
                except Exception as exc:
                    logger.warning(
                        "cian_newbuilding fetch failed for house_id=%s url=%s: %s",
                        house_id,
                        zhk_url,
                        exc,
                    )
                    result.houses_failed_fetch += 1
                    await asyncio.sleep(delay)
                    continue

                if enrichment is None:
                    logger.warning(
                        "cian_newbuilding fetch returned None for house_id=%s url=%s",
                        house_id,
                        zhk_url,
                    )
                    result.houses_failed_fetch += 1
                    await asyncio.sleep(delay)
                    continue

                try:
                    save_newbuilding_enrichment(db, house_id, enrichment)
                    result.houses_succeeded += 1
                except Exception as exc:
                    logger.warning(
                        "cian_newbuilding save failed for house_id=%s: %s", house_id, exc
                    )
                    result.houses_failed_save += 1
                    # Roll back to clean session state so next house can proceed.
                    # save_newbuilding_enrichment commits on success; on failure the
                    # transaction is left open/dirty — rollback to avoid session poison.
                    try:
                        db.rollback()
                    except Exception as rb_exc:
                        logger.warning(
                            "cian_newbuilding rollback failed for house_id=%s: %s",
                            house_id,
                            rb_exc,
                        )

                await asyncio.sleep(delay)

    # ── 3. Cian listings без external_valuations (price prediction backfill) ──
    if do_valuations:
        rows = (
            db.execute(
                text("""
                SELECT l.id, l.address, l.area_m2, l.rooms, l.floor, l.total_floors
                FROM listings l
                WHERE l.source = 'cian'
                  AND l.address IS NOT NULL
                  AND l.area_m2 IS NOT NULL
                  AND l.rooms IS NOT NULL
                  AND l.floor IS NOT NULL
                  AND l.total_floors IS NOT NULL
                  AND NOT EXISTS (
                      SELECT 1 FROM external_valuations ev
                      WHERE ev.source = 'cian_valuation'
                        AND ev.listing_id = l.id
                        AND ev.expires_at > NOW()
                  )
                LIMIT :lim
            """),
                {"lim": batch_size},
            )
            .mappings()
            .all()
        )
        result.valuations_total = len(rows)

        if dry_run:
            logger.info(
                "dry_run: would process %d cian listings for valuation", result.valuations_total
            )
        else:
            for row in rows:
                result.valuations_processed += 1
                if on_progress is not None:
                    on_progress(result)
                try:
                    cval = await estimate_via_cian_valuation(
                        db,
                        config=RealScraperConfig(),
                        address=row["address"],
                        total_area=float(row["area_m2"]),
                        rooms_count=int(row["rooms"]),
                        floor=int(row["floor"]),
                        total_floors=int(row["total_floors"]),
                        repair_type="cosmetic",
                        deal_type="sale",
                        use_cache=False,  # force fresh fetch — populate cache
                        listing_id=int(row["id"]),  # canonical link via mig 044
                    )
                    if cval is not None and cval.sale_price_rub:
                        result.valuations_succeeded += 1
                    else:
                        result.valuations_failed += 1
                except Exception as exc:
                    logger.warning(
                        "cian_valuation backfill failed for listing_id=%s: %s", row["id"], exc
                    )
                    result.valuations_failed += 1
                await asyncio.sleep(delay)

    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) "
        "valuations=%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.valuations_processed,
        result.valuations_total,
        result.valuations_succeeded,
        result.valuations_failed,
        result.duration_sec,
    )
    return result