fix(sf): ekb_ppt_tep post-фильтры + врезка в analyze (#1136) #1299
7 changed files with 449 additions and 6 deletions
|
|
@ -87,6 +87,11 @@ def _parse_table11(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
|||
|
||||
Структура: № | Наименование зоны | Сущ., га (%) | Проект, га (%)
|
||||
Заголовок (первые 2 строки) пропускается по признаку: №/None в col[0].
|
||||
|
||||
Post-фильтр (#1136 follow-up #1133): отбрасываем строки с пустым
|
||||
``zone_name`` (преамбульные артефакты merged-cells на образце ppt2018_22823
|
||||
давали 2 «фантомные» строки без названия зоны при 12 реальных). «Итого»-
|
||||
строка остаётся (распознаётся по тексту, не по zone_name).
|
||||
"""
|
||||
result: list[dict[str, Any]] = []
|
||||
header_skipped = False
|
||||
|
|
@ -116,6 +121,12 @@ def _parse_table11(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
|||
if not no and not is_total:
|
||||
continue
|
||||
|
||||
# Post-фильтр (#1136): отбрасываем строки без осмысленного zone_name
|
||||
# (преамбульный шум merged-cells). Исключение — Итого-строка: у неё
|
||||
# zone_name может быть пуст (значение часто в col[0]).
|
||||
if not zone and not is_total:
|
||||
continue
|
||||
|
||||
record: dict[str, Any] = {
|
||||
"no": no,
|
||||
"zone_name": zone,
|
||||
|
|
@ -193,8 +204,18 @@ def _parse_table13(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
|||
Нас интересуют: фаза (col[0]), состав объекта (col[1..2]), зона (col[13]),
|
||||
площадь (col[15]), финансирование (col[17]), программа (col[19]).
|
||||
Извлекаем только строки с непустым полем «зона» или «состав объекта».
|
||||
|
||||
Post-фильтры (#1136 follow-up #1133): на образце ppt2018_22823 merged-cells
|
||||
давали ~364 строк continuation-шума при ~30-40 реальных. Применяем:
|
||||
|
||||
1. Дедуп по ключу (phase, composition, zone, area_ha_norm) — повторы
|
||||
continuation-строк merged-cells.
|
||||
2. Отбрасываем «короткий-без-зоны» шум: composition < 4 символов И zone пуст
|
||||
(артефакты разрыва ячеек).
|
||||
3. Отбрасываем строки, у которых composition — одна цифра/один знак
|
||||
(нумерация осколков merged-cells).
|
||||
"""
|
||||
result: list[dict[str, Any]] = []
|
||||
rows_raw: list[dict[str, Any]] = []
|
||||
current_phase: str = ""
|
||||
header_skipped = False
|
||||
|
||||
|
|
@ -248,7 +269,7 @@ def _parse_table13(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
|||
if not zone and "объект" in composition.lower() and len(composition) < 50:
|
||||
continue
|
||||
|
||||
result.append(
|
||||
rows_raw.append(
|
||||
{
|
||||
"phase": current_phase,
|
||||
"composition": composition,
|
||||
|
|
@ -260,6 +281,37 @@ def _parse_table13(raw_rows: list[list[str | None]]) -> list[dict[str, Any]]:
|
|||
}
|
||||
)
|
||||
|
||||
# Post-фильтр (#1136): шум merged-cells.
|
||||
# 1. Короткий-без-зоны: composition короткий (< 4 симв.) И zone пуст —
|
||||
# типично осколок ячейки («1», «-», «—»).
|
||||
# 2. Composition — одна цифра/знак-пунктуация: «1», «2», «-» и т.п.
|
||||
def _is_noise(rec: dict[str, Any]) -> bool:
|
||||
comp = rec["composition"]
|
||||
zone = rec["zone"]
|
||||
# Короткий composition без зоны — осколок merged-cell.
|
||||
if not zone and len(comp) < 4:
|
||||
return True
|
||||
# Composition — только цифра/знак (один знак, или комбинация цифр/пунктуации).
|
||||
if comp and not re.search(r"[A-Za-zА-Яа-я]", comp) and len(comp) < 6:
|
||||
return True
|
||||
return False
|
||||
|
||||
rows_clean = [r for r in rows_raw if not _is_noise(r)]
|
||||
|
||||
# Post-фильтр (#1136): дедуп по (phase, composition, zone, area_ha_norm).
|
||||
# area_ha сравниваем по числовому значению (если есть) — иначе по строке.
|
||||
seen: set[tuple[str, str, str, str]] = set()
|
||||
result: list[dict[str, Any]] = []
|
||||
for rec in rows_clean:
|
||||
area_key = (
|
||||
f"{rec['area_ha_num']:.4f}" if rec["area_ha_num"] is not None else rec["area_ha"]
|
||||
)
|
||||
key = (rec["phase"], rec["composition"], rec["zone"], area_key)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
result.append(rec)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ from app.services.site_finder.krt_lookup import parcel_krt_requisites
|
|||
from app.services.site_finder.okn_lookup import parcel_okn_objects
|
||||
from app.services.site_finder.pat_lookup import parcel_pat_subzones
|
||||
from app.services.site_finder.planning_lookup import parcel_planning_overlaps
|
||||
from app.services.site_finder.ppt_tep_lookup import parcel_ppt_tep
|
||||
from app.services.site_finder.reservation_lookup import parcel_reservations
|
||||
from app.services.site_finder.zone_regulation import get_or_fetch_zone_regulation
|
||||
|
||||
|
|
@ -127,6 +128,9 @@ def build_ird_analyze_block(
|
|||
dict: ``ird_overlaps``/``ird_by_kind`` (DB-overlay, incl opportunity_*),
|
||||
``opportunity_overlaps`` (подмножество layer_kind='opportunity_*'),
|
||||
``planning_projects`` (ППТ/ПМТ overlap, DB, м.134 #1085),
|
||||
``ppt_tep`` (list, ТЭП-контент ППТ/ПМТ из ekb_ppt_tep #1085/#1136: zone_balance / tep /
|
||||
phasing для участков в границах ППТ/ПМТ; связка doc_ref↔source_key/doc_full_name,
|
||||
graceful, пусто если ingest ещё не запускался или матча нет),
|
||||
``reservations`` (list, изъятие/резервирование из land_reservation по cad_num, #1091/#1062),
|
||||
``krt_requisites`` (list, КРТ-реквизиты из ekb_krt_sites + planning_projects, #1060),
|
||||
``functional_zones`` (list, генплан-зоны из ekb_genplan_functional_zone, #1058),
|
||||
|
|
@ -151,6 +155,12 @@ def build_ird_analyze_block(
|
|||
"ird_by_kind": overlay.get("ird_by_kind", {}),
|
||||
"opportunity_overlaps": opportunity,
|
||||
"planning_projects": parcel_planning_overlaps(db, geom_wkt),
|
||||
# ТЭП-контент ППТ/ПМТ (#1136 follow-up #1133, GG-форсайт #1067 / #1085):
|
||||
# zone_balance / tep / phasing из ekb_ppt_tep, привязанные к planning_projects по
|
||||
# doc_ref↔source_key/doc_full_name. DB JOIN, дёшево (GIST), graceful — пусто если
|
||||
# таблица ekb_ppt_tep ещё пуста (нет ingest) или матча doc_ref нет. Идёт рядом с
|
||||
# planning_projects/krt_requisites — DB-источники основного потока.
|
||||
"ppt_tep": parcel_ppt_tep(db, geom_wkt),
|
||||
"reservations": parcel_reservations(db, cad_num),
|
||||
# Граддокументация (#1061): status «разработана» + реквизиты ПАГЕ из planning_projects.
|
||||
# DB, дёшево (ST_Intersects GIST), graceful. Вместе с planning_projects — один источник.
|
||||
|
|
|
|||
94
backend/app/services/site_finder/ppt_tep_lookup.py
Normal file
94
backend/app/services/site_finder/ppt_tep_lookup.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""ТЭП ППТ/ПМТ-lookup для analyze_parcel (#1136 follow-up #1133, GG-форсайт #1067).
|
||||
|
||||
Читает ``ekb_ppt_tep`` (м.138, наполняется ``ekb_ppt_tep_sync`` Celery-воркером #1085) и
|
||||
отдаёт ТЭП-контент пояснительных записок (zone_balance / tep / phasing) для проектов
|
||||
планировки/межевания, пересекающих участок.
|
||||
|
||||
Связка ТЭП ↔ геометрия (best-effort, как описано в м.138):
|
||||
``ekb_ppt_tep.doc_ref`` ↔ ``planning_projects.source_key`` (текстовое равенство)
|
||||
ИЛИ ``planning_projects.doc_full_name`` (LIKE).
|
||||
|
||||
Прямой FK не ставим: doc_ref пишется парсером по PDF-источнику (slug/seed), а в WFS
|
||||
source_key — целое и не совпадает форматом. Сопоставление — две дешёвые ветки:
|
||||
1) ``doc_ref = source_key::text`` (прямое равенство);
|
||||
2) ``doc_ref ILIKE '%doc_full_name%'`` (fallback для slug-ов вида ppt2018_22823).
|
||||
|
||||
Граф вызова: build_ird_analyze_block → parcel_ppt_tep → JOIN planning_projects ∩ ekb_ppt_tep.
|
||||
Graceful: нет таблицы / пустой WKT / БД-ошибка → []. Зеркалит стиль ``planning_lookup.py``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.exc import OperationalError, ProgrammingError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ТЭП-контент для проектов планировки, пересекающих участок.
|
||||
# Join: planning_projects (геометрия) ⋈ ekb_ppt_tep (контент) по doc_ref↔source_key/doc_full_name.
|
||||
# ST_Intersects через GIST idx_planning_projects_geom (4326 ↔ WKT 4326).
|
||||
# CAST psycopg v3 — никогда :param::type (vault Pattern_CAST_AS_Type).
|
||||
_PPT_TEP_OVERLAP_SQL = text(
|
||||
"""
|
||||
SELECT
|
||||
pp.project_type,
|
||||
pp.doc_status_name,
|
||||
pp.full_name,
|
||||
pp.doc_full_name,
|
||||
pp.source_key,
|
||||
t.doc_ref,
|
||||
t.zone_balance,
|
||||
t.tep,
|
||||
t.phasing,
|
||||
t.source_pdf,
|
||||
t.fetched_at
|
||||
FROM planning_projects pp
|
||||
JOIN ekb_ppt_tep t ON (
|
||||
t.doc_ref = CAST(pp.source_key AS text)
|
||||
OR (pp.doc_full_name IS NOT NULL AND t.doc_ref ILIKE '%' || pp.doc_full_name || '%')
|
||||
)
|
||||
WHERE ST_Intersects(pp.geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
|
||||
ORDER BY pp.dmd_actual_year DESC NULLS LAST, pp.project_type
|
||||
"""
|
||||
)
|
||||
|
||||
|
||||
def parcel_ppt_tep(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
|
||||
"""ТЭП-контент ППТ/ПМТ для участка (Табл.11/12/13 пояснительных записок).
|
||||
|
||||
Args:
|
||||
db: SQLAlchemy-сессия.
|
||||
parcel_wkt: геометрия участка в WKT (EPSG:4326). None / пусто → [].
|
||||
|
||||
Returns:
|
||||
list[{project_type, doc_status_name, full_name, doc_full_name, source_key,
|
||||
doc_ref, zone_balance, tep, phasing, source_pdf, fetched_at}].
|
||||
Каждая запись — один ППТ/ПМТ-документ, чья геометрия пересекает участок И
|
||||
для которого есть ТЭП-контент в ``ekb_ppt_tep``.
|
||||
|
||||
Пустой список при:
|
||||
- parcel_wkt пуст / None;
|
||||
- таблица ``ekb_ppt_tep`` или ``planning_projects`` ещё не задеплоена;
|
||||
- нет пересечений по геометрии;
|
||||
- нет матча doc_ref ↔ source_key/doc_full_name (best-effort, graceful);
|
||||
- БД-ошибка (analyze не падает).
|
||||
"""
|
||||
if not parcel_wkt:
|
||||
return []
|
||||
try:
|
||||
rows = db.execute(_PPT_TEP_OVERLAP_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
|
||||
except (OperationalError, ProgrammingError) as exc:
|
||||
# Таблица ещё не задеплоена / БД-ошибка — graceful degrade.
|
||||
logger.warning("parcel_ppt_tep: ekb_ppt_tep / planning_projects недоступна, skip: %s", exc)
|
||||
return []
|
||||
except Exception as exc:
|
||||
logger.warning("parcel_ppt_tep: неожиданная ошибка, skip: %s", exc)
|
||||
return []
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
__all__ = ["parcel_ppt_tep"]
|
||||
|
|
@ -20,13 +20,26 @@ from app.workers.celery_app import celery_app
|
|||
logger = logging.getLogger(__name__)
|
||||
|
||||
# ── Seed-документы ─────────────────────────────────────────────────────────────
|
||||
# Один образец: ППТ 22823 (2018), пояснительная записка.
|
||||
# URL — placeholder: VERIFY на екатеринбург.рф/документация/ппт/...
|
||||
# Реальный URL уточнить у пользователя перед прод-прогоном.
|
||||
# Образец: ППТ 22823 (2018), пояснительная записка.
|
||||
#
|
||||
# URL — placeholder. Реальный URL пояснительной записки лежит на ГИСОГД ЕКБ
|
||||
# (https://gisogd.ekburg.ru/) под номером проекта планировки, но прямой PDF-линк
|
||||
# требует ручного поиска через UI (#1136). До тех пор — ingest скипает с
|
||||
# WARNING и метрики остаются {"docs": 0}.
|
||||
#
|
||||
# Override-пути для прогона:
|
||||
# 1. Передать в task: ingest_ppt_tep([{"doc_ref": "...", "url": "...",
|
||||
# "project_type": "ppt"}]).
|
||||
# 2. Подменить _SEED_DOCS в env-fixture для теста.
|
||||
#
|
||||
# Acceptance follow-up #1136: после уточнения URL — разовый прогон,
|
||||
# verify {"docs": 1} + непустые zone_balance/tep/phasing.
|
||||
_SEED_DOCS: list[dict] = [
|
||||
{
|
||||
"doc_ref": "ppt2018_22823",
|
||||
"url": "# VERIFY URL: https://xn--80acgfbsl1azdqr.xn--p1ai/документация/ппт/22823",
|
||||
# VERIFY URL: реальный документ на gisogd.ekburg.ru или екатеринбург.рф.
|
||||
# placeholder начинается с «# VERIFY» — ingest пропускает с WARNING.
|
||||
"url": "# VERIFY URL: https://gisogd.ekburg.ru/documents/ppt/22823/poyasnit_PPT.pdf",
|
||||
"project_type": "ppt",
|
||||
},
|
||||
]
|
||||
|
|
|
|||
|
|
@ -179,6 +179,36 @@ class TestParseTable11:
|
|||
result = _parse_table11(garbage)
|
||||
assert result == []
|
||||
|
||||
def test_preamble_artifact_filtered_out(self) -> None:
|
||||
"""Post-фильтр #1136: строки с № но без zone_name отбрасываются.
|
||||
|
||||
Это «фантомные» строки преамбульного merged-cells шума (на ppt2018_22823
|
||||
давали 2 артефакта при 12 реальных зонах).
|
||||
"""
|
||||
rows: list[list[str | None]] = [
|
||||
["№", "Наименование зоны", "Площадь", None],
|
||||
[None, None, "Сущ., га (%)", "Проект, га\n(%)"],
|
||||
# Артефакт: № есть, но zone_name пустой — должен отброситься.
|
||||
["1", "", "15,41 га", "11,34 га"],
|
||||
# Артефакт: № есть, zone_name пустой (None) — должен отброситься.
|
||||
["2", None, "0,24 га", "0,25 га"],
|
||||
# Реальная зона: остаётся.
|
||||
["3", "Зона жилой застройки", "10,0 га", "12,0 га"],
|
||||
# Итого: остаётся (zone_name пуст, но is_total).
|
||||
["Итого:", None, "25,65 га", "23,59 га"],
|
||||
]
|
||||
result = _parse_table11(rows)
|
||||
zone_names = [r["zone_name"] for r in result]
|
||||
# Реальная зона должна быть в результате
|
||||
assert "Зона жилой застройки" in zone_names
|
||||
# Артефакты (zone_name пуст, не is_total) отброшены
|
||||
non_total_no_zone = [
|
||||
r for r in result if not r["zone_name"] and not r.get("is_total")
|
||||
]
|
||||
assert non_total_no_zone == []
|
||||
# Итого-строка осталась
|
||||
assert any(r.get("is_total") for r in result)
|
||||
|
||||
|
||||
# ── Тесты _parse_table12 ──────────────────────────────────────────────────────
|
||||
|
||||
|
|
@ -237,6 +267,92 @@ class TestParseTable13:
|
|||
def test_empty_input_returns_empty(self) -> None:
|
||||
assert _parse_table13([]) == []
|
||||
|
||||
def test_dedup_removes_continuation_noise(self) -> None:
|
||||
"""Post-фильтр #1136: дубликаты merged-cells continuation схлопываются.
|
||||
|
||||
На ppt2018_22823 364 «строк» сворачивались в ~30-40 реальных.
|
||||
Проверяем dedup по (phase, composition, zone, area_ha_norm).
|
||||
"""
|
||||
# Реальная строка и три её continuation-копии (merged-cells даёт повторы).
|
||||
real_row: list[str | None] = [
|
||||
"I этап\n2018-2021",
|
||||
"",
|
||||
"",
|
||||
"Многофункциональная ледовая арена",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
"",
|
||||
"Зона объектов спорта",
|
||||
None,
|
||||
"4,97",
|
||||
None,
|
||||
"Внебюджетные",
|
||||
None,
|
||||
"",
|
||||
]
|
||||
rows: list[list[str | None]] = [
|
||||
_TABLE13_ROWS[0],
|
||||
_TABLE13_ROWS[1],
|
||||
real_row,
|
||||
real_row, # continuation-дубль 1
|
||||
real_row, # continuation-дубль 2
|
||||
real_row, # continuation-дубль 3
|
||||
]
|
||||
result = _parse_table13(rows)
|
||||
# Все 4 копии должны схлопнуться в 1.
|
||||
matching = [
|
||||
r
|
||||
for r in result
|
||||
if r["zone"] == "Зона объектов спорта" and r.get("area_ha_num") == pytest.approx(4.97)
|
||||
]
|
||||
assert len(matching) == 1
|
||||
|
||||
def test_short_noise_rows_filtered(self) -> None:
|
||||
"""Post-фильтр #1136: composition < 4 символов без zone — шум merged-cells."""
|
||||
rows: list[list[str | None]] = [
|
||||
_TABLE13_ROWS[0],
|
||||
_TABLE13_ROWS[1],
|
||||
# Шум: composition «-», без zone — должен отброситься
|
||||
[None, "-"] + [None] * 18,
|
||||
# Шум: composition «1», без zone — должен отброситься
|
||||
[None, "1"] + [None] * 18,
|
||||
# Реальная строка
|
||||
[
|
||||
"I этап",
|
||||
"",
|
||||
"Жилой комплекс",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
"Зона Ж-1",
|
||||
None,
|
||||
"5,0",
|
||||
None,
|
||||
"Бюджет",
|
||||
None,
|
||||
"",
|
||||
],
|
||||
]
|
||||
result = _parse_table13(rows)
|
||||
compositions = [r["composition"] for r in result]
|
||||
assert "-" not in compositions
|
||||
# Одиночная «1» должна отлететь как номер-осколок
|
||||
# (real_row даёт composition='Жилой комплекс').
|
||||
assert any("Жилой" in c for c in compositions)
|
||||
|
||||
|
||||
# ── Тест parse_ppt_tep с mock pdfplumber ─────────────────────────────────────
|
||||
|
||||
|
|
|
|||
|
|
@ -75,6 +75,8 @@ def _patch_overlay(monkeypatch: Any, overlay: dict[str, Any]) -> None:
|
|||
monkeypatch.setattr(ird_analyze, "parcel_granddoc", lambda db, wkt: [])
|
||||
# По умолчанию: пустой список — pat_subzones=[] (graceful, таблица может не быть задеплоена).
|
||||
monkeypatch.setattr(ird_analyze, "parcel_pat_subzones", lambda db, wkt: [])
|
||||
# По умолчанию: пустой список — ppt_tep=[] (graceful, ingest ещё не запускался #1136).
|
||||
monkeypatch.setattr(ird_analyze, "parcel_ppt_tep", lambda db, wkt: [])
|
||||
|
||||
|
||||
def test_full_block_assembly(monkeypatch: Any) -> None:
|
||||
|
|
@ -505,3 +507,58 @@ def test_pat_subzones_empty_when_no_overlap(monkeypatch: Any) -> None:
|
|||
|
||||
assert "pat_subzones" in block
|
||||
assert block["pat_subzones"] == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ТЭП ППТ/ПМТ (#1136 follow-up #1133)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_PPT_TEP_ROWS: list[dict[str, Any]] = [
|
||||
{
|
||||
"project_type": "ppt",
|
||||
"doc_status_name": "действующий",
|
||||
"full_name": "ПАГЕ № 22823 от 12.10.2018",
|
||||
"doc_full_name": "ППТ ул. Челюскинцев",
|
||||
"source_key": 22823,
|
||||
"doc_ref": "ppt2018_22823",
|
||||
"zone_balance": [{"zone_name": "Зона Ж-1", "project_ha_num": 5.0}],
|
||||
"tep": [{"indicator": "Площадь территории", "target": "36,16"}],
|
||||
"phasing": [{"phase": "I этап", "zone": "Зона Ж-1", "area_ha_num": 5.0}],
|
||||
"source_pdf": "https://example.com/ppt22823.pdf",
|
||||
"fetched_at": "2026-06-07T11:55:00Z",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_ppt_tep_wired_to_block(monkeypatch: Any) -> None:
|
||||
"""parcel_ppt_tep возвращает запись → ключ ppt_tep присутствует в блоке (#1136)."""
|
||||
_patch_overlay(monkeypatch, _OVERLAY)
|
||||
monkeypatch.setattr(ird_analyze, "parcel_ppt_tep", lambda db, wkt: list(_PPT_TEP_ROWS))
|
||||
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
|
||||
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
|
||||
client = _FakeClient(fz=[])
|
||||
|
||||
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", None, None, client=client)
|
||||
|
||||
assert "ppt_tep" in block
|
||||
assert len(block["ppt_tep"]) == 1
|
||||
row = block["ppt_tep"][0]
|
||||
assert row["doc_ref"] == "ppt2018_22823"
|
||||
assert row["zone_balance"][0]["zone_name"] == "Зона Ж-1"
|
||||
assert row["tep"][0]["target"] == "36,16"
|
||||
assert row["phasing"][0]["phase"] == "I этап"
|
||||
|
||||
|
||||
def test_ppt_tep_empty_when_no_overlap(monkeypatch: Any) -> None:
|
||||
"""parcel_ppt_tep пуст (ingest ещё не запускался / нет матча) → ppt_tep == []."""
|
||||
_patch_overlay(monkeypatch, _OVERLAY)
|
||||
# _patch_overlay уже мокает parcel_ppt_tep → [], переопределяем явно для ясности.
|
||||
monkeypatch.setattr(ird_analyze, "parcel_ppt_tep", lambda db, wkt: [])
|
||||
monkeypatch.setattr(ird_analyze, "get_or_fetch_zone_regulation", lambda *a, **kw: None)
|
||||
monkeypatch.setattr(ird_analyze, "_krt_at", lambda db, lon, lat: [])
|
||||
client = _FakeClient(fz=[])
|
||||
|
||||
block = ird_analyze.build_ird_analyze_block(object(), "POINT(60 56)", 60.6, 56.8, client=client)
|
||||
|
||||
assert "ppt_tep" in block
|
||||
assert block["ppt_tep"] == []
|
||||
|
|
|
|||
101
backend/tests/services/test_ppt_tep_lookup.py
Normal file
101
backend/tests/services/test_ppt_tep_lookup.py
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
"""Тесты parcel_ppt_tep (#1136 follow-up #1133) — ТЭП ППТ/ПМТ overlap по участку.
|
||||
|
||||
БД не дёргается: фейковая сессия с .execute().mappings().all(). Проверяем graceful на
|
||||
отсутствие участка/таблицы и проброс строк.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.exc import ProgrammingError
|
||||
|
||||
from app.services.site_finder.ppt_tep_lookup import parcel_ppt_tep
|
||||
|
||||
|
||||
class _Result:
|
||||
def __init__(self, rows: list[dict[str, Any]]) -> None:
|
||||
self._rows = rows
|
||||
|
||||
def mappings(self) -> _Result:
|
||||
return self
|
||||
|
||||
def all(self) -> list[dict[str, Any]]:
|
||||
return self._rows
|
||||
|
||||
|
||||
class _DB:
|
||||
def __init__(self, rows: list[dict[str, Any]] | Exception) -> None:
|
||||
self._rows = rows
|
||||
|
||||
def execute(self, sql: Any, params: dict[str, Any]) -> _Result:
|
||||
if isinstance(self._rows, Exception):
|
||||
raise self._rows
|
||||
return _Result(self._rows)
|
||||
|
||||
|
||||
_PPT_TEP_ROW: dict[str, Any] = {
|
||||
"project_type": "ppt",
|
||||
"doc_status_name": "действующий",
|
||||
"full_name": "ПАГЕ № 22823 от 12.10.2018",
|
||||
"doc_full_name": "ППТ ул. Челюскинцев",
|
||||
"source_key": 22823,
|
||||
"doc_ref": "ppt2018_22823",
|
||||
"zone_balance": [{"zone_name": "Зона Ж-1", "project_ha_num": 5.0}],
|
||||
"tep": [{"indicator": "Площадь территории", "target": "36,16"}],
|
||||
"phasing": [{"phase": "I этап", "zone": "Зона Ж-1", "area_ha_num": 5.0}],
|
||||
"source_pdf": "https://example.com/ppt22823.pdf",
|
||||
"fetched_at": "2026-06-07T11:55:00Z",
|
||||
}
|
||||
|
||||
|
||||
def test_returns_ppt_tep_overlap() -> None:
|
||||
"""Один match по геометрии и doc_ref↔source_key → проброс строки."""
|
||||
out = parcel_ppt_tep(_DB([_PPT_TEP_ROW]), "POINT(60 56)") # type: ignore[arg-type]
|
||||
assert len(out) == 1
|
||||
row = out[0]
|
||||
assert row["doc_ref"] == "ppt2018_22823"
|
||||
assert row["source_key"] == 22823
|
||||
assert row["zone_balance"] == [{"zone_name": "Зона Ж-1", "project_ha_num": 5.0}]
|
||||
assert row["tep"][0]["target"] == "36,16"
|
||||
assert row["phasing"][0]["zone"] == "Зона Ж-1"
|
||||
|
||||
|
||||
def test_empty_wkt_returns_empty() -> None:
|
||||
"""parcel_wkt=None → []."""
|
||||
assert parcel_ppt_tep(_DB([]), None) == [] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_empty_string_wkt_returns_empty() -> None:
|
||||
"""parcel_wkt='' → []."""
|
||||
assert parcel_ppt_tep(_DB([]), "") == [] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_no_overlap_returns_empty() -> None:
|
||||
"""Запрос отработал, но пересечений нет → []."""
|
||||
assert parcel_ppt_tep(_DB([]), "POINT(60 56)") == [] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_missing_table_is_graceful() -> None:
|
||||
"""ekb_ppt_tep ещё не задеплоена → ProgrammingError → [] (analyze не падает)."""
|
||||
db = _DB(ProgrammingError("stmt", {}, Exception("relation ekb_ppt_tep does not exist")))
|
||||
assert parcel_ppt_tep(db, "POINT(60 56)") == [] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_unexpected_exception_is_graceful() -> None:
|
||||
"""Любая неожиданная ошибка БД → [] (analyze не падает)."""
|
||||
db = _DB(RuntimeError("connection refused"))
|
||||
assert parcel_ppt_tep(db, "POINT(60 56)") == [] # type: ignore[arg-type]
|
||||
|
||||
|
||||
def test_multiple_rows_preserved() -> None:
|
||||
"""Если match'ит несколько ППТ — все возвращаются."""
|
||||
row2: dict[str, Any] = {
|
||||
**_PPT_TEP_ROW,
|
||||
"doc_ref": "ppt2020_99999",
|
||||
"source_key": 99999,
|
||||
"full_name": "ПАГЕ № 99999 от 01.06.2020",
|
||||
}
|
||||
out = parcel_ppt_tep(_DB([_PPT_TEP_ROW, row2]), "POINT(60 56)") # type: ignore[arg-type]
|
||||
assert len(out) == 2
|
||||
assert {r["doc_ref"] for r in out} == {"ppt2018_22823", "ppt2020_99999"}
|
||||
Loading…
Add table
Reference in a new issue