"""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 (kit-профиль `DEFAULT_IMPERSONATE` + 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 берёт только то, что парсер в принципе может разобрать.

Но «не по тому URL» ≠ «нечего обогащать» (разобрано 2026-08-12, см. комментарий
у OFFER_ID_PATTERN): у ВСЕХ таких строк в source_id лежит yandex offerId, и по
собранному из него каноническому URL страница отдаётся и парсится. Поэтому в
очередь они входят по адресу, ВЫЧИСЛЕННОМУ из source_id, а counters разделены:
url_from_offer_id — сколько ждёт починки адреса, unenrichable_pending — сколько
не адресуемо вообще (ни offer-URL, ни числового source_id).

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 (kit-профиль impersonation + mobile proxy) is already proven
  by yandex_address_backfill, which fetches identical offer detail pages.

Природа блока (#3196): счётчика blocked у Яндекса не было вовсе, поэтому ветка
перевода прогона в 'banned' (scrape_runs.mark_backfill_finished) была недостижима
по построению — за 14 дней 12 done, 1 failed, 1 zombie и НОЛЬ банов. Теперь
non-200 ответ считается блоком, а его диагноз берётся из HTTP-статуса
(ban_kind_from_status: 403/429 → platform, 5xx → infra, прочее → unknown), а не
из текстовых маркеров страницы, снятых с чужой площадки.
"""

from __future__ import annotations

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

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

from app.services import scrape_runs as runs_mod
from app.services.proxy_egress import resolve_proxy_url
from app.services.scrape_runs import BAN_KIND_UNKNOWN

logger = logging.getLogger(__name__)

__all__ = [
    "CANONICAL_URL_SQL",
    "OFFER_ID_PATTERN",
    "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, …) — так карточки новостроек ведут с выдачи Яндекса.
#
# Вред не в бесполезности, а в том, что они идут ПАЧКАМИ (один свип — один
# застройщик) и упираются в брейкер «5 parse-None подряд», обрывающий ВЕСЬ прогон:
# 32 прогона из 53 закончились ровно так — attempted=5, enriched=0, 23 секунды.
# Плюс каждая такая попытка — запрос на чужой сайт, который мы всё равно выбросим.
OFFER_URL_PATTERN = "/offer/[0-9]+"

# ── «Непригодных» не бывает без причины (разобрано 2026-08-12) ────────────────
# Симптом: unenrichable_pending шесть прогонов подряд равнялся РОВНО 3535 — ни на
# единицу, при том что очередь обогащалась по ~500/прогон. Замер на проде:
#
#   * счётчик считается живым SELECT'ом, кэша/матвьюхи нет — арифметика честная;
#   * множество замкнуто: новых строк в него не приходит (0 из 6892 yandex-строк,
#     вставленных после самой свежей его строки, id 2583989), и выйти из него
#     нельзя (обогащение недостижимо, source_url не переписывается). Замкнутое
#     множество и обязано быть константой — вопрос был не «почему не растёт», а
#     «правда ли они непригодны».
#
# Непригодны они НЕ были. У всех 3535 в source_id лежит числовой yandex offerId
# (у 3523 он же продублирован в yandex_offer_id), а канонический адрес оффера из
# него собирается — это инвариант #2235 (`_canonical_source_url` в
# scraper_kit/providers/yandex/serp.py) и та же формула, которой миграция 164
# чинила легаси-строки. Живая проба 2026-08-12 прод-трактом (тот же прокси,
# curl_cffi с kit-профилем (на момент замера — Chrome 120), тот же parse): 6 из 6
# — HTTP 200 и parse OK, включая строки, чей сохранённый source_url — рекламный
# редирект na100.pro/go.php.
#
# Откуда взялся стухший адрес: source_url пишется ТОЛЬКО при вставке — его нет ни
# в `ON CONFLICT DO UPDATE`, ни в reconcile-UPDATE у `save_listings`. Значит #2235
# вылечил только новые строки, миграция 164 — только те легаси, чей URL ДЕЛИЛИ
# несколько строк (она искала дубли URL, а не непарсимость). Строки с уникальной
# ссылкой на карточку застройщика не попали ни туда, ни туда и носят адрес,
# замороженный в момент вставки, хотя свип переобходит ~511 из них в сутки.
#
# Поэтому адресуем такие строки вычисленным URL, а не сохранённым. Починка самой
# колонки (одноразовый UPDATE, тот же 164 без условия на дубли) — за миграцией:
# от неё зависит и yandex_address_backfill, где 1618 из 5217 кандидатов ходят
# на сайты застройщиков вместо Яндекса.
OFFER_ID_PATTERN = "^[0-9]+$"
CANONICAL_URL_SQL = "'https://realty.yandex.ru/offer/' || source_id || '/'"


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

    attempted: int = 0
    enriched: int = 0
    # Отказы площадки среди попыток (non-200 ответ) с разобранной природой.
    # Подмножество failed, а не отдельная корзина: смысл failed («попытка не дала
    # обогащения») не переписываем. Без этого счётчика ветка перевода прогона в
    # 'banned' у Яндекса недостижима по построению (#3196).
    blocked: int = 0
    failed: int = 0
    # Ждут обогащения, сохранённый source_url непарсим, но адрес восстановим из
    # source_id — идут в очередь по вычисленному URL. Должен убывать от прогона к
    # прогону; замер на месте = очередь снова читается не тем признаком.
    url_from_offer_id: int = 0
    # Ждут обогащения и адресовать их НЕЧЕМ: ни offer-URL, ни числового source_id.
    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,
            "blocked": self.blocked,
            "failed": self.failed,
            "url_from_offer_id": self.url_from_offer_id,
            "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=DEFAULT_IMPERSONATE) + 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).
        # В очередь идёт то, для чего есть АДРЕС, который парсер примет: либо
        # сохранённый source_url подходит под OFFER_URL_PATTERN, либо адрес
        # собирается из source_id (см. комментарий у OFFER_ID_PATTERN). Что шире
        # этого условия — гарантированный parse→None пачкой и обрыв по брейкеру.
        snapshot = (
            db.execute(
                text(
                    f"""
                    SELECT id,
                           CASE
                               WHEN source_url ~ CAST(:offer_url_pattern AS text)
                                   THEN source_url
                               ELSE {CANONICAL_URL_SQL}
                           END AS 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)
                            )
                            OR source_id ~ CAST(:offer_id_pattern AS text)
                      )
                    ORDER BY is_active DESC NULLS LAST, scraped_at DESC NULLS LAST
                    LIMIT CAST(:batch_size AS int)
                    """
                    # f-string здесь безопасен: CANONICAL_URL_SQL — литерал модуля,
                    # не пользовательский ввод. Всё изменяемое — bind-параметры.
                ),
                {
                    "batch_size": batch_size,
                    "offer_url_pattern": OFFER_URL_PATTERN,
                    "offer_id_pattern": OFFER_ID_PATTERN,
                },
            )
            .mappings()
            .all()
        )

        # Отброшенное не должно исчезнуть из виду: без этого счётчика «обогащено
        # 12 тыс. из 15,5 тыс.» снова стало бы необъяснимым нулём (#2674). И оно
        # разделено по ПРИЧИНЕ: одно число на две разные судьбы читалось как
        # «тут делать нечего» и держало 3535 квартир вне обогащения неделю.
        pending = db.execute(
            text(
                """
                SELECT
                    count(*) FILTER (
                        WHERE source_id ~ CAST(:offer_id_pattern AS text)
                    ) AS url_from_offer_id,
                    count(*) FILTER (
                        WHERE source_id IS NULL
                           OR source_id !~ CAST(:offer_id_pattern AS text)
                    ) AS unenrichable_pending
                FROM listings
                WHERE source = 'yandex'
                  AND detail_enriched_at IS NULL
                  AND (
                        source_url IS NULL
                        OR source_url !~ CAST(:offer_url_pattern AS text)
                  )
                """
            ),
            {"offer_url_pattern": OFFER_URL_PATTERN, "offer_id_pattern": OFFER_ID_PATTERN},
        ).one()
        counters.url_from_offer_id = int(pending.url_from_offer_id)
        counters.unenrichable_pending = int(pending.unenrichable_pending)
        if counters.url_from_offer_id:
            logger.info(
                "yandex_detail_backfill: run_id=%d — у %d объявлений сохранённый "
                "source_url не ведёт на карточку Яндекса; адресуем их по offerId из "
                "source_id (колонку чинит миграция, см. OFFER_ID_PATTERN)",
                run_id,
                counters.url_from_offer_id,
            )
        if counters.unenrichable_pending:
            logger.warning(
                "yandex_detail_backfill: run_id=%d — %d объявлений вне очереди: нет ни "
                "offer-URL (%s), ни числового source_id — адресовать их нечем",
                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.
        # Резолвер по источнику (#2825): пул scrape_proxies с учётом
        # scrape_proxy_source_bans, fallback на settings.scraper_proxy_url только если
        # пул пуст (легитимный dev/staging-сценарий). Пул не пуст, но все забанены/
        # нездоровы для yandex -- resolve_proxy_url бросает ProxyPoolExhaustedError
        # (fail-closed, #2616): НАРОЧНО не ловим здесь -- штатный except Exception ниже
        # (mark_failed + logger.exception + raise) уже даёт явную деградацию run'а с
        # понятным логом, отдельный catch не нужен.
        _proxy = resolve_proxy_url(db, "yandex")
        _proxies = {"http": _proxy, "https": _proxy} if _proxy else None

        consecutive_none = 0
        # #3196: два счётчика серий — разной природы. consecutive_none — весь подряд
        # идущий неуспех (фетч-ошибка, non-200, парс-None) и двигает только ABORT
        # парс-None (наш дефект/дрейф разметки). consecutive_blocks — подряд идущие
        # ТОЛЬКО non-200 ответы и двигает aborted_by_blocks/'banned': серию промахов
        # парсера при HTTP 200 нельзя засчитывать как серию блоков площадки.
        consecutive_blocks = 0
        do_sleep = False
        aborted_by_blocks = False
        # Перепись диагнозов блоков (kind -> сколько раз). Кратности нужны целыми:
        # доминирующий вид выбирает _dominant_ban_kind в scrape_runs.py (#3178).
        block_ban_kinds: Counter[str] = Counter()
        scraper = YandexDetailScraper()

        async with AsyncSession(
            impersonate=DEFAULT_IMPERSONATE,
            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:
                        # Природу отказа берём из HTTP-статуса, а не из текста
                        # страницы (#3196): 403/429 -> platform, 5xx -> infra.
                        # Статус, который о блоке ничего не сообщает, честно
                        # остаётся 'unknown' — это «не знаем, чей отказ», а не
                        # «отказа не было».
                        ban_kind = ban_kind_from_status(resp.status_code) or BAN_KIND_UNKNOWN
                        counters.blocked += 1
                        block_ban_kinds[ban_kind] += 1
                        consecutive_none += 1
                        consecutive_blocks += 1
                        counters.failed += 1
                        logger.warning(
                            "yandex_detail_backfill: run_id=%d listing_id=%d "
                            "HTTP %d ban_kind=%s (consecutive=%d)",
                            run_id,
                            listing_id,
                            resp.status_code,
                            ban_kind,
                            consecutive_blocks,
                        )
                        if consecutive_blocks >= max_consecutive_blocks:
                            aborted_by_blocks = True
                            logger.error(
                                "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive "
                                "non-200 responses. enriched=%d attempted=%d",
                                run_id,
                                consecutive_blocks,
                                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.
                        # НАРОЧНО не считаем блоком (#3196): при HTTP 200 отличить
                        # капчу от промаха нашего парсера нечем, а записав это в
                        # blocked, мы объявляли бы 'banned' (внешняя причина) любой
                        # прогон, сломанный на нашей стороне, и потеряли бы сигнал
                        # 'failed', ради которого он заведён (#2674).
                        consecutive_none += 1
                        # Площадка ОТВЕТИЛА (HTTP 200) — серии подтверждённых блоков
                        # нет, что бы ни случилось дальше с парсингом (#3196).
                        consecutive_blocks = 0
                        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
                    consecutive_blocks = 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",
            aborted_by_blocks=aborted_by_blocks,
            ban_kinds=block_ban_kinds,
        )
        logger.info(
            "yandex_detail_backfill: run_id=%d FINISHED -- attempted=%d enriched=%d "
            "blocked=%d failed=%d duration=%.1fs",
            run_id,
            counters.attempted,
            counters.enriched,
            counters.blocked,
            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