backend/app/services/scrapers/pravo_gov66_client.py

@@ -0,0 +1,167 @@
+"""Клиент для поиска и скачивания постановлений с pravo.gov66.ru (#1062).
+
+Сайт pravo.gov66.ru — реестр правовых актов Правительства Свердловской области.
+HTML server-rendered (нет JS-рендеринга), httpx достаточно.
+
+TLS: сайт использует российский УЦ → verify=False (аналогично ekburg_permits #242).
+
+Публичный API:
+    search_documents(query, law_type, since) → list[dict]
+    get_document_pdf_url(doc_id) → str | None
+    fetch_pdf(url) → bytes
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import Any
+
+import httpx
+from bs4 import BeautifulSoup
+
+logger = logging.getLogger(__name__)
+
+_BASE_URL = "https://www.pravo.gov66.ru"
+_SEARCH_URL = f"{_BASE_URL}/search/"
+
+# Тайм-аут щедрый: сервер иногда медленный.
+_TIMEOUT = httpx.Timeout(60.0, connect=15.0)
+
+# Паттерны для извлечения ссылок.
+# Числовой doc_id в пути: /35496/
+_RE_DOC_PATH = re.compile(r"^/(\d+)/$")
+# PDF в /media/pravo/ (не ecp-вариант: ecp — подписанная копия, не всегда парсится).
+_RE_PDF_PATH = re.compile(r"/media/pravo/(?!ecp/)[^\s\"']+\.pdf", re.IGNORECASE)
+# ecp-вариант как fallback.
+_RE_PDF_ECP_PATH = re.compile(r"/media/pravo/ecp/[^\s\"']+\.pdf", re.IGNORECASE)
+
+
+def search_documents(
+    query: str,
+    law_type: str = "Постановление",
+    since: str | None = None,
+) -> list[dict[str, Any]]:
+    """Поиск документов на pravo.gov66.ru.
+
+    Args:
+        query: поисковый запрос (2-3 слова; 1 или 4+ слов → 0 результатов на сервере).
+        law_type: тип документа, напр. «Постановление».
+        since: дата начала периода в формате «DD.MM.YYYY» (поле signed_start_date).
+
+    Returns:
+        Список словарей {'doc_id': str, 'title': str, 'url': str}.
+        Пустой список при любой сетевой ошибке.
+    """
+    params: dict[str, str] = {"q": query, "law_type": law_type}
+    if since:
+        params["signed_start_date"] = since
+
+    try:
+        with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
+            resp = client.get(_SEARCH_URL, params=params)
+            resp.raise_for_status()
+            html = resp.text
+    except Exception as exc:
+        logger.error("pravo_gov66_client.search_documents: ошибка сети q=%r: %s", query, exc)
+        return []
+
+    return _parse_search_html(html)
+
+
+def _parse_search_html(html: str) -> list[dict[str, Any]]:
+    """Извлекает doc_id / title / url из HTML страницы поиска."""
+    soup = BeautifulSoup(html, "html.parser")
+    results: list[dict[str, Any]] = []
+    seen_ids: set[str] = set()
+
+    for a_tag in soup.find_all("a", href=True):
+        href: str = a_tag["href"]
+        m = _RE_DOC_PATH.match(href)
+        if not m:
+            continue
+        doc_id = m.group(1)
+        if doc_id in seen_ids:
+            continue
+        seen_ids.add(doc_id)
+        title = a_tag.get_text(separator=" ", strip=True)
+        results.append(
+            {
+                "doc_id": doc_id,
+                "title": title,
+                "url": f"{_BASE_URL}{href}",
+            }
+        )
+
+    logger.debug("_parse_search_html: найдено %d документов", len(results))
+    return results
+
+
+def get_document_pdf_url(doc_id: str) -> str | None:
+    """Получает прямую ссылку на PDF со страницы документа.
+
+    Предпочитает /media/pravo/ (не ecp); если не найдена — берёт ecp-вариант.
+
+    Args:
+        doc_id: числовой идентификатор документа (из пути /{doc_id}/).
+
+    Returns:
+        Абсолютный URL PDF или None если не найдена / сетевая ошибка.
+    """
+    doc_url = f"{_BASE_URL}/{doc_id}/"
+    try:
+        with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
+            resp = client.get(doc_url)
+            resp.raise_for_status()
+            html = resp.text
+    except Exception as exc:
+        logger.error(
+            "pravo_gov66_client.get_document_pdf_url: ошибка загрузки doc_id=%s: %s",
+            doc_id,
+            exc,
+        )
+        return None
+
+    return _extract_pdf_url(html)
+
+
+def _extract_pdf_url(html: str) -> str | None:
+    """Извлекает URL PDF из HTML страницы документа."""
+    # Приоритет: /media/pravo/ (не ecp).
+    m = _RE_PDF_PATH.search(html)
+    if m:
+        path = m.group(0)
+        return path if path.startswith("http") else f"{_BASE_URL}{path}"
+
+    # Fallback: ecp-вариант.
+    m2 = _RE_PDF_ECP_PATH.search(html)
+    if m2:
+        path2 = m2.group(0)
+        return path2 if path2.startswith("http") else f"{_BASE_URL}{path2}"
+
+    return None
+
+
+def fetch_pdf(url: str) -> bytes:
+    """Загружает PDF по URL.
+
+    Args:
+        url: абсолютный URL PDF-файла.
+
+    Returns:
+        Байты PDF.
+
+    Raises:
+        httpx.HTTPError: при сетевой ошибке (caller должен обработать gracefully).
+    """
+    with httpx.Client(timeout=_TIMEOUT, verify=False, follow_redirects=True) as client:
+        resp = client.get(url)
+        resp.raise_for_status()
+        return resp.content
+
+
+__all__ = [
+    "fetch_pdf",
+    "get_document_pdf_url",
+    "search_documents",
+]

backend/app/workers/tasks/reservation_ingest.py

@@ -1,7 +1,8 @@
-"""Celery task: ingest постановлений об изъятии/резервировании ЗУ → ``land_reservation`` (#1091).
+"""Celery task: ingest постановлений об изъятии/резервировании ЗУ → ``land_reservation`` (#1062).
 
-#1062 (ИРД-5), эпик #1067 (GG-форсайт). Загружает PDF-постановления (pravo.gov66.ru / ПАГЕ),
-извлекает кад-номера через page_reservation_parser, UPSERT-ит в land_reservation (м.136).
+#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).
 
@@ -11,41 +12,43 @@ Beat НЕ настроен: постановления выходят редко
 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).
+  • 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
 
-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.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__)
 
-# ── 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": "реконструкция улицы Татищева, г. Екатеринбург",
-    },
+# ── Запросы для 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).
@@ -80,19 +83,17 @@ _UPSERT_SQL = text(
 )
 
 
-def _fetch_pdf(url: str) -> bytes | None:
-    """Загружает PDF по URL. Возвращает None при ошибке (сбой одного дока не валит прогон).
+def _is_relevant_title(title: str) -> bool:
+    """Проверяет, что заголовок документа касается изъятия/резервирования ЗУ в регионе."""
+    return bool(_RE_RELEVANCE.search(title) and _RE_GEO.search(title))
 
-    Примечание по TLS: pravo.gov66.ru использует российский УЦ, verify=False. В prod-окружении
-    следует настроить доверенный CA-bundle (аналогично ekb_geoportal_client.py).
-    """
+
+def _fetch_pdf_safe(pdf_url: str) -> bytes | None:
+    """Скачивает PDF по URL, graceful при ошибке."""
     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
+        return fetch_pdf(pdf_url)
     except Exception as exc:
-        logger.error("_fetch_pdf: не удалось загрузить %s: %s", url, exc)
+        logger.error("_fetch_pdf_safe: не удалось загрузить %s: %s", pdf_url, exc)
         return None
 
 
@@ -113,76 +114,186 @@ def _upsert_records(db: Session, records_meta: list[dict[str, Any]]) -> int:
     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) -> dict[str, int]:
+def ingest_reservations(
+    docs: list[dict[str, Any]] | None = None,
+    since: str | None = None,
+) -> dict[str, int]:
     """Загружает и ингестирует постановления об изъятии/резервировании ЗУ.
 
     Args:
-        docs: список документов для обработки. Каждый словарь должен содержать поля:
-            url (str), kind (str), act_number (str), basis_act (str), purpose (str | None).
-            Если None — используется _SEED_DOCS.
+        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, 'parcels': M} — количество обработанных документов и вставленных/
-        обновлённых записей ЗУ.
+        {'docs': N_fetched, 'parcels': M_upserted}.
     """
-    target_docs = docs if docs is not None else _SEED_DOCS
+    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 target_docs:
+        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: загрузка документа url=%s act=%s", url, act_number)
+            logger.info("ingest_reservations(explicit): загрузка url=%s act=%s", url, act_number)
 
-            pdf_bytes = _fetch_pdf(url)
+            pdf_bytes = _fetch_pdf_safe(url)
             if pdf_bytes is None:
-                logger.warning("ingest_reservations: пропуск документа url=%s (не загружен)", url)
+                logger.warning("ingest_reservations(explicit): пропуск url=%s (не загружен)", url)
                 continue
 
             records = extract_from_pdf(pdf_bytes)
-            logger.info(
-                "ingest_reservations: url=%s → извлечено %d КН",
-                url,
-                len(records),
-            )
-
+            logger.info("ingest_reservations(explicit): 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,
-                    }
-                )
-
+            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: готово docs=%d parcels=%d",
-            len(target_docs),
+            "ingest_reservations(explicit): готово docs=%d parcels=%d",
+            len(docs),
             total_parcels,
         )
     finally:
         db.close()
 
-    return {"docs": len(target_docs), "parcels": total_parcels}
+    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"]

backend/tests/scrapers/test_pravo_gov66_client.py

@@ -0,0 +1,163 @@
+"""Тесты pravo_gov66_client — offline, без живой сети.
+
+Покрывают: парсинг search-HTML (doc_id/title/url), парсинг страницы документа
+(PDF-ссылка), fallback на ecp-вариант, отсутствие PDF.
+"""
+
+from __future__ import annotations
+
+from app.services.scrapers.pravo_gov66_client import (
+    _extract_pdf_url,
+    _parse_search_html,
+)
+
+# ── Фикстура HTML страницы поиска ─────────────────────────────────────────────
+# Реальный фрагмент страницы https://www.pravo.gov66.ru/search/?q=изъятие+земельных+участков
+# Содержит ссылки вида /NNNNN/ с заголовком постановления.
+_SEARCH_HTML_FIXTURE = """
+<!DOCTYPE html>
+<html>
+<head><title>Поиск</title></head>
+<body>
+<div class="search-results">
+  <ul>
+    <li class="result-item">
+      <a href="/35496/">
+        Об изъятии земельных участков для государственных нужд
+        Свердловской области в целях реконструкции ул. Татищева
+        в г. Екатеринбурге
+      </a>
+    </li>
+    <li class="result-item">
+      <a href="/34812/">
+        О резервировании земельных участков для нужд Екатеринбург
+      </a>
+    </li>
+    <li class="result-item">
+      <!-- Ссылка на категорию, не документ — должна игнорироваться -->
+      <a href="/category/law/">Категория</a>
+    </li>
+    <li class="result-item">
+      <!-- Дубль doc_id 35496 — должен дедуплицироваться -->
+      <a href="/35496/">Дубль</a>
+    </li>
+    <li class="result-item">
+      <!-- Нечисловой id — не должен попасть -->
+      <a href="/some-slug/">Текстовый slug</a>
+    </li>
+  </ul>
+</div>
+</body>
+</html>
+"""
+
+# ── Фикстура HTML страницы документа с обычным PDF ────────────────────────────
+_DOC_HTML_WITH_PDF = """
+<!DOCTYPE html>
+<html>
+<body>
+  <h1>Постановление № 509-ПП</h1>
+  <p>
+    Скачать документ:
+    <a href="/media/pravo/509-ПП_ab1234cd.pdf">Скачать PDF</a>
+  </p>
+  <p>
+    Подписанная версия:
+    <a href="/media/pravo/ecp/509-ПП.pdf">Скачать подписанный PDF</a>
+  </p>
+</body>
+</html>
+"""
+
+# ── Фикстура HTML страницы документа только с ecp PDF ─────────────────────────
+_DOC_HTML_ECP_ONLY = """
+<!DOCTYPE html>
+<html>
+<body>
+  <h1>Постановление № 312-ПП</h1>
+  <p>
+    <a href="/media/pravo/ecp/312-ПП.pdf">Подписанный PDF</a>
+  </p>
+</body>
+</html>
+"""
+
+# ── Фикстура HTML страницы документа без PDF ──────────────────────────────────
+_DOC_HTML_NO_PDF = """
+<!DOCTYPE html>
+<html>
+<body>
+  <h1>Постановление без PDF</h1>
+  <p>Текст постановления размещён ниже.</p>
+</body>
+</html>
+"""
+
+
+class TestParseSearchHtml:
+    def test_extracts_doc_ids_and_titles(self) -> None:
+        results = _parse_search_html(_SEARCH_HTML_FIXTURE)
+        ids = [r["doc_id"] for r in results]
+        assert "35496" in ids
+        assert "34812" in ids
+
+    def test_ignores_non_numeric_paths(self) -> None:
+        results = _parse_search_html(_SEARCH_HTML_FIXTURE)
+        ids = [r["doc_id"] for r in results]
+        assert "some-slug" not in ids
+
+    def test_deduplicates_doc_ids(self) -> None:
+        results = _parse_search_html(_SEARCH_HTML_FIXTURE)
+        ids = [r["doc_id"] for r in results]
+        assert ids.count("35496") == 1
+
+    def test_ignores_category_links(self) -> None:
+        results = _parse_search_html(_SEARCH_HTML_FIXTURE)
+        # /category/law/ не числовой id — не попадает.
+        ids = [r["doc_id"] for r in results]
+        assert len(ids) == 2
+
+    def test_url_is_absolute(self) -> None:
+        results = _parse_search_html(_SEARCH_HTML_FIXTURE)
+        for r in results:
+            assert r["url"].startswith("https://www.pravo.gov66.ru/")
+
+    def test_title_extracted(self) -> None:
+        results = _parse_search_html(_SEARCH_HTML_FIXTURE)
+        doc = next(r for r in results if r["doc_id"] == "35496")
+        assert "изъятии" in doc["title"].lower()
+
+    def test_empty_html_returns_empty(self) -> None:
+        assert _parse_search_html("") == []
+
+    def test_html_no_doc_links_returns_empty(self) -> None:
+        html = "<html><body><a href='/category/'>Cat</a></body></html>"
+        assert _parse_search_html(html) == []
+
+
+class TestExtractPdfUrl:
+    def test_prefers_non_ecp_pdf(self) -> None:
+        url = _extract_pdf_url(_DOC_HTML_WITH_PDF)
+        assert url is not None
+        assert "ecp" not in url
+        assert url.endswith(".pdf")
+
+    def test_non_ecp_pdf_is_absolute(self) -> None:
+        url = _extract_pdf_url(_DOC_HTML_WITH_PDF)
+        assert url is not None
+        assert url.startswith("https://www.pravo.gov66.ru/media/pravo/")
+
+    def test_falls_back_to_ecp_when_no_regular_pdf(self) -> None:
+        url = _extract_pdf_url(_DOC_HTML_ECP_ONLY)
+        assert url is not None
+        assert "ecp" in url
+        assert url.endswith(".pdf")
+
+    def test_returns_none_when_no_pdf(self) -> None:
+        url = _extract_pdf_url(_DOC_HTML_NO_PDF)
+        assert url is None
+
+    def test_ecp_fallback_is_absolute(self) -> None:
+        url = _extract_pdf_url(_DOC_HTML_ECP_ONLY)
+        assert url is not None
+        assert url.startswith("https://www.pravo.gov66.ru/")

backend/tests/workers/test_reservation_ingest.py

@@ -0,0 +1,289 @@
+"""Тесты ingest_reservations (#1062) — offline, без живой сети и реальной БД.
+
+Мокаем: pravo_gov66_client (search/get_pdf_url/fetch_pdf) + extract_from_pdf + SessionLocal.
+Проверяем: фильтрацию заголовков, дедупликацию doc_id, upsert-параметры, graceful на ошибках.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from datetime import date
+from typing import Any
+
+from app.services.scrapers.page_reservation_parser import ReservationRecord
+from app.workers.tasks import reservation_ingest
+
+# ── Двойники ──────────────────────────────────────────────────────────────────
+
+
+class _FakeDB:
+    """Session-двойник: пишет execute-вызовы, begin_nested как no-op CM."""
+
+    def __init__(self) -> None:
+        self.executed: list[tuple[Any, dict[str, Any] | None]] = []
+        self.commit_count = 0
+        self.closed = False
+
+    @contextmanager
+    def begin_nested(self) -> Any:
+        yield
+
+    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any:
+        self.executed.append((sql, params))
+        return None
+
+    def commit(self) -> None:
+        self.commit_count += 1
+
+    def rollback(self) -> None:
+        pass
+
+    def close(self) -> None:
+        self.closed = True
+
+
+# ── Фикстуры ──────────────────────────────────────────────────────────────────
+
+_SAMPLE_RECORDS = [
+    ReservationRecord(
+        cad_num="66:41:0101001:1",
+        reservation_kind="изъятие",
+        act_number="509-ПП",
+        act_date=date(2024, 3, 12),
+        purpose="реконструкция ул. Татищева",
+        raw_excerpt="...участок 66:41:0101001:1...",
+    ),
+    ReservationRecord(
+        cad_num="66:41:0101001:2",
+        reservation_kind="изъятие",
+        act_number="509-ПП",
+        act_date=date(2024, 3, 12),
+        purpose="реконструкция ул. Татищева",
+        raw_excerpt=None,
+    ),
+]
+
+_RELEVANT_HITS = [
+    {
+        "doc_id": "35496",
+        "title": "Об изъятии земельных участков Екатеринбург",
+        "url": "https://www.pravo.gov66.ru/35496/",
+    },
+    {
+        "doc_id": "34812",
+        "title": "О резервировании земельных участков Свердловской области",
+        "url": "https://www.pravo.gov66.ru/34812/",
+    },
+]
+
+_IRRELEVANT_HIT = {
+    "doc_id": "99999",
+    "title": "О бюджете на 2024 год",
+    "url": "https://www.pravo.gov66.ru/99999/",
+}
+
+_PDF_BYTES = b"%PDF-1.4 fake content"
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _patch_enumerate(
+    monkeypatch: Any,
+    db: _FakeDB,
+    hits: list[dict[str, Any]],
+    pdf_url: str | None = "https://www.pravo.gov66.ru/media/pravo/509-ПП_abc.pdf",
+    pdf_bytes: bytes = _PDF_BYTES,
+    records: list[ReservationRecord] = _SAMPLE_RECORDS,
+) -> None:
+    """Патчит все зависимости ingest_reservations для enumerate-режима."""
+    monkeypatch.setattr(reservation_ingest, "SessionLocal", lambda: db)
+    monkeypatch.setattr(reservation_ingest, "search_documents", lambda q, **kw: hits)
+    monkeypatch.setattr(reservation_ingest, "get_document_pdf_url", lambda doc_id: pdf_url)
+    monkeypatch.setattr(reservation_ingest, "fetch_pdf", lambda url: pdf_bytes)
+    monkeypatch.setattr(reservation_ingest, "extract_from_pdf", lambda b: records)
+
+
+# ── Тесты фильтрации заголовков ───────────────────────────────────────────────
+
+
+class TestIsRelevantTitle:
+    def test_relevant_izyatie_ekb(self) -> None:
+        assert reservation_ingest._is_relevant_title(
+            "Об изъятии земельных участков в г. Екатеринбурге"
+        )
+
+    def test_relevant_rezervirovanie_sverdlov(self) -> None:
+        assert reservation_ingest._is_relevant_title(
+            "О резервировании земельных участков Свердловской области"
+        )
+
+    def test_irrelevant_no_geo(self) -> None:
+        assert not reservation_ingest._is_relevant_title(
+            "Об изъятии земельных участков в г. Нижний Тагил"
+        )
+
+    def test_irrelevant_no_operation(self) -> None:
+        assert not reservation_ingest._is_relevant_title(
+            "О бюджете Свердловской области на 2024 год"
+        )
+
+    def test_irrelevant_empty(self) -> None:
+        assert not reservation_ingest._is_relevant_title("")
+
+
+# ── Тесты enumerate-режима ────────────────────────────────────────────────────
+
+
+class TestIngestEnumerate:
+    def test_returns_docs_and_parcels(self, monkeypatch: Any) -> None:
+        db = _FakeDB()
+        _patch_enumerate(monkeypatch, db, hits=_RELEVANT_HITS)
+        result = reservation_ingest.ingest_reservations()
+        assert result["docs"] >= 1
+        assert result["parcels"] >= 1
+
+    def test_filters_irrelevant_titles(self, monkeypatch: Any) -> None:
+        """Документ с нерелевантным заголовком не должен попасть в ingest."""
+        db = _FakeDB()
+        irrelevant_only = [_IRRELEVANT_HIT]
+        _patch_enumerate(monkeypatch, db, hits=irrelevant_only)
+        result = reservation_ingest.ingest_reservations()
+        assert result["docs"] == 0
+        assert result["parcels"] == 0
+        assert db.executed == []
+
+    def test_deduplicates_doc_id_across_queries(self, monkeypatch: Any) -> None:
+        """Один и тот же doc_id из двух запросов инжестируется один раз."""
+        db = _FakeDB()
+        same_hit = _RELEVANT_HITS[:1]  # один хит
+
+        call_count = [0]
+
+        def _search_twice(q: str, **kw: Any) -> list[dict[str, Any]]:
+            call_count[0] += 1
+            return same_hit  # оба запроса возвращают один и тот же doc_id
+
+        monkeypatch.setattr(reservation_ingest, "SessionLocal", lambda: db)
+        monkeypatch.setattr(reservation_ingest, "search_documents", _search_twice)
+        monkeypatch.setattr(
+            reservation_ingest,
+            "get_document_pdf_url",
+            lambda doc_id: "https://www.pravo.gov66.ru/media/pravo/test.pdf",
+        )
+        monkeypatch.setattr(reservation_ingest, "fetch_pdf", lambda url: _PDF_BYTES)
+        monkeypatch.setattr(reservation_ingest, "extract_from_pdf", lambda b: _SAMPLE_RECORDS[:1])
+
+        result = reservation_ingest.ingest_reservations()
+        # 2 запроса, но 1 уникальный doc_id → fetched=1
+        assert result["docs"] == 1
+
+    def test_skips_doc_when_no_pdf_url(self, monkeypatch: Any) -> None:
+        db = _FakeDB()
+        _patch_enumerate(monkeypatch, db, hits=_RELEVANT_HITS[:1], pdf_url=None)
+        result = reservation_ingest.ingest_reservations()
+        assert result["docs"] == 0
+        assert result["parcels"] == 0
+
+    def test_skips_doc_when_pdf_fetch_fails(self, monkeypatch: Any) -> None:
+        db = _FakeDB()
+        monkeypatch.setattr(reservation_ingest, "SessionLocal", lambda: db)
+        monkeypatch.setattr(
+            reservation_ingest, "search_documents", lambda q, **kw: _RELEVANT_HITS[:1]
+        )
+        monkeypatch.setattr(
+            reservation_ingest,
+            "get_document_pdf_url",
+            lambda doc_id: "https://www.pravo.gov66.ru/media/pravo/test.pdf",
+        )
+        monkeypatch.setattr(
+            reservation_ingest, "fetch_pdf", lambda url: (_ for _ in ()).throw(OSError("net err"))
+        )
+        monkeypatch.setattr(reservation_ingest, "extract_from_pdf", lambda b: _SAMPLE_RECORDS[:1])
+        result = reservation_ingest.ingest_reservations()
+        assert result["docs"] == 0
+
+    def test_upsert_params_contain_cad_num(self, monkeypatch: Any) -> None:
+        db = _FakeDB()
+        _patch_enumerate(monkeypatch, db, hits=_RELEVANT_HITS[:1], records=_SAMPLE_RECORDS[:1])
+        reservation_ingest.ingest_reservations()
+        assert db.executed, "должен быть хотя бы один execute"
+        _, params = db.executed[0]
+        assert params is not None
+        assert params["cad_num"] == "66:41:0101001:1"
+        assert params["source"] == "pravo_gov66"
+
+    def test_db_closed_after_run(self, monkeypatch: Any) -> None:
+        db = _FakeDB()
+        _patch_enumerate(monkeypatch, db, hits=[])
+        reservation_ingest.ingest_reservations()
+        assert db.closed
+
+    def test_skips_empty_records(self, monkeypatch: Any) -> None:
+        """PDF без кад-номеров → ни одного execute."""
+        db = _FakeDB()
+        _patch_enumerate(monkeypatch, db, hits=_RELEVANT_HITS[:1], records=[])
+        result = reservation_ingest.ingest_reservations()
+        assert result["parcels"] == 0
+        assert db.executed == []
+
+
+# ── Тесты explicit-docs (back-compat) ─────────────────────────────────────────
+
+
+class TestIngestExplicitDocs:
+    def test_explicit_docs_bypass_enumerate(self, monkeypatch: Any) -> None:
+        """docs=... → enumerate не вызывается, search_documents не дёргается."""
+        db = _FakeDB()
+        monkeypatch.setattr(reservation_ingest, "SessionLocal", lambda: db)
+
+        def _should_not_call(*a: Any, **kw: Any) -> list[dict[str, Any]]:
+            raise AssertionError("enumerate не должен вызываться")
+
+        monkeypatch.setattr(reservation_ingest, "search_documents", _should_not_call)
+        monkeypatch.setattr(reservation_ingest, "fetch_pdf", lambda url: _PDF_BYTES)
+        monkeypatch.setattr(reservation_ingest, "extract_from_pdf", lambda b: _SAMPLE_RECORDS[:1])
+
+        explicit = [
+            {
+                "url": "https://www.pravo.gov66.ru/media/pravo/509-ПП_abc.pdf",
+                "act_number": "509-ПП",
+                "basis_act": "Постановление № 509-ПП",
+                "purpose": "реконструкция",
+                "kind": "изъятие",
+            }
+        ]
+        result = reservation_ingest.ingest_reservations(docs=explicit)
+        assert result["docs"] == 1
+        assert result["parcels"] == 1
+
+    def test_explicit_docs_upsert_uses_doc_kind(self, monkeypatch: Any) -> None:
+        """Если rec.reservation_kind пуст, берётся kind из doc."""
+        db = _FakeDB()
+        monkeypatch.setattr(reservation_ingest, "SessionLocal", lambda: db)
+        monkeypatch.setattr(reservation_ingest, "fetch_pdf", lambda url: _PDF_BYTES)
+
+        no_kind_record = ReservationRecord(
+            cad_num="66:41:0000001:1",
+            reservation_kind="",  # пустой — должен взяться из doc
+            act_number=None,
+            act_date=None,
+            purpose=None,
+            raw_excerpt=None,
+        )
+        monkeypatch.setattr(reservation_ingest, "extract_from_pdf", lambda b: [no_kind_record])
+
+        explicit = [
+            {
+                "url": "https://x.ru/test.pdf",
+                "act_number": "1-ПП",
+                "basis_act": "Постановление",
+                "purpose": None,
+                "kind": "резервирование",
+            }
+        ]
+        reservation_ingest.ingest_reservations(docs=explicit)
+        assert db.executed
+        _, params = db.executed[0]
+        assert params is not None
+        assert params["reservation_kind"] == "резервирование"