"""Celery task: ingest постановлений об изъятии/резервировании ЗУ → ``land_reservation`` (#1091). #1062 (ИРД-5), эпик #1067 (GG-форсайт). Загружает PDF-постановления (pravo.gov66.ru / ПАГЕ), извлекает кад-номера через 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 text) — НИКОГДА :x::text (psycopg v3). • ON CONFLICT (cad_num, act_number) → обновляет метаданные + fetched_at. • httpx.Client (не requests). Сбой одного дока не валит прогон. """ from __future__ import annotations import logging from typing import Any import httpx 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.workers.celery_app import celery_app logger = logging.getLogger(__name__) # ── Seed-документы ───────────────────────────────────────────────────────────── # Единственный известный seed (2026-06): Постановление Правительства Свердловской обл. # № 509-ПП об изъятии ЗУ под реконструкцию ул. Татищева, ЕКБ. # VERIFY URL: точный PDF-URL нужно уточнить на pravo.gov66.ru — документ может лежать под # разными путями (поиск по «509-ПП» или номеру в реестре). Placeholder установлен намеренно. _SEED_DOCS: list[dict[str, Any]] = [ { "url": "https://pravo.gov66.ru/documents/view/509-pp", # VERIFY URL — placeholder "kind": "изъятие", "act_number": "509-ПП", "basis_act": "Постановление Правительства Свердловской области № 509-ПП", "purpose": "реконструкция улицы Татищева, г. Екатеринбург", }, ] # 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 _fetch_pdf(url: str) -> bytes | None: """Загружает PDF по URL. Возвращает None при ошибке (сбой одного дока не валит прогон). Примечание по TLS: pravo.gov66.ru использует российский УЦ, verify=False. В prod-окружении следует настроить доверенный CA-bundle (аналогично ekb_geoportal_client.py). """ try: with httpx.Client(timeout=60.0, verify=False, follow_redirects=True) as client: resp = client.get(url) resp.raise_for_status() return resp.content except Exception as exc: logger.error("_fetch_pdf: не удалось загрузить %s: %s", 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 @celery_app.task(name="tasks.reservation_ingest.ingest_reservations") def ingest_reservations(docs: list[dict[str, Any]] | None = None) -> dict[str, int]: """Загружает и ингестирует постановления об изъятии/резервировании ЗУ. Args: docs: список документов для обработки. Каждый словарь должен содержать поля: url (str), kind (str), act_number (str), basis_act (str), purpose (str | None). Если None — используется _SEED_DOCS. Returns: {'docs': N, 'parcels': M} — количество обработанных документов и вставленных/ обновлённых записей ЗУ. """ target_docs = docs if docs is not None else _SEED_DOCS total_parcels = 0 db: Session = SessionLocal() try: for doc in target_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: загрузка документа url=%s act=%s", url, act_number) pdf_bytes = _fetch_pdf(url) if pdf_bytes is None: logger.warning("ingest_reservations: пропуск документа url=%s (не загружен)", url) continue records = extract_from_pdf(pdf_bytes) logger.info( "ingest_reservations: url=%s → извлечено %d КН", url, len(records), ) if not records: continue rows: list[dict[str, Any]] = [] for rec in records: rows.append( { "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, } ) count = _upsert_records(db, rows) db.commit() total_parcels += count logger.info( "ingest_reservations: готово docs=%d parcels=%d", len(target_docs), total_parcels, ) finally: db.close() return {"docs": len(target_docs), "parcels": total_parcels} __all__ = ["ingest_reservations"]