"""Scheduled backfill: detail-enrichment (Layer B) for domklik listings (issue #2000).

Nightly window 15:00-18:00 UTC (migration 175, source=domclick_detail_backfill).
Offset from avito_detail_backfill (09-12 UTC) and yandex_detail_backfill (12-15 UTC)
to avoid parallel egress across scraper sources sharing the tradein-browser pool.

Problem: DomClick SERP layer (Layer A, scraper_kit.providers.domclick.serp) discovers
listings via the BFF JSON API, but detail-enrichment (Layer B — repair_state,
living/kitchen area, year_built, owners_count, price history) requires a per-card
browser-fetch that is QRATOR-guarded (issue #2000). Two prior PRs solved the block at
the single-fetch level:
  - PR #2430 -- organic same-site SERP-origin navigation before the card fetch.
  - PR #2433 -- cookie-injection MVP (authenticated test-account session bypasses the
    QRATOR reputation-block even on an already-suspicious proxy-IP).
Both are wired together in the debug endpoint `POST /scrape/domclick/debug/detail-fetch`
(app/api/v1/admin.py). This module ports that same session -> cookies -> fetch_detail
wiring into the production scheduled orchestrator (previously only reachable manually).

Solution: single snapshot SELECT at start (guarantees termination) + one BrowserFetcher
per run (async context manager, source="domclick" -- узел берётся из ОБЩЕГО пула;
выделенного узла у Домклика больше нет, резервацию сняла миграция 253 (#2800),
на 13.08 все четыре узла имеют provider_affinity='any') + cookies loaded ONCE via
domclick_session.load_session(db) and threaded into every fetch_detail() call.

NAMING TRAP (verified live against prod DB 2026-07-04, do NOT "fix" this anywhere):
listings.source value for DomClick rows is 'domklik' (with a "k") -- that is the
DATA/business identifier used across listings.source, DomClickScraper.name/.source,
schemas, PDF exporter labels. 'domclick' (with a "c") is the newer INFRA/transport
identifier -- BrowserFetcher(source="domclick"), scrape_proxies.provider_affinity,
the SSRF host-guard, cookie-session tables. Both strings are correct in their own
context; this split is intentional and pre-existing, not a bug to unify here.

No curl fallback: unlike Avito's dual-mode (curl_cffi backconnect OR browser),
DomClick only has a BrowserFetcher path -- scraper_kit.providers.domclick.detail.
fetch_detail() raises DomClickBlockedError if the browser fetch itself fails. Exactly
one BrowserFetcher is constructed per run.

Exception triad differs from Avito:
  - DomClickBlockedError (QRATOR challenge page OR any browser-fetch failure) --
    increments consecutive_blocks, abort once max_consecutive_blocks is hit.
    Статус такого прогона — 'banned' (#2674, см. runs.mark_backfill_finished):
    блок это external constraint, не наш баг, но и НЕ успех — раньше здесь стоял
    mark_done, и 24 из 30 прогонов с нулём обогащений назывались успешными.
    No IP-rotation/cooldown recovery step exists here -- an aborted run simply
    retries the remaining backlog next window.
    УСТАРЕВШЕЕ ОБОСНОВАНИЕ, снято 13.08: здесь стояло «DomClick uses one dedicated
    residential proxy, not a rotating pool». Это перестало быть правдой на миграции
    253 (#2800), снявшей резервацию узла; сегодня узлов четыре и все общие. То есть
    отсутствие ротации больше НЕ следует из «ротировать нечего» — это просто
    непринятое решение. Разбор цены и рисков: #2854 (блок бьёт внутри первой
    комнатной корзины, buckets_completed=0 во ВСЕХ прогонах; свежий узел, судя по
    длительности до блока 111-332 с, получает свой бюджет).
    ОГРАНИЧЕНИЕ (#2764): диагноз scrape_runs.ban_kind этот прогон НЕ передаёт и
    получает 'unknown'. Один и тот же DomClickBlockedError поднимается и на
    распознанном QRATOR-маркере (площадка), и на любом сбое браузерного fetch
    (наш тракт) -- см. providers/domclick/detail.py::fetch_detail. Пока эти два
    случая не разведены отдельным подтипом (как AvitoSidecarUnavailableError у
    avito), любой диагноз отсюда был бы назначенным, а не установленным.
  - DomClickParseError (__SSR_STATE__ missing/malformed -- schema drift, NOT a
    block) -- counted as failed++, logged, does NOT touch consecutive_blocks and
    does NOT abort the run (neutral to the block-breaker, mirrors how Avito's
    AvitoListingGoneError is neutral to its breaker for a different reason).
  - There is no "listing gone / 404" exception type for DomClick yet -- not
    invented here.

Cookie injection is mandatory wiring, not optional: cookies are loaded ONCE per run.
If None (no valid session uploaded / expired) -- the run still proceeds (cookie-
injection is a QRATOR-defeat mechanism, not a hard requirement; organic SERP-origin
navigation from PR #2430 still applies) but an ERROR is logged once at run start so
operators notice the test-account session needs refreshing via
`POST /scrape/domclick/upload-cookies` (no auto-login -- documented MVP limitation,
see app/services/domclick_session.py module docstring).

#2674: раньше это был WARNING, который в скрапер-контейнере событием не становится
(LoggingIntegration event_level=ERROR) — куки протухли 2026-08-03 и об этом никто не
узнал. Теперь два сигнала вместо одного: ERROR по факту (_alert_domclick_cookies) и
ERROR ЗАРАНЕЕ, пока куки ещё живы (_warn_before_domclick_cookies_expire) — по образцу
#2658 для Циана, ручное обновление кук требует запаса времени.
"""

from __future__ import annotations

import asyncio
import logging
import random
import time
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta

from scraper_kit.browser_fetcher import BrowserFetcher
from scraper_kit.domclick_exceptions import DomClickBlockedError, DomClickParseError
from scraper_kit.providers.domclick.detail import fetch_detail, save_detail_enrichment
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.shutdown import shutdown_requested
from app.services import domclick_session as domclick_session_svc
from app.services import scrape_runs as runs_mod

logger = logging.getLogger(__name__)

__all__ = [
    "DomClickDetailBackfillResult",
    "run_domclick_detail_backfill",
]


def _alert_domclick_cookies(db: Session, run_id: int) -> None:
    """Громкий сигнал «обогащение идёт без кук» — logger.error, не warning (#2674).

    В контейнере скрапера GlitchTip поднят с LoggingIntegration(event_level=ERROR)
    (scheduler_main.py), поэтому прежний WARNING событием не становился: куки протухли
    на проде 2026-08-03, и единственным следом была строка в docker-логе, которая
    теряется при редеплое. Прогон при этом НЕ прерываем — cookie-инъекция это
    механизм обхода QRATOR, а не жёсткое требование (см. докстринг модуля), — но
    состояние требует ручного действия человека, значит должно быть событием.

    Причину различаем так же, как #2658 у Циана: «кук нет вовсе» и «протухли N дней
    назад» лечатся одинаково, но диагностируются по-разному.
    """
    expires_at = domclick_session_svc.session_expires_at(db)
    now = datetime.now(tz=UTC)
    if expires_at is None:
        detail = "кук DomClick нет в БД"
    elif expires_at <= now:
        detail = (
            f"куки DomClick протухли {expires_at:%Y-%m-%d} "
            f"({(now - expires_at).days} дн. назад)"
        )
    else:
        detail = "куки DomClick помечены невалидными (last_invalid_at)"
    logger.error(
        "domclick_detail_backfill: run_id=%d — %s; обогащение идёт БЕЗ cookie-инъекции "
        "(QRATOR-обход деградировал до organic SERP-origin навигации, PR #2430). "
        "Перезалейте сессию test-аккаунта: POST /scrape/domclick/upload-cookies",
        run_id,
        detail,
    )


def _warn_before_domclick_cookies_expire(db: Session, run_id: int) -> None:
    """Предупредить ЗАРАНЕЕ, пока куки ещё рабочие (#2674, образец — #2658 для Циана).

    Сигнал по факту протухания приходит, когда обогащение уже встало; обновление кук
    ручное, человеку нужен запас. valid_only=True — срок ИМЕННО той записи, которую
    взял load_session (при нескольких аккаунтах свежайшая-любая может быть чужой).
    """
    expires_at = domclick_session_svc.session_expires_at(db, valid_only=True)
    if expires_at is None:
        return
    left = expires_at - datetime.now(tz=UTC)
    if left <= timedelta(days=domclick_session_svc.COOKIE_EXPIRY_WARN_DAYS):
        logger.error(
            "domclick_detail_backfill: run_id=%d — куки DomClick протухнут %s "
            "(осталось %.1f дн.); обновите заранее, иначе обогащение деградирует молча",
            run_id,
            expires_at.date().isoformat(),
            left.total_seconds() / 86400,
        )


@dataclass
class DomClickDetailBackfillResult:
    """Counters for one backfill run."""

    attempted: int = 0
    enriched: int = 0
    blocked: int = 0
    failed: 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,
            "duration_sec": int(self.duration_sec),
        }


async def run_domclick_detail_backfill(
    db: Session,
    *,
    run_id: int,
    params: dict,
) -> DomClickDetailBackfillResult:
    """Backfill detail_enriched_at for domklik listings via cookie-injected browser fetch.

    Params (from default_params jsonb in scrape_schedules):
        batch_size: int -- snapshot size (SELECT LIMIT), default 200.
        budget_sec: float -- wall-clock budget per run, default 3600s.
        request_delay_sec: float -- delay between listings, default 12.0s.
        max_consecutive_blocks: int -- abort threshold, default 3.

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

    counters = DomClickDetailBackfillResult()
    current_counters: dict[str, int] = counters.to_dict()
    start = time.monotonic()

    logger.info("domclick_detail_backfill: run_id=%d starting", run_id)

    try:
        # Cookie injection (#2000 PR #2433) -- loaded ONCE per run, threaded into every
        # fetch_detail() call below. Прогон продолжается и без кук (см. докстринг), но
        # это состояние требует ЧЕЛОВЕКА: обновление сессии — ручная операция.
        cookies = domclick_session_svc.load_session(db)
        if cookies is None:
            _alert_domclick_cookies(db, run_id)
        else:
            _warn_before_domclick_cookies_expire(db, run_id)

        runs_mod.update_heartbeat(db, run_id, current_counters)

        # SNAPSHOT: single SELECT at start -- NOT re-selected in loop.
        # NAMING TRAP (see module docstring): source='domklik' here (data identifier) is
        # NOT the same string as BrowserFetcher(source="domclick") below (infra
        # identifier) -- both are correct, do not "fix" one to match the other.
        snapshot = (
            db.execute(
                text(
                    """
                    SELECT id, source_url
                    FROM listings
                    WHERE source = 'domklik'
                      AND detail_enriched_at IS NULL
                      AND source_url IS NOT NULL
                      AND is_active = TRUE
                    ORDER BY (lat IS NULL) DESC, scraped_at DESC NULLS LAST
                    LIMIT CAST(:batch_size AS int)
                    """
                ),
                {"batch_size": batch_size},
            )
            .mappings()
            .all()
        )

        if not snapshot:
            logger.info(
                "domclick_detail_backfill: run_id=%d -- no pending listings "
                "(detail_enriched_at IS NULL = 0), done",
                run_id,
            )
            runs_mod.mark_done(db, run_id, current_counters)
            return counters

        logger.info(
            "domclick_detail_backfill: run_id=%d snapshot=%d (budget=%.0fs "
            "delay=%.1fs max_blocks=%d cookies=%s)",
            run_id,
            len(snapshot),
            budget_sec,
            request_delay_sec,
            max_consecutive_blocks,
            "yes" if cookies is not None else "no",
        )

        consecutive_blocks = 0
        aborted_by_blocks = False
        do_sleep = False

        # Exactly ONE BrowserFetcher per run (no curl fallback for DomClick, see
        # module docstring). source="domclick" -- infra identifier, dedicated
        # residential proxy (scrape_proxies.provider_affinity='domclick',
        # 173_scrape_proxies_add_domclick_affinity.sql).
        async with BrowserFetcher(source="domclick", endpoint=settings.browser_http_endpoint) as bf:
            for idx, row in enumerate(snapshot):
                # Budget guard
                elapsed = time.monotonic() - start
                if elapsed > budget_sec:
                    logger.info(
                        "domclick_detail_backfill: run_id=%d -- budget %.0fs exhausted "
                        "(elapsed=%.1fs), stopping at #%d/%d",
                        run_id,
                        budget_sec,
                        elapsed,
                        idx,
                        len(snapshot),
                    )
                    break

                # #1182 Phase 2: кооперативный SIGTERM-drain (деплой recreate scraper).
                if shutdown_requested():
                    logger.info(
                        "domclick_detail_backfill: run_id=%d SIGTERM-drain — stopping at " "#%d/%d",
                        run_id,
                        idx,
                        len(snapshot),
                    )
                    break

                # Jittered delay (±30%) -- organic pacing, matches the request_delay_sec
                # spirit of PR #2430's same-site navigation (less robotic cadence).
                if do_sleep:
                    await asyncio.sleep(request_delay_sec * random.uniform(0.7, 1.4))
                do_sleep = True

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

                try:
                    enrichment = await fetch_detail(source_url, browser_fetcher=bf, cookies=cookies)
                    if save_detail_enrichment(db, listing_id, enrichment):
                        counters.enriched += 1
                    consecutive_blocks = 0

                except DomClickParseError as e:
                    # Schema drift, not a block -- neutral to the block-breaker (does
                    # NOT touch consecutive_blocks, does NOT abort the run).
                    counters.failed += 1
                    logger.warning(
                        "domclick_detail_backfill: run_id=%d listing %s PARSE-ERROR "
                        "(schema drift, not a block): %s",
                        run_id,
                        source_url,
                        e,
                    )

                except DomClickBlockedError as e:
                    consecutive_blocks += 1
                    counters.blocked += 1
                    logger.warning(
                        "domclick_detail_backfill: run_id=%d BLOCKED #%d/%d "
                        "(consecutive=%d): %s",
                        run_id,
                        idx + 1,
                        len(snapshot),
                        consecutive_blocks,
                        e,
                    )
                    if consecutive_blocks >= max_consecutive_blocks:
                        logger.error(
                            "domclick_detail_backfill: run_id=%d ABORT -- %d consecutive "
                            "blocks, QRATOR reputation likely burned for the session/proxy. "
                            "enriched=%d attempted=%d",
                            run_id,
                            consecutive_blocks,
                            counters.enriched,
                            counters.attempted,
                        )
                        aborted_by_blocks = True
                        break

                except Exception as e:
                    counters.failed += 1
                    logger.warning(
                        "domclick_detail_backfill: run_id=%d listing %s failed: %s",
                        run_id,
                        source_url,
                        e,
                    )
                    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="domclick_detail_backfill",
            aborted_by_blocks=aborted_by_blocks,
        )
        logger.info(
            "domclick_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(
            "domclick_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