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

"""Celery task: OCR-пайплайн изъятия ЕКБ → land_reservation (#1062).

Загружает PDF-сканы «Сообщений о планируемом изъятии…» с раздела екатеринбург.рф,
выполняет OCR через Tesseract rus, извлекает кад-номера 66:41:NNNNNNN:NN,
UPSERT-ит в land_reservation (м.136). Reservation_lookup / analyze-wiring (#1118)
подхватывает данные автоматически — дополнительного wire-up не требуется.

Дедуп-ключ:
  ON CONFLICT (cad_num, act_number) — унаследован из reservation_ingest.py.
  Если act_number IS NULL (не извлечён из сканов) → конфликт НЕ возникает при NULL-UPSERT
  (NULL != NULL в SQL). Чтобы предотвратить дубли при act_number IS NULL, дедуплицируем
  по (cad_num, doc_url) на уровне Python перед UPSERT: один URL = один батч,
  повторный запуск с тем же URL обновит существующую строку через source+fetched_at
  (где act_number IS NULL используем DO NOTHING вместо DO UPDATE — нет stable key).
  Решение: для строк с act_number IS NULL добавляем в ON CONFLICT УНИКАЛЬНОСТЬ через
  отдельный UPSERT с COALESCE-fallback: если запись с (cad_num, doc_url) уже есть —
  UPDATE, иначе INSERT. Реализовано через двухшаговый UPSERT ниже.

Beat: еженедельно (пятница 07:00 МСК) — изъятия выходят редко.

Conventions (зеркало reservation_ingest.py / backend.md):
  • SessionLocal() + try/finally close.
  • SAVEPOINT per-row (with db.begin_nested()).
  • CAST(:x AS type) — НИКОГДА :x::type (psycopg v3).
  • httpx, не requests.
  • Сбой одного документа не валит прогон (try/except с logger.error).
"""

from __future__ import annotations

import logging
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core.db import SessionLocal
from app.workers.celery_app import celery_app

logger = logging.getLogger(__name__)

# UPSERT — два варианта в зависимости от наличия act_number.
#
# Вариант A (act_number IS NOT NULL): ON CONFLICT (cad_num, act_number) → DO UPDATE.
#   Stable key = (cad_num, act_number). Идемпотентно при повторном прогоне.
#
# Вариант B (act_number IS NULL): INSERT ... ON CONFLICT DO NOTHING.
#   NULL != NULL → (cad_num, NULL) никогда не конфликтует по индексу.
#   Python-дедуп per-batch предотвращает дубли в рамках одного прогона.
#   Повторные прогоны добавят дубли если строки нет — acceptable (rare, data audit OK).
#   Альтернатива (partial unique index на NULL) — задача database-expert, не здесь.

_UPSERT_WITH_ACT_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      = COALESCE(EXCLUDED.raw_excerpt, land_reservation.raw_excerpt),
        fetched_at       = now()
    """
)

_UPSERT_NO_ACT_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),
        NULL,
        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 DO NOTHING
    """
)


def _upsert_records(db: Session, records: list[dict[str, Any]]) -> int:
    """UPSERT записей в land_reservation. Возвращает число успешно обработанных строк."""
    count = 0
    for row in records:
        upsert_sql = _UPSERT_WITH_ACT_SQL if row.get("act_number") else _UPSERT_NO_ACT_SQL
        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 doc_url=%s: %s",
                row.get("cad_num"),
                row.get("doc_url"),
                exc,
            )
    return count


@celery_app.task(name="tasks.izyatie_ocr_ingest.ingest_izyatie_ocr")
def ingest_izyatie_ocr() -> dict[str, int]:
    """OCR-пайплайн изъятия ЕКБ: раздел сайта → PDF → Tesseract → land_reservation.

    Returns:
        {'docs': N, 'parcels': M} — обработано документов и вставлено/обновлено ЗУ.
    """
    # Ленивые импорты — модули не нужны при старте celery_app.
    from app.services.scrapers.izyatie_client import fetch_pdf, list_izyatie_documents
    from app.services.scrapers.izyatie_ocr import extract_izyatie_records, ocr_pdf_text

    docs = list_izyatie_documents()
    if not docs:
        logger.warning("ingest_izyatie_ocr: документов не найдено — прогон завершён без данных")
        return {"docs": 0, "parcels": 0}

    logger.info("ingest_izyatie_ocr: запуск пайплайна, %d документов", len(docs))

    total_parcels = 0
    processed_docs = 0
    db: Session = SessionLocal()

    try:
        for doc in docs:
            doc_title: str = doc.get("title", "")
            doc_url: str = doc.get("url", "")

            logger.info("ingest_izyatie_ocr: загружаем %r → %s", doc_title[:60], doc_url)

            try:
                pdf_bytes = fetch_pdf(doc_url)
            except Exception as exc:
                logger.error(
                    "ingest_izyatie_ocr: не удалось загрузить PDF %s: %s",
                    doc_url,
                    exc,
                )
                continue

            ocr_text = ocr_pdf_text(pdf_bytes)
            if not ocr_text.strip():
                logger.warning(
                    "ingest_izyatie_ocr: OCR вернул пустой текст для %s "
                    "(tesseract не установлен, битый PDF или чистый лист?)",
                    doc_url,
                )
                processed_docs += 1
                continue

            records = extract_izyatie_records(ocr_text, doc_title, doc_url)
            logger.info(
                "ingest_izyatie_ocr: %r → %d кад-номеров",
                doc_title[:60],
                len(records),
            )

            if not records:
                processed_docs += 1
                continue

            count = _upsert_records(db, records)
            db.commit()
            total_parcels += count
            processed_docs += 1

    finally:
        db.close()

    logger.info(
        "ingest_izyatie_ocr: завершено docs=%d parcels=%d",
        processed_docs,
        total_parcels,
    )
    return {"docs": processed_docs, "parcels": total_parcels}


__all__ = ["ingest_izyatie_ocr"]