gendesign/backend/app/workers/tasks/reservation_ingest.py

"""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):
  • SessionLocal() + try/finally close.
  • 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 хитов).
"""

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(db: Session, records_meta: list[dict[str, Any]]) -> int:
    """UPSERT записей в land_reservation. Возвращает число обработанных строк."""
    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 _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 с ручным вызовом)."""
    total_parcels = 0
    db: Session = SessionLocal()
    try:
        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
            ]
            count = _upsert_records(db, rows)
            db.commit()
            total_parcels += count

        logger.info(
            "ingest_reservations(explicit): готово docs=%d parcels=%d",
            len(docs),
            total_parcels,
        )
    finally:
        db.close()

    return {"docs": len(docs), "parcels": total_parcels}


def _ingest_enumerated_docs(enumerated: list[dict[str, Any]]) -> dict[str, int]:
    """Ингестирует документы, найденные через enumerate с pravo.gov66."""
    total_parcels = 0
    fetched_docs = 0
    db: Session = SessionLocal()
    try:
        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
            ]
            count = _upsert_records(db, rows)
            db.commit()
            total_parcels += count

        logger.info(
            "ingest_reservations(enum): готово fetched=%d parcels=%d",
            fetched_docs,
            total_parcels,
        )
    finally:
        db.close()

    return {"docs": fetched_docs, "parcels": total_parcels}


__all__ = ["ingest_reservations"]