"""Celery task: ingest постановлений об изъятии/резервировании ЗУ → ``land_reservation`` (#1062).

#1062 (ИРД-5), эпик #1067 (GG-форсайт). Загружает PDF-постановления с pravo.gov66.ru
через pravo_gov66_client, извлекает кад-номера через page_reservation_parser,
UPSERT-ит в land_reservation (м.136).

Геометрии зон здесь НЕТ — join к cad_parcels.geom на read-side (reservation_lookup.py).

Beat НЕ настроен: постановления выходят редко, запуск ручной/по требованию.
  celery call tasks.reservation_ingest.ingest_reservations

Conventions (зеркало ird_harvest.py / backend.md):
  • Session lifecycle (#2445 D4): PDF fetch происходит БЕЗ открытой сессии — pravo.gov66
    иногда медленный (таймаут до 60с), держать connection-pool slot на всё время внешнего
    запроса нельзя. Короткоживущий ``SessionLocal()`` открывается ТОЛЬКО на запись строк
    одного документа (open → upsert → commit → close per item), не на весь цикл.
  • SAVEPOINT per-row (``with db.begin_nested():``) — битая строка не валит батч.
  • CAST(:x AS type) — НИКОГДА :x::text (psycopg v3).
  • ON CONFLICT (cad_num, act_number) → обновляет метаданные + fetched_at.
  • httpx.Client (не requests). Сбой одного дока не валит прогон.
  • q-запросы — 2-3 слова (движок pravo.gov66 строгий AND/фраза, 1 или 4+ → 0 хитов).
  • Pacing/backoff к pravo.gov66.ru — внутри pravo_gov66_client (_get_with_backoff),
    НЕ здесь: каждый search/page/pdf запрос уже пейсится на уровне клиента.
"""

from __future__ import annotations

import logging
import re
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.services.scrapers.page_reservation_parser import extract_from_pdf
from app.services.scrapers.pravo_gov66_client import (
    fetch_pdf,
    get_document_pdf_url,
    search_documents,
)
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)

# ── Запросы для enumerate ──────────────────────────────────────────────────────
# 2-3 слова: движок pravo.gov66 строгий AND/фраза; 1 или 4+ → 0 результатов.
_RESERVATION_QUERIES: list[str] = [
    "изъятие земельных участков",
    "резервирование земельных участков",
]

# Фильтр заголовков: документ должен содержать изъят|резерв И Екатеринбург|Свердлов.
_RE_RELEVANCE = re.compile(r"изъят|резерв", re.IGNORECASE | re.UNICODE)
_RE_GEO = re.compile(r"Екатеринбург|Свердлов", re.IGNORECASE | re.UNICODE)

# UPSERT: ON CONFLICT (cad_num, act_number) → обновляем все поля кроме id.
# CAST(:x AS type) — psycopg v3. DO NOTHING fallback если act_number IS NULL (NULL != NULL → нет
# конфликта, вставляется дубль — acceptable для NULL-act_number edge-case).
_UPSERT_SQL = text(
    """
    INSERT INTO land_reservation (
        cad_num, reservation_kind, basis_act, act_number, act_date,
        purpose, doc_url, source, is_active, raw_excerpt, fetched_at
    ) VALUES (
        CAST(:cad_num AS text),
        CAST(:reservation_kind AS text),
        CAST(:basis_act AS text),
        CAST(:act_number AS text),
        CAST(:act_date AS date),
        CAST(:purpose AS text),
        CAST(:doc_url AS text),
        CAST(:source AS text),
        true,
        CAST(:raw_excerpt AS text),
        now()
    )
    ON CONFLICT (cad_num, act_number) DO UPDATE SET
        reservation_kind = EXCLUDED.reservation_kind,
        basis_act        = EXCLUDED.basis_act,
        act_date         = EXCLUDED.act_date,
        purpose          = EXCLUDED.purpose,
        doc_url          = EXCLUDED.doc_url,
        source           = EXCLUDED.source,
        raw_excerpt      = EXCLUDED.raw_excerpt,
        fetched_at       = now()
    """
)


def _is_relevant_title(title: str) -> bool:
    """Проверяет, что заголовок документа касается изъятия/резервирования ЗУ в регионе."""
    return bool(_RE_RELEVANCE.search(title) and _RE_GEO.search(title))


def _fetch_pdf_safe(pdf_url: str) -> bytes | None:
    """Скачивает PDF по URL, graceful при ошибке."""
    try:
        return fetch_pdf(pdf_url)
    except Exception as exc:
        logger.error("_fetch_pdf_safe: не удалось загрузить %s: %s", pdf_url, exc)
        return None


def _upsert_records_in_session(db: Session, records_meta: list[dict[str, Any]]) -> int:
    """UPSERT записей в land_reservation на уже открытой сессии.

    Возвращает число upsert'нутых строк.
    """
    count = 0
    for row in records_meta:
        try:
            with db.begin_nested():  # SAVEPOINT per-row
                db.execute(_UPSERT_SQL, row)
            count += 1
        except Exception as exc:
            logger.warning(
                "_upsert_records: пропуск cad_num=%s: %s",
                row.get("cad_num"),
                exc,
            )
    return count


def _upsert_records(records_meta: list[dict[str, Any]]) -> int:
    """Открывает короткоживущую сессию ТОЛЬКО на запись одного документа → commit → close.

    Session lifecycle (#2445 D4): вызывается ПОСЛЕ внешнего PDF-fetch, никогда во время
    него — держать открытую транзакцию/connection-pool slot на время сетевого запроса
    к pravo.gov66.ru (таймаут до 60с) недопустимо (см. WAF-инцидент #2443).
    """
    if not records_meta:
        return 0
    db: Session = SessionLocal()
    try:
        count = _upsert_records_in_session(db, records_meta)
        db.commit()
        return count
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()


def _enumerate_documents(since: str | None = None) -> list[dict[str, Any]]:
    """Обходит все поисковые запросы и возвращает дедуплицированный список документов.

    Args:
        since: дата начала периода «DD.MM.YYYY» для инкрементального обхода.

    Returns:
        Список {'doc_id', 'title', 'url'}, дедуплицированный по doc_id.
    """
    seen_ids: set[str] = set()
    docs: list[dict[str, Any]] = []

    for query in _RESERVATION_QUERIES:
        hits = search_documents(query, since=since)
        logger.info(
            "_enumerate_documents: query=%r → %d хитов (до фильтрации)",
            query,
            len(hits),
        )
        for hit in hits:
            doc_id: str = hit["doc_id"]
            title: str = hit.get("title", "")
            if doc_id in seen_ids:
                continue
            if not _is_relevant_title(title):
                logger.debug(
                    "_enumerate_documents: пропуск doc_id=%s (нерелевантный заголовок): %r",
                    doc_id,
                    title,
                )
                continue
            seen_ids.add(doc_id)
            docs.append(hit)

    logger.info("_enumerate_documents: итого %d релевантных документов", len(docs))
    return docs


@celery_app.task(name="tasks.reservation_ingest.ingest_reservations")
def ingest_reservations(
    docs: list[dict[str, Any]] | None = None,
    since: str | None = None,
) -> dict[str, int]:
    """Загружает и ингестирует постановления об изъятии/резервировании ЗУ.

    Args:
        docs: список документов для обработки. Каждый словарь должен содержать:
            url (str) — прямая ссылка на PDF или страницу документа,
            act_number (str | None) — номер акта (опц.),
            basis_act (str) — заголовок/описание акта,
            purpose (str | None) — цель (опц.).
            Если None — enumerate с pravo.gov66 по _RESERVATION_QUERIES.
        since: дата начала периода «DD.MM.YYYY» для инкрементального обхода.
            Игнорируется если docs задан явно.

    Returns:
        {'docs': N_fetched, 'parcels': M_upserted}.
    """
    if docs is not None:
        # Ручной override: docs передан явно, enumerate не делаем.
        return _ingest_explicit_docs(docs)

    # Enumerate с pravo.gov66.
    enumerated = _enumerate_documents(since=since)
    return _ingest_enumerated_docs(enumerated)


def _ingest_explicit_docs(docs: list[dict[str, Any]]) -> dict[str, int]:
    """Ингестирует явно переданный список документов (back-compat с ручным вызовом).

    Session lifecycle (#2445 D4): PDF fetch (сетевой, может занять до 60с) выполняется
    БЕЗ открытой DB-сессии; ``_upsert_records`` открывает свою короткоживущую сессию
    только на запись строк одного документа (commit+close сразу после).
    """
    total_parcels = 0
    for doc in docs:
        url = doc.get("url", "")
        act_number = doc.get("act_number")
        basis_act = doc.get("basis_act", "")
        doc_purpose = doc.get("purpose")
        doc_kind = doc.get("kind", "изъятие")

        logger.info("ingest_reservations(explicit): загрузка url=%s act=%s", url, act_number)

        pdf_bytes = _fetch_pdf_safe(url)  # без открытой сессии — только сетевой запрос
        if pdf_bytes is None:
            logger.warning("ingest_reservations(explicit): пропуск url=%s (не загружен)", url)
            continue

        records = extract_from_pdf(pdf_bytes)
        logger.info("ingest_reservations(explicit): url=%s → %d КН", url, len(records))
        if not records:
            continue

        rows = [
            {
                "cad_num": rec.cad_num,
                "reservation_kind": rec.reservation_kind or doc_kind,
                "basis_act": basis_act,
                "act_number": rec.act_number or act_number,
                "act_date": rec.act_date,
                "purpose": rec.purpose or doc_purpose,
                "doc_url": url,
                "source": "page_pdf",
                "raw_excerpt": rec.raw_excerpt,
            }
            for rec in records
        ]
        total_parcels += _upsert_records(rows)  # short-lived session: open→upsert→commit→close

    logger.info(
        "ingest_reservations(explicit): готово docs=%d parcels=%d",
        len(docs),
        total_parcels,
    )
    return {"docs": len(docs), "parcels": total_parcels}


def _ingest_enumerated_docs(enumerated: list[dict[str, Any]]) -> dict[str, int]:
    """Ингестирует документы, найденные через enumerate с pravo.gov66.

    Session lifecycle (#2445 D4): оба сетевых вызова на документ (get_document_pdf_url,
    fetch_pdf) выполняются БЕЗ открытой DB-сессии; ``_upsert_records`` открывает свою
    короткоживущую сессию только на запись строк одного документа.
    """
    total_parcels = 0
    fetched_docs = 0
    for hit in enumerated:
        doc_id: str = hit["doc_id"]
        title: str = hit.get("title", "")
        doc_page_url: str = hit["url"]

        logger.info("ingest_reservations(enum): doc_id=%s title=%r", doc_id, title)

        pdf_url = get_document_pdf_url(doc_id)  # сеть, без открытой сессии
        if pdf_url is None:
            logger.warning(
                "ingest_reservations(enum): doc_id=%s — PDF не найден на странице", doc_id
            )
            continue

        pdf_bytes = _fetch_pdf_safe(pdf_url)  # сеть, без открытой сессии
        if pdf_bytes is None:
            logger.warning("ingest_reservations(enum): doc_id=%s — не удалось скачать PDF", doc_id)
            continue

        fetched_docs += 1
        records = extract_from_pdf(pdf_bytes)
        logger.info("ingest_reservations(enum): doc_id=%s → %d КН", doc_id, len(records))
        if not records:
            continue

        rows = [
            {
                "cad_num": rec.cad_num,
                "reservation_kind": rec.reservation_kind,
                "basis_act": title,
                "act_number": rec.act_number,
                "act_date": rec.act_date,
                "purpose": rec.purpose,
                "doc_url": doc_page_url,
                "source": "pravo_gov66",
                "raw_excerpt": rec.raw_excerpt,
            }
            for rec in records
        ]
        total_parcels += _upsert_records(rows)  # short-lived session: open→upsert→commit→close

    logger.info(
        "ingest_reservations(enum): готово fetched=%d parcels=%d",
        fetched_docs,
        total_parcels,
    )
    return {"docs": fetched_docs, "parcels": total_parcels}


__all__ = ["ingest_reservations"]