"""T8a: Cian offer price-history backfill service.

Fetches Cian detail pages (curl_cffi, без Playwright) for listings that have
no rows in offer_price_history, extracts priceChanges from
_cianConfig['frontend-offer-card'] defaultState, writes to offer_price_history.

Deliberately NOT wired into scheduler — rate-limit risk with datacenter IPs.
Triggered manually via POST /api/v1/admin/scrape/cian-price-history.
"""

from __future__ import annotations

import asyncio
import logging
import time
from dataclasses import dataclass, field

# #2306: fetch_detail/save_detail_enrichment migrated to scraper_kit (byte-identical
# golden-parity была доказана против legacy cian_detail-модуля до его удаления,
# #2397 Part E2; extract_state/ScrapedLot parity-тесты убраны вместе с остальным
# legacy scrapers-каталогом, #2397 финальный шаг E — kit единственный живой путь).
from scraper_kit.providers.cian.detail import fetch_detail, save_detail_enrichment
from scraper_kit.proxy_errors import NoProxyAvailableError
from sqlalchemy import text
from sqlalchemy.orm import Session

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

logger = logging.getLogger(__name__)


class _PoolCurlConfig(RealScraperConfig):
    """RealScraperConfig с принудительно включённым pool-режимом curl (#2830).

    `USE_PROXY_POOL_CURL` задан только контейнеру `scraper` (docker-compose.prod.yml
    services.scraper.environment), а этот бэкфилл запускается ручкой
    `POST /admin/scrape/cian-price-history` в контейнере `backend`, где переменной нет
    → `settings.use_proxy_pool_curl` = False. С ней `providers/_proxy.py::curl_proxy_url`
    ИГНОРИРУЕТ переданный `proxy_provider` и уходит на статичный `SCRAPER_PROXY_URL`:
    один `proxy_provider=` был бы правкой без эффекта (зелёный тест, нулевой прод).

    Флаг — рубильник раскатки pool-режима для планировщика, а не решение «этому пути
    пул не нужен»: инцидент 2026-08-10 (#2830) — ровно про то, что нужен именно ему.
    """

    @property
    def use_proxy_pool_curl(self) -> bool:
        return True


@dataclass
class CianPriceHistoryResult:
    checked: int = 0
    saved: int = 0
    skipped: int = 0
    errors: int = 0
    duration_sec: float = field(default=0.0)


async def backfill_cian_price_history(
    db: Session,
    *,
    batch_size: int = 50,
    listing_id: int | None = None,
) -> CianPriceHistoryResult:
    """Fetch Cian detail pages and write missing price-history rows.

    Args:
        db: SQLAlchemy session (caller-owned; commits internally per listing).
        batch_size: max listings to process when listing_id is None.
        listing_id: process a single specific listing (ignores batch_size).

    Selection query picks cian listings with no existing offer_price_history rows.
    Idempotent: re-run safe via ON CONFLICT DO NOTHING in save_detail_enrichment.
    """
    result = CianPriceHistoryResult()
    t0 = time.time()
    delay = get_scraper_delay("cian")  # default 5.0s
    # Egress через пул с учётом `scrape_proxy_source_bans` (#2830): узел выбирает
    # `curl_proxy_url` внутри `fetch_detail`, он же на выходе возвращает вердикт
    # (mark_banned на CianBlockedError / mark_health / release).
    scraper_config = _PoolCurlConfig()
    proxy_provider = RealProxyProvider()

    if listing_id is not None:
        rows = (
            db.execute(
                text("""
                    SELECT l.id, l.source_url
                    FROM listings l
                    WHERE l.id = CAST(:lid AS bigint)
                      AND l.source = 'cian'
                      AND l.source_url IS NOT NULL
                """),
                {"lid": listing_id},
            )
            .mappings()
            .all()
        )
    else:
        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
                    ORDER BY l.id
                    LIMIT :lim
                """),
                {"lim": batch_size},
            )
            .mappings()
            .all()
        )

    result.checked = len(rows)
    logger.info(
        "cian_price_history backfill: checked=%d delay=%.1fs",
        result.checked,
        delay,
    )

    for i, row in enumerate(rows):
        lid: int = row["id"]
        url: str = row["source_url"]

        try:
            # config= обязателен — без него kit fetch_detail идёт напрямую, а без прокси
            # datacenter-IP блокируется Cian (#806). proxy_provider= — узел из пула
            # (#2830): раньше здесь был статичный SCRAPER_PROXY_URL, не знающий про
            # `scrape_proxy_source_bans`, и 403 от отбитого узла никому не сообщался.
            enrichment = await fetch_detail(
                url, config=scraper_config, proxy_provider=proxy_provider
            )
        except NoProxyAvailableError as exc:
            # Fail-closed (#2616): пул пуст/недоступен в проде. Остальные листинги
            # упрутся в то же самое — рвём батч сразу, а не 50 раз по 5 секунд с
            # логом, который читается как «Циан нас блокирует».
            logger.error(
                "cian_price_history: нет доступного прокси в пуле (%s) — батч прерван "
                "на listing_id=%s (обработано %d из %d)",
                exc,
                lid,
                i,
                len(rows),
            )
            result.errors += 1
            break
        except Exception as exc:
            logger.warning(
                "cian_price_history: fetch failed listing_id=%s url=%s: %s",
                lid,
                url,
                exc,
            )
            result.errors += 1
            await asyncio.sleep(delay)
            continue

        if enrichment is None:
            logger.warning(
                "cian_price_history: fetch returned None listing_id=%s url=%s",
                lid,
                url,
            )
            result.errors += 1
            await asyncio.sleep(delay)
            continue

        if not enrichment.price_changes:
            logger.debug("cian_price_history: no price_changes listing_id=%s", lid)
            result.skipped += 1
        else:
            try:
                # Count rows actually inserted: save_detail_enrichment skips
                # changes without change_time/price_rub and uses ON CONFLICT
                # DO NOTHING, so len(price_changes) overcounts on invalid
                # elements or idempotent re-runs. Diff the row count instead.
                before = db.execute(
                    text(
                        "SELECT COUNT(*) FROM offer_price_history "
                        "WHERE listing_id = CAST(:lid AS bigint)"
                    ),
                    {"lid": lid},
                ).scalar_one()
                # matcher (#2435): резолвит канонический дом из bti_data (address/geo
                # листинга) — иначе BTI-снапшот распарсен, но выбрасывается.
                save_detail_enrichment(db, lid, enrichment, matcher=RealMatcherAdapter())
                after = db.execute(
                    text(
                        "SELECT COUNT(*) FROM offer_price_history "
                        "WHERE listing_id = CAST(:lid AS bigint)"
                    ),
                    {"lid": lid},
                ).scalar_one()
                result.saved += max(0, int(after) - int(before))
            except Exception as exc:
                logger.warning("cian_price_history: save failed listing_id=%s: %s", lid, exc)
                result.errors += 1
                try:
                    db.rollback()
                except Exception:
                    pass
                await asyncio.sleep(delay)
                continue

        if i < len(rows) - 1:
            await asyncio.sleep(delay)

    result.duration_sec = time.time() - t0
    logger.info(
        "cian_price_history done: checked=%d saved=%d skipped=%d errors=%d %.1fs",
        result.checked,
        result.saved,
        result.skipped,
        result.errors,
        result.duration_sec,
    )
    return result