"""Scheduled backfill: detail-enrichment for legacy yandex listings (#1553).

Nightly window 12:00-15:00 UTC (migration 113, source=yandex_detail_backfill).
Offset from avito_detail_backfill (09-12 UTC) to avoid parallel egress on shared IP.

Problem: ~3952 yandex listings have detail_enriched_at IS NULL.
Yandex city sweep does not call YandexDetailScraper — SERP data only.
area_m2 coverage 23%, living/kitchen 0%, repair_state 1% on prod.

Solution: single snapshot SELECT at start (guarantees termination), fetch each
offer detail page via curl_cffi AsyncSession (chrome120 + proxy) — mirrors
yandex_address_backfill.py which already gets full HTML from Yandex on prod.
Parse HTML via YandexDetailScraper.parse (pure, no network). Persist via
save_detail_enrichment. Track consecutive parse→None results; abort after
max_consecutive_blocks. Прогон с нулём обогащений теперь 'failed', не 'done'
(#2674, runs.mark_backfill_finished): на проде 31 прогон из 52 упирался ровно в
этот брейкер (attempted=5 failed=5) и все 31 назывались успешными. Остаток
снапшота уедет в следующую ночь через NULL detail_enriched_at.

Почему брейкер срабатывал так часто (разобрано 2026-08-06, замеры в комментарии
у OFFER_URL_PATTERN): в очереди лежали карточки новостроек, у которых source_url
ведёт на сайт застройщика, а не на realty.yandex.ru/offer/<id>/. Парсер отвергает
такие URL регуляркой ДО сети — это не капча, а предрешённый parse→None. Идут они
пачками, поэтому «5 подряд» набиралось на первых же строках и обрывало прогон
целиком. Теперь снапшот-SELECT берёт только то, что парсер в принципе может
разобрать, а размер отброшенного видно в counters.unenrichable_pending.

Why curl_cffi and not YandexDetailScraper.fetch_detail:
  fetch_detail uses BaseScraper._http_get (plain httpx, no proxy, no TLS
  fingerprinting). On datacenter IPs Yandex returns captcha / shell-HTML
  → parse always returns None → backfill would be 0% effective. The
  curl_cffi path (chrome120 impersonation + mobile proxy) is already proven
  by yandex_address_backfill, which fetches identical offer detail pages.
"""

from __future__ import annotations

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

from curl_cffi.requests import AsyncSession
from scraper_kit.providers.yandex.detail import YandexDetailScraper, save_detail_enrichment
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.services import scrape_runs as runs_mod

logger = logging.getLogger(__name__)

__all__ = [
    "OFFER_URL_PATTERN",
    "YandexDetailBackfillResult",
    "run_yandex_detail_backfill",
]

# Условие, при котором обогащение этого объявления вообще возможно (#2723-класс).
# `YandexDetailScraper.parse` первым делом ищет в URL `/offer/<цифры>/` и без него
# возвращает None ЕЩЁ ДО обращения к HTML (providers/yandex/detail.py:150) — то есть
# отказ предрешён регуляркой, а не капчей.
#
# Замер прода 2026-08-06: из 15 511 необогащённых yandex-объявлений 3 535 имеют
# source_url на сайт застройщика (macroserver.ru, prospect-federation.ru,
# strana.com, …) — так карточки новостроек ведут с выдачи Яндекса. Обогащено из
# них за всю историю 0; все 1 210 обогащённых — вида realty.yandex.ru/offer/<id>/.
#
# Вред не в бесполезности, а в том, что они идут ПАЧКАМИ (один свип — один
# застройщик) и упираются в брейкер «5 parse-None подряд», обрывающий ВЕСЬ прогон:
# 32 прогона из 53 закончились ровно так — attempted=5, enriched=0, 23 секунды.
# Плюс каждая такая попытка — запрос на чужой сайт, который мы всё равно выбросим.
OFFER_URL_PATTERN = "/offer/[0-9]+"


@dataclass
class YandexDetailBackfillResult:
    """Counters for one yandex detail backfill run."""

    attempted: int = 0
    enriched: int = 0
    failed: int = 0
    unenrichable_pending: int = 0
    duration_sec: float = field(default=0.0)

    def to_dict(self) -> dict[str, int]:
        return {
            "attempted": self.attempted,
            "enriched": self.enriched,
            "failed": self.failed,
            "unenrichable_pending": self.unenrichable_pending,
            "duration_sec": int(self.duration_sec),
        }


async def run_yandex_detail_backfill(
    db: Session,
    *,
    run_id: int,
    params: dict,
) -> YandexDetailBackfillResult:
    """Backfill detail_enriched_at for legacy yandex listings via curl_cffi + parse.

    Params (from default_params jsonb in scrape_schedules):
        batch_size: int -- snapshot size (SELECT LIMIT), default 800.
        budget_sec: float -- wall-clock budget per run, default 3600s.
        request_delay_sec: float -- delay between listings, default 5s.
        max_consecutive_blocks: int -- consecutive parse→None before abort, default 5.

    Fetch mechanism:
        curl_cffi AsyncSession(impersonate="chrome120") + scraper_proxy_url — mirrors
        yandex_address_backfill. On HTTP 200: pass resp.text to
        YandexDetailScraper().parse(html, offer_url). parse→None counts as a fail
        (possible captcha wall); consecutive None → abort after max_consecutive_blocks.

    Lifecycle: update_heartbeat -> snapshot -> loop with budget guard ->
    mark_backfill_finished (done / failed при нуле обогащений, #2674);
    mark_failed напрямую — только при исключении.
    """
    batch_size = int(params.get("batch_size", 800))
    budget_sec = float(params.get("budget_sec", 3600))
    request_delay_sec = float(params.get("request_delay_sec", 5.0))
    max_consecutive_blocks = int(params.get("max_consecutive_blocks", 5))

    counters = YandexDetailBackfillResult()
    current_counters: dict[str, int] = counters.to_dict()

    start = time.monotonic()

    try:
        runs_mod.update_heartbeat(db, run_id, current_counters)

        # SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
        # Priority: is_active DESC (active first), scraped_at DESC (newest first).
        # Гейт по OFFER_URL_PATTERN — тот же признак, по которому парсер отказывает
        # (см. комментарий у константы): в очередь не берём то, что заведомо
        # непарсимо, иначе пачка карточек застройщика обрывает прогон брейкером.
        snapshot = (
            db.execute(
                text(
                    """
                    SELECT id, source_url
                    FROM listings
                    WHERE source = 'yandex'
                      AND detail_enriched_at IS NULL
                      AND source_url IS NOT NULL
                      AND source_url ~ CAST(:offer_url_pattern AS text)
                    ORDER BY is_active DESC NULLS LAST, scraped_at DESC NULLS LAST
                    LIMIT CAST(:batch_size AS int)
                    """
                ),
                {"batch_size": batch_size, "offer_url_pattern": OFFER_URL_PATTERN},
            )
            .mappings()
            .all()
        )

        # Отброшенное не должно исчезнуть из виду: без этого счётчика «обогащено
        # 12 тыс. из 15,5 тыс.» снова стало бы необъяснимым нулём (#2674).
        counters.unenrichable_pending = int(
            db.execute(
                text(
                    """
                    SELECT count(*)
                    FROM listings
                    WHERE source = 'yandex'
                      AND detail_enriched_at IS NULL
                      AND source_url IS NOT NULL
                      AND source_url !~ CAST(:offer_url_pattern AS text)
                    """
                ),
                {"offer_url_pattern": OFFER_URL_PATTERN},
            ).scalar_one()
        )
        if counters.unenrichable_pending:
            logger.info(
                "yandex_detail_backfill: run_id=%d — %d объявлений вне очереди: "
                "source_url ведёт не на карточку Яндекса (%s), парсер их отвергает "
                "до сети",
                run_id,
                counters.unenrichable_pending,
                OFFER_URL_PATTERN,
            )

        if not snapshot:
            logger.info(
                "yandex_detail_backfill: run_id=%d -- no pending listings "
                "(detail_enriched_at IS NULL = 0), done",
                run_id,
            )
            # to_dict(), а не current_counters: пустая очередь при непустом
            # unenrichable_pending — самый важный случай этого счётчика.
            runs_mod.mark_done(db, run_id, counters.to_dict())
            return counters

        logger.info(
            "yandex_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
            "delay=%.1fs max_consecutive_none=%d)",
            run_id,
            len(snapshot),
            budget_sec,
            request_delay_sec,
            max_consecutive_blocks,
        )

        # Build proxies dict once — mirrors yandex_address_backfill.py
        _proxy = settings.scraper_proxy_url
        _proxies = {"http": _proxy, "https": _proxy} if _proxy else None

        consecutive_none = 0
        do_sleep = False
        scraper = YandexDetailScraper()

        async with AsyncSession(
            impersonate="chrome120",
            timeout=30.0,
            proxies=_proxies,
            headers={
                "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8",
            },
        ) as session:
            for idx, row in enumerate(snapshot):
                # Budget guard
                elapsed = time.monotonic() - start
                if elapsed > budget_sec:
                    logger.info(
                        "yandex_detail_backfill: run_id=%d -- budget %.0fs exhausted "
                        "(elapsed=%.1fs), stopping at #%d/%d",
                        run_id,
                        budget_sec,
                        elapsed,
                        idx,
                        len(snapshot),
                    )
                    break

                # Delay before each request except the first
                if do_sleep:
                    await asyncio.sleep(request_delay_sec)
                do_sleep = True

                source_url: str = row["source_url"]
                listing_id: int = row["id"]
                counters.attempted += 1

                try:
                    try:
                        resp = await session.get(source_url, allow_redirects=True)
                    except Exception as fetch_exc:
                        consecutive_none += 1
                        counters.failed += 1
                        logger.warning(
                            "yandex_detail_backfill: run_id=%d listing_id=%d "
                            "fetch error (consecutive=%d): %s",
                            run_id,
                            listing_id,
                            consecutive_none,
                            fetch_exc,
                        )
                        if consecutive_none >= max_consecutive_blocks:
                            logger.error(
                                "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
                                "errors. enriched=%d attempted=%d",
                                run_id,
                                consecutive_none,
                                counters.enriched,
                                counters.attempted,
                            )
                            break
                        continue

                    if resp.status_code != 200:
                        consecutive_none += 1
                        counters.failed += 1
                        logger.warning(
                            "yandex_detail_backfill: run_id=%d listing_id=%d "
                            "HTTP %d (consecutive=%d)",
                            run_id,
                            listing_id,
                            resp.status_code,
                            consecutive_none,
                        )
                        if consecutive_none >= max_consecutive_blocks:
                            logger.error(
                                "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
                                "non-200 responses. enriched=%d attempted=%d",
                                run_id,
                                consecutive_none,
                                counters.enriched,
                                counters.attempted,
                            )
                            break
                        continue

                    enrichment = scraper.parse(resp.text, offer_url=source_url)

                    if enrichment is None:
                        # parse→None: captcha wall / shell-HTML / no JSON-LD.
                        # Do not mark listing as done — retry next night.
                        consecutive_none += 1
                        counters.failed += 1
                        logger.warning(
                            "yandex_detail_backfill: run_id=%d listing_id=%d source_url=%s "
                            "-> parse None (consecutive=%d)",
                            run_id,
                            listing_id,
                            source_url,
                            consecutive_none,
                        )
                        if consecutive_none >= max_consecutive_blocks:
                            logger.error(
                                "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
                                "parse-None results (captcha wall?). enriched=%d attempted=%d",
                                run_id,
                                consecutive_none,
                                counters.enriched,
                                counters.attempted,
                            )
                            break
                        continue

                    consecutive_none = 0
                    if save_detail_enrichment(db, listing_id, enrichment):
                        counters.enriched += 1

                except Exception as exc:
                    counters.failed += 1
                    logger.warning(
                        "yandex_detail_backfill: save/iteration error for listing_id=%d: %s",
                        listing_id,
                        exc,
                    )
                    try:
                        db.rollback()
                    except Exception:
                        pass

                if counters.attempted % 25 == 0:
                    current_counters = counters.to_dict()
                    runs_mod.update_heartbeat(db, run_id, current_counters)

        counters.duration_sec = time.monotonic() - start
        current_counters = counters.to_dict()
        runs_mod.mark_backfill_finished(
            db, run_id, current_counters, source="yandex_detail_backfill"
        )
        logger.info(
            "yandex_detail_backfill: run_id=%d FINISHED -- attempted=%d enriched=%d "
            "failed=%d duration=%.1fs",
            run_id,
            counters.attempted,
            counters.enriched,
            counters.failed,
            counters.duration_sec,
        )
        return counters

    except Exception as exc:
        counters.duration_sec = time.monotonic() - start
        logger.exception(
            "yandex_detail_backfill: run_id=%d FAILED after %.1fs",
            run_id,
            counters.duration_sec,
        )
        runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters.to_dict())
        raise