fix(full-report): §7 на MultiPolygon + чистка печатного шума §3/§4 (вычитка живого отчёта)
All checks were successful
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI Trade-In / changes (pull_request) Successful in 7s
CI / changes (pull_request) Successful in 7s
CI Trade-In / backend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 2m0s
CI / backend-tests (pull_request) Successful in 14m44s

Постраничная вычитка PDF многоконтурного участка 66:41:0702017:131:

1. MultiPolygon роняет concept-стек через geometry._parse_polygon
   (expected Polygon) ДО буфер-стадии, которая мультиконтур уже умеет →
   §7 без концепции, §5 деградирует, market-price падает. Новый
   _largest_polygon_geojson нормализует к крупнейшему контуру (паттерн
   geometry.py:263-265) до ConceptInput; Polygon проходит identity,
   битая геометрия — вход как есть. Интерактивный POST /concepts на
   мультиконтурах по-прежнему 422 — вынесено в #2295.
2. Сиротские заголовки: page-break-after avoid для h2/h3.
3. Тепло/вода §3: агрегат «Суммарно (ЕКБ)», фильтр областного шума
   (Регионгаз по Ирбиту/Тавде и т.п.), кап топ-15/25 + «и ещё K»,
   усечение имён 120 симв; дефицитные ЕКБ-резервы гарантированно
   видимы (bucket до капа, тест-лок).
4. Конкуренты §4: скип безымянных с 0 лотов, группировка корпусов
   (имя+девелопер → min дистанция, сумма лотов, «(K корпусов)»).
5. ЗОУИТ-противоречие: overlaps непусты при has_zouit=false →
   «да (по данным НСПД)»; обратное направление не тронуто.
6. Призрачные НСПД-строки, дубль-колонка confidence, absorption → %,
   средний чек без «+» (unsigned-форматтер).

Refs #2259, #2295
This commit is contained in:
bot-backend 2026-07-04 00:17:01 +05:00
parent d15ff3d99d
commit 35a03df12b
5 changed files with 644 additions and 51 deletions

View file

@ -57,6 +57,39 @@ MAP_CONCEPT_PLACEHOLDER = "{{MAP_CONCEPT}}"
_DASH = ""
_NO_DATA = "нет данных"
# ── §3 connection-capacity: кап-строки + фильтр «шума» области ───────────────────
# Печатный отчёт по участку ЕКБ, а heat_system_reserves несёт ВСЕ ~56 систем области
# (Ирбит/Тавда/Красноуфимск/Первоуральск/Лесной/Нижняя Тура…). Не хардкодим список
# городов — отсекаем по ЯВНЫМ не-ЕКБ маркерам в имени организации/системы; остальное
# капим топ-N по резерву. Дефицит (отрицательный резерв) под кап НЕ прячем — честность.
_NON_EKB_MARKERS: tuple[str, ...] = (
"ирбит",
"тавда",
"красноуфимск",
"первоуральск",
"лесной",
"нижняя тура",
"нижний тагил",
"каменск",
"серов",
"асбест",
"ревда",
"полевск",
"березовск", # покрывает «Березовский» / «Берёзовский» (ё нормализуем ниже)
"верхняя пышма",
"среднеуральск",
"заречный",
"новоуральск",
"качканар",
"краснотурьинск",
)
_EKB_MARKER = "екатеринбург"
# Кап видимых строк тепло/вода-таблиц (сверх — «и ещё K систем …»).
_HEAT_ROW_CAP = 15
_WATER_ROW_CAP = 25
# Усечение длинных бюрократических имён систем.
_SYSTEM_NAME_MAX = 120
# Заголовки секций (якоря — id внутри) и титул документа.
_TITLE_DOC = "Отчёт по участку — Site Finder ПТИЦА"
_TITLE_S1 = "§1. Участок"
@ -149,7 +182,9 @@ p { margin: 4pt 0; }
не рвётся внутри заголовка (WeasyPrint поддерживает break-before/inside). */
.section { margin-bottom: 16pt; break-inside: avoid-page; }
.section + .section { break-before: page; }
h2, h3 { break-after: avoid-page; }
/* Заголовок НЕ должен осиротеть в конце страницы, оторвавшись от своей таблицы:
и логический break-after (WeasyPrint 60+), и легаси page-break-after (fallback). */
h2, h3 { page-break-after: avoid; break-after: avoid-page; }
table { break-inside: avoid-page; }
/* Титул */
@ -285,6 +320,17 @@ def _fmt_money_signed(value: Any) -> str:
return f"{sign}{f'{round(magnitude):,}'.replace(',', ' ')}"
def _fmt_money(value: Any) -> str:
"""Деньги округлением до млн БЕЗ знака: «8 млн ₽» / «216 млн ₽». PURE.
Для абсолютных величин-ЦЕН (средний чек, справочная цена), где «+» неуместен это
не дельта. Отрицательные (маловероятны для цены) с типографским минусом. Тонкая
обёртка над `_fmt_money_signed`: срезаем ведущий «+».
"""
signed = _fmt_money_signed(value)
return signed[1:] if signed.startswith("+") else signed
def _fmt_pct(fraction: Any) -> str:
"""Доля 0.184 → «18.4%». Не-число → «—». PURE."""
if isinstance(fraction, bool) or not isinstance(fraction, int | float):
@ -440,20 +486,35 @@ def _build_zoning(result: dict[str, Any]) -> str:
def _build_zouit(result: dict[str, Any]) -> str:
"""ЗОУИТ-ограничения: сводка + список пересечений (тип / № границы / покрытие)."""
encumbrance = _as_dict(result.get("encumbrance"))
overlaps = _as_list(result.get("nspd_zouit_overlaps"))
overlaps = [ov for ov in _as_list(result.get("nspd_zouit_overlaps")) if isinstance(ov, dict)]
has_zouit = encumbrance.get("has_zouit")
zouit_count = encumbrance.get("zouit_count")
zouit_types = _as_list(encumbrance.get("zouit_types"))
# Противоречие: encumbrance говорит «нет ЗОУИТ», но НСПД-пересечения ЕСТЬ. Доверяем
# фактическим пересечениям (более свежий геослой) — иначе сводка «нет / 0» врёт под
# таблицей с реальными строками. Типы/кол-во достаём из самих overlaps.
if not has_zouit and overlaps:
has_zouit = "да (по данным НСПД)"
if zouit_count in (None, 0):
zouit_count = len(overlaps)
if not zouit_types:
zouit_types = [
str(t)
for t in dict.fromkeys(ov.get("type_zone") or ov.get("name") for ov in overlaps)
if t not in (None, "")
]
summary_pairs: list[tuple[str, Any]] = [
("Есть ЗОУИТ", encumbrance.get("has_zouit")),
("Кол-во типов ЗОУИТ", encumbrance.get("zouit_count")),
("Есть ЗОУИТ", has_zouit),
("Кол-во типов ЗОУИТ", zouit_count),
]
zouit_types = _as_list(encumbrance.get("zouit_types"))
if zouit_types:
summary_pairs.append(("Типы", ", ".join(str(t) for t in zouit_types)))
rows: list[list[Any]] = []
for ov in overlaps:
if not isinstance(ov, dict):
continue
for ov in overlaps: # already filtered to dicts above
coverage = ov.get("coverage_pct")
coverage_str = _fmt_pct(coverage) if isinstance(coverage, int | float) else _DASH
rows.append(
@ -674,19 +735,24 @@ def _build_engineering_nearby(result: dict[str, Any]) -> str:
"""Инженерные сооружения НСПД рядом (name / назначение / расстояние из raw_props)."""
items = _as_list(result.get("nspd_engineering_nearby"))
rows: list[list[Any]] = []
any_item = False
for it in items:
if not isinstance(it, dict):
continue
any_item = True
raw = _as_dict(it.get("raw_props"))
name = it.get("name") or raw.get("params_name") or raw.get("cad_number")
purpose = it.get("type") or raw.get("params_purpose")
# И имя, И назначение пустые → строка-призрак («— — 7»), скипаем.
if name in (None, "") and purpose in (None, ""):
continue
rows.append([name, purpose, _fmt_int_ru(it.get("distance_m"))])
if not rows:
# Секции вообще нет в payload → не рисуем заголовок. Есть записи, но все пустые →
# честное «нет данных» под заголовком (а не пустой невидимый блок).
if not any_item:
return ""
return (
f"<h3>Инженерные сооружения рядом (НСПД)</h3>"
f"{_data_table(['Сооружение', 'Назначение', 'Расстояние, м'], rows)}"
)
table = _data_table(["Сооружение", "Назначение", "Расстояние, м"], rows)
return f"<h3>Инженерные сооружения рядом (НСПД)</h3>{table}"
def _build_alternatives(result: dict[str, Any]) -> str:
@ -749,6 +815,96 @@ def _build_alternatives(result: dict[str, Any]) -> str:
"""
def _truncate_name(value: Any, limit: int = _SYSTEM_NAME_MAX) -> Any:
"""Усечь длинное бюрократическое имя системы до `limit` символов с «…». PURE.
Не-строка / короткая строка как есть (форматтер `_fmt`/`_esc` доработает). Режем по
границе символов, добавляя одноточечное многоточие (U+2026), чтобы длинные имена вроде
«Централизованная система теплоснабжения муниципального образования » не разваливали
вёрстку печатной таблицы.
"""
if not isinstance(value, str) or len(value) <= limit:
return value
return value[: limit - 1].rstrip() + ""
def _is_non_ekb(*fields: Any) -> bool:
"""True, если В ЛЮБОМ из полей есть ЯВНЫЙ не-ЕКБ городской маркер (Ирбит/Тавда/…). PURE.
Организация/система из другого города области «шум» для отчёта по участку ЕКБ.
Нормализуем ёе и регистр; НЕ хардкодим полный справочник городов (только явные
крупные маркеры) прочее схлопывается капом, а не этим фильтром.
"""
haystack = " ".join(str(f) for f in fields if f).lower().replace("ё", "е")
if not haystack:
return False
return any(marker in haystack for marker in _NON_EKB_MARKERS)
def _mentions_ekb(*fields: Any) -> bool:
"""True, если хотя бы одно поле явно упоминает Екатеринбург (ё-нормализовано). PURE."""
haystack = " ".join(str(f) for f in fields if f).lower().replace("ё", "е")
return _EKB_MARKER in haystack
def _reserve_num(value: Any) -> float:
"""Резерв как float для сортировки/суммы; не-число → 0.0 (нейтрально). PURE."""
if isinstance(value, bool) or not isinstance(value, int | float):
return 0.0
return float(value)
def _capacity_rows_capped(
items: list[Any],
*,
reserve_key: str,
name_fields: tuple[str, ...],
row_cap: int,
row_builder: Any,
) -> tuple[list[list[Any]], int]:
"""Отфильтровать областной шум + капнуть строки резерв-таблицы (тепло/вода). PURE.
Порядок честности:
1. Дефицитные (резерв < 0) ЕКБ-системы ВСЕГДА видимы, под кап не попадают.
2. Остальные ЕКБ (или неопределённые по городу) сортируются по резерву , берётся
топ до заполнения `row_cap` (с учётом уже показанных дефицитных).
3. Явно не-ЕКБ (Ирбит/Тавда/) и хвост сверх капа схлопываются в счётчик `hidden`.
Args:
items: сырые dict-строки (`heat.systems` / `cap.water`).
reserve_key: ключ резерва в dict (`reserve_gcal_h` / `reserve_thousand_m3_day`).
name_fields: ключи, по которым определяем город (org / system_name).
row_cap: максимум видимых строк.
row_builder: `dict -> list[cell]` построитель ячеек строки данных.
Returns:
`(visible_rows, hidden_count)` строки для таблицы + сколько схлопнуто.
"""
dicts = [it for it in items if isinstance(it, dict)]
deficit: list[dict[str, Any]] = []
surplus_ekb: list[dict[str, Any]] = []
hidden = 0
for it in dicts:
names = tuple(it.get(f) for f in name_fields)
reserve = _reserve_num(it.get(reserve_key))
# Дефицит ЕКБ/неопределённых — всегда показываем (даже если город не-ЕКБ:
# отрицательный резерв рядом — сигнал, честнее показать, чем спрятать).
if reserve < 0 and not _is_non_ekb(*names):
deficit.append(it)
continue
if _is_non_ekb(*names) and not _mentions_ekb(*names):
hidden += 1
continue
surplus_ekb.append(it)
surplus_ekb.sort(key=lambda d: _reserve_num(d.get(reserve_key)), reverse=True)
remaining = max(row_cap - len(deficit), 0)
shown_surplus = surplus_ekb[:remaining]
hidden += len(surplus_ekb) - len(shown_surplus)
visible = [row_builder(it) for it in (*deficit, *shown_surplus)]
return visible, hidden
def _build_connection_capacity(cap: dict[str, Any] | None) -> str:
"""Ресурсные резервы (ЦП/вода/газ/тепло) + сети рядом из connection-capacity (#2259 PR-D).
@ -781,12 +937,22 @@ def _build_connection_capacity(cap: dict[str, Any] | None) -> str:
if pairs:
blocks.append("<h3>Электроснабжение — свободная мощность</h3>" + _kv_table(pairs))
# Вода: строки резервов ЦСВ/ЦСК за последний период.
water_rows = [
[w.get("system_name"), _fmt(w.get("reserve_thousand_m3_day")), w.get("period")]
for w in _as_list(cap.get("water"))
if isinstance(w, dict)
]
# Вода: строки резервов ЦСВ/ЦСК за последний период (ЕКБ-системы + кап ~25).
water_rows, water_hidden = _capacity_rows_capped(
_as_list(cap.get("water")),
reserve_key="reserve_thousand_m3_day",
name_fields=("system_name", "org"),
row_cap=_WATER_ROW_CAP,
row_builder=lambda w: [
_truncate_name(w.get("system_name")),
_fmt(w.get("reserve_thousand_m3_day")),
w.get("period"),
],
)
if water_hidden:
water_rows.append(
[f"… и ещё {water_hidden} систем (полный список в веб-версии §3)", _DASH, _DASH]
)
if water_rows:
blocks.append(
"<h3>Водоснабжение/водоотведение — резервы</h3>"
@ -810,17 +976,47 @@ def _build_connection_capacity(cap: dict[str, Any] | None) -> str:
+ _data_table(["ГРС", "Свободно, тыс. м³/ч", "Свободно, %"], gas_rows)
)
# Тепло: резервы систем теплоснабжения.
# Тепло: резервы систем теплоснабжения. heat_system_reserves несёт ВСЕ ~56 систем
# области — сначала агрегат по ЕКБ (сумма резервов ЕКБ-систем), затем топ-15 + кап.
heat = _as_dict(cap.get("heat"))
heat_rows = [
[h.get("org"), h.get("system_name"), _fmt(h.get("reserve_gcal_h")), h.get("period")]
for h in _as_list(heat.get("systems"))
if isinstance(h, dict)
]
if heat_rows:
heat_systems = [h for h in _as_list(heat.get("systems")) if isinstance(h, dict)]
if heat_systems:
ekb_systems = [
h for h in heat_systems if not _is_non_ekb(h.get("org"), h.get("system_name"))
]
ekb_total = sum(_reserve_num(h.get("reserve_gcal_h")) for h in ekb_systems)
heat_rows, heat_hidden = _capacity_rows_capped(
heat_systems,
reserve_key="reserve_gcal_h",
name_fields=("org", "system_name"),
row_cap=_HEAT_ROW_CAP,
row_builder=lambda h: [
_truncate_name(h.get("org")),
_truncate_name(h.get("system_name")),
_fmt(h.get("reserve_gcal_h")),
h.get("period"),
],
)
# Агрегатная строка «Суммарно (ЕКБ)» сверху.
agg_row = [
"Суммарно (ЕКБ)",
f"{len(ekb_systems)} систем",
_fmt(round(ekb_total, 1)),
_DASH,
]
rows: list[list[Any]] = [agg_row, *heat_rows]
if heat_hidden:
rows.append(
[
f"… и ещё {heat_hidden} систем (полный список в веб-версии §3)",
_DASH,
_DASH,
_DASH,
]
)
blocks.append(
"<h3>Теплоснабжение — резервы систем</h3>"
+ _data_table(["Организация", "Система", "Резерв, Гкал/ч", "Период"], heat_rows)
+ _data_table(["Организация", "Система", "Резерв, Гкал/ч", "Период"], rows)
)
# Позитив-разрез: сетевые охранные зоны рядом (где физически проходит сеть).
@ -914,7 +1110,9 @@ def _build_market_metrics(forecast: dict[str, Any]) -> str:
("Темп продаж (velocity), ед./мес", metrics.get("unit_velocity")),
("Темп продаж (площадь), м²/мес", metrics.get("area_velocity")),
("Окно расчёта, мес", metrics.get("window_months")),
("Ставка абсорбции", metrics.get("absorption_rate")),
# absorption_rate — доля стока, продаваемая в месяц (0.0112). Как «0.01» это
# нечитаемо → проценты («1.1%»). Темп в штуках уже есть в unit_velocity выше.
("Ставка абсорбции (в мес.)", _fmt_pct(metrics.get("absorption_rate"))),
("Месяцев запаса (months of supply)", metrics.get("months_of_supply")),
("Индекс затоварки (overstock)", metrics.get("overstock_index")),
("Sell-through, %", metrics.get("sell_through_pct")),
@ -923,22 +1121,70 @@ def _build_market_metrics(forecast: dict[str, Any]) -> str:
return _kv_table(pairs)
def _competitor_lots(c: dict[str, Any]) -> int:
"""Число лотов конкурента как int; не-число / None → 0. PURE."""
n = c.get("flat_count")
if isinstance(n, bool) or not isinstance(n, int | float):
return 0
return int(n)
def _build_market_competitors(forecast: dict[str, Any]) -> str:
"""Конкуренты рынка сейчас: ЖК / девелопер / класс / расстояние / лотов."""
"""Конкуренты рынка сейчас: ЖК / девелопер / класс / расстояние / лотов.
Прод-payload несёт по строке НА КОРПУС: один ЖК с 6 корпусами 6 строк, а безымянные
записи с 0 лотов мусор. Схлопываем: (а) строки без имени И с 0 лотов скипаем;
(б) группируем по (имя, девелопер) ближайшая дистанция, лоты суммой, класс/девелопер
от первого, «(K корпусов)» в имени при K>1. Порядок групп по первому появлению.
"""
market_now = _fc_as_dict(forecast.get("market_now"))
competitors = _as_list(market_now.get("competitors"))
competitors = [c for c in _as_list(market_now.get("competitors")) if isinstance(c, dict)]
groups: dict[tuple[str, str], dict[str, Any]] = {}
order: list[tuple[str, str]] = []
for c in competitors:
name = c.get("comm_name")
lots = _competitor_lots(c)
# Безымянная запись без лотов — мусор (пустые «—» корпуса-призраки).
if name in (None, "") and lots == 0:
continue
key = (str(name or ""), str(c.get("dev_name") or ""))
dist = c.get("distance_m")
dist_val = (
float(dist) if isinstance(dist, int | float) and not isinstance(dist, bool) else None
)
if key not in groups:
order.append(key)
groups[key] = {
"name": name,
"dev_name": c.get("dev_name"),
"obj_class": c.get("obj_class"),
"distance_m": dist_val,
"lots": lots,
"corpus": 1,
}
else:
g = groups[key]
g["lots"] += lots
g["corpus"] += 1
if dist_val is not None and (g["distance_m"] is None or dist_val < g["distance_m"]):
g["distance_m"] = dist_val
if g["obj_class"] in (None, "") and c.get("obj_class"):
g["obj_class"] = c.get("obj_class")
rows: list[list[Any]] = []
for c in competitors:
if not isinstance(c, dict):
continue
for key in order:
g = groups[key]
name = g["name"]
if g["corpus"] > 1:
name = f"{_fmt(name)} ({g['corpus']} корпусов)"
rows.append(
[
c.get("comm_name"),
c.get("dev_name"),
c.get("obj_class"),
_fmt_int_ru(c.get("distance_m")),
_fmt_int_ru(c.get("flat_count")),
name,
g["dev_name"],
g["obj_class"],
_fmt_int_ru(g["distance_m"]),
_fmt_int_ru(g["lots"]),
]
)
return _data_table(["ЖК", "Девелопер", "Класс", "Расстояние, м", "Лотов"], rows)
@ -949,19 +1195,33 @@ def _build_market_coverage(forecast: dict[str, Any]) -> str:
confidence = _fc_as_dict(forecast.get("confidence"))
factors = _fc_as_dict(confidence.get("factors"))
rows: list[list[Any]] = []
# «Комментарий» почти всегда дублирует «Фактор» (label==note, либо note ⊃ label) —
# тогда две колонки = визуальный шум. Показываем 3-ю колонку ТОЛЬКО если хоть у одного
# фактора комментарий несёт что-то сверх метки; иначе схлопываем в «Фактор/уровень».
parsed: list[tuple[Any, Any, Any]] = []
for _key, payload in factors.items():
data = _fc_as_dict(payload)
if not data:
continue
rows.append(
[
data.get("label") or data.get("note"),
_fc_level_ru(data.get("level")),
data.get("note"),
]
)
coverage_table = _data_table(["Фактор", "Уровень", "Комментарий"], rows)
label = data.get("label") or data.get("note")
note = data.get("note")
parsed.append((label, _fc_level_ru(data.get("level")), note))
def _note_adds_info(label: Any, note: Any) -> bool:
if not isinstance(note, str) or note == "":
return False
if not isinstance(label, str):
return True
return note.strip() != label.strip() and label.strip() not in note
show_note = any(_note_adds_info(label, note) for label, _level, note in parsed)
if show_note:
rows = [[label, level, note] for label, level, note in parsed]
coverage_table = _data_table(["Фактор", "Уровень", "Комментарий"], rows)
else:
# Комментарий ничего не добавляет → две колонки «Фактор» + «Уровень».
rows = [[label, level] for label, level, _note in parsed]
coverage_table = _data_table(["Фактор", "Уровень"], rows)
level_pairs: list[tuple[str, Any]] = [
("Итоговая уверенность отчёта", _fc_level_ru(confidence.get("level"))),
@ -1098,7 +1358,7 @@ def _build_market_affordability(forecast: dict[str, Any]) -> str:
return ""
pairs: list[tuple[str, Any]] = [
("Рыночная цена, ₽/м²", _fmt_int_ru(detail.get("price_per_m2"))),
("Средний чек лота, ₽", _fmt_money_signed(detail.get("avg_ticket_rub"))),
("Средний чек лота, ₽", _fmt_money(detail.get("avg_ticket_rub"))),
("Референс-площадь, м²", detail.get("ref_area_m2")),
("Индекс избытка предложения", detail.get("oversupply_risk")),
]
@ -1276,9 +1536,7 @@ def _build_concept_program(variant: dict[str, Any]) -> str:
if not order:
return ""
rows = [[stype, floors, groups[(stype, floors)]] for stype, floors in order]
return (
"<h3>Программа застройки</h3>" f"{_data_table(['Тип дома', 'Этажность', 'Секций'], rows)}"
)
return f"<h3>Программа застройки</h3>{_data_table(['Тип дома', 'Этажность', 'Секций'], rows)}"
def _build_concept_variant(variant: dict[str, Any]) -> str:

View file

@ -93,6 +93,43 @@ def render_full_report_pdf(html: str) -> bytes:
return pdf_bytes
def _largest_polygon_geojson(geom: dict[str, Any]) -> dict[str, Any]:
"""MultiPolygon → GeoJSON крупнейшего полигона-контура; Polygon и прочее — как есть.
Многоконтурный участок приходит как MultiPolygon, а concept-стек (`parse_parcel` +
`_parcel_centroid_wkt`) принимает только Polygon `ParcelGeometryError`. Берём
контур с максимальной площадью (тот же приём, что generative-геометрия применяет к
buildable-мультиполигону после буфера, geometry.py:263-265). Любой сбой парсинга /
не-MultiPolygon возвращаем geom без изменений (best-effort, не роняем концепцию).
Args:
geom: GeoJSON-геометрия участка (Polygon / MultiPolygon / Feature-обёртка).
Returns:
GeoJSON Polygon крупнейшего контура (если вход был MultiPolygon), иначе `geom`.
"""
geom_dict = geom.get("geometry") if geom.get("type") == "Feature" else geom
if not isinstance(geom_dict, dict) or geom_dict.get("type") != "MultiPolygon":
return geom
try:
from shapely.geometry import mapping, shape
multi = shape(geom_dict)
if multi.is_empty or not hasattr(multi, "geoms"):
return geom
largest = max(multi.geoms, key=lambda g: g.area)
logger.info(
"build_full_report: multi-contour участок — взят крупнейший контур из %d",
len(list(multi.geoms)),
)
return dict(mapping(largest))
except Exception:
# Вырожденная/битая геометрия — отдаём как есть, concept-стек сам решит (best-effort).
logger.exception("build_full_report: не удалось выделить крупнейший контур MultiPolygon")
return geom
def _generate_concept_result(db: Session, analyze: dict[str, Any]) -> dict[str, Any] | None:
"""Сгенерировать концепцию server-side как это делает POST /concepts (best-effort).
@ -115,6 +152,13 @@ def _generate_concept_result(db: Session, analyze: dict[str, Any]) -> dict[str,
logger.info("build_full_report: analyze-payload без geom_geojson → §7-концепция пропущена")
return None
# Многоконтурный участок → geom = MultiPolygon, а concept-стек (parse_parcel +
# _parcel_centroid_wkt через _parse_polygon) принимает ТОЛЬКО Polygon и роняет
# ParcelGeometryError("expected Polygon, got MultiPolygon"). Берём крупнейший контур —
# ровно как generative-геометрия после буфера (geometry.py:263-265). Иначе §7 и
# market-price молча деградируют на любом мультиконтуре.
geom = _largest_polygon_geojson(geom)
try:
# Lazy import — тяжёлый generative-стек не нужен на module-load; concepts-хелперы
# цены живут в API-слое (он знает БД), переиспользуем ИМЕННО их (single source).

View file

@ -386,6 +386,120 @@ def test_connection_capacity_absent_when_none() -> None:
assert "Газоснабжение — свободная мощность ГРС" not in html2
# ── FIX 5: противоречие ЗОУИТ-блоков (has_zouit falsy, но overlaps есть) ─────────
def test_zouit_reconciles_falsy_flag_with_overlaps() -> None:
"""has_zouit=нет + zouit_count=0, но nspd_zouit_overlaps непуст → «да (по данным НСПД)»."""
result = {
"encumbrance": {"has_zouit": False, "zouit_count": 0},
"nspd_zouit_overlaps": [
{"type_zone": "Охранная зона ЛЭП", "reg_numb_border": "66:41-1", "coverage_pct": 0.3},
{"type_zone": "Санитарная зона", "reg_numb_border": "66:41-2", "coverage_pct": 0.1},
],
}
html = build_full_report_html_part_a(result, cad="X")
assert "да (по данным НСПД)" in html
# Кол-во типов = число пересечений; типы восстановлены из overlaps.
assert "Охранная зона ЛЭП" in html
assert "Санитарная зона" in html
def test_zouit_flag_true_unchanged() -> None:
"""has_zouit=да → сводка не переписывается (FIX 5 трогает только противоречие)."""
result = {
"encumbrance": {"has_zouit": True, "zouit_count": 1, "zouit_types": ["Охранная"]},
"nspd_zouit_overlaps": [{"type_zone": "Охранная", "coverage_pct": 0.5}],
}
html = build_full_report_html_part_a(result, cad="X")
assert "да (по данным НСПД)" not in html
# ── FIX 6a: НСПД-сооружения — пропуск строк-призраков ────────────────────────────
def test_engineering_nearby_skips_ghost_rows() -> None:
"""Строки где И имя, И назначение пустые («— — 7») скипаются (FIX 6a)."""
result = {
"nspd_engineering_nearby": [
{"name": None, "type": None, "distance_m": 7, "raw_props": {}}, # призрак
{"name": "Трубопровод", "type": "тепло", "distance_m": 40, "raw_props": {}},
]
}
html = build_full_report_html_part_a(result, cad="X")
assert "Трубопровод" in html
# Заголовок есть (записи были), призрак-строки нет.
assert "Инженерные сооружения рядом (НСПД)" in html
def test_engineering_nearby_all_empty_shows_no_data() -> None:
"""Все записи пустые → заголовок + «нет данных» (не пустой невидимый блок, FIX 6a)."""
result = {
"nspd_engineering_nearby": [
{"name": None, "type": None, "distance_m": 7, "raw_props": {}},
{"name": "", "type": "", "distance_m": 9, "raw_props": {}},
]
}
html = build_full_report_html_part_a(result, cad="X")
assert "Инженерные сооружения рядом (НСПД)" in html
assert _NO_DATA in html
# ── FIX 3: тепло/вода — фильтр областного шума + агрегат ЕКБ + кап ───────────────
def _heat_systems_province() -> dict[str, Any]:
"""56-строчная тепло-выборка: горсть ЕКБ + много не-ЕКБ (Ирбит/Тавда/…)."""
systems: list[dict[str, Any]] = [
{"org": "ЕТК", "system_name": "СТ Екатеринбург №54", "reserve_gcal_h": 3.1},
{"org": "Т Плюс", "system_name": "СТ Екатеринбург ТЭЦ", "reserve_gcal_h": 12.0},
# Дефицит ЕКБ — под кап НЕ прячем.
{"org": "ЕТК", "system_name": "СТ Екатеринбург №7", "reserve_gcal_h": -2.5},
]
# Много областного шума (не-ЕКБ).
for city in ("Ирбит", "Тавда", "Красноуфимск", "Первоуральск", "Лесной", "Нижняя Тура"):
systems.append(
{"org": "Регионгаз-инвест", "system_name": f"СТ {city} №1", "reserve_gcal_h": 1.0}
)
return {"systems": systems, "total_reserve_gcal_h": 100.0}
def test_heat_table_aggregates_ekb_and_hides_province_noise() -> None:
"""Тепло: агрегат «Суммарно (ЕКБ)» + не-ЕКБ схлопнуты в «… и ещё K систем» (FIX 3)."""
cap = {"heat": _heat_systems_province()}
html = build_full_report_html_part_a(_synthetic_result(), cad="X", connection_capacity=cap)
assert "Суммарно (ЕКБ)" in html
# ЕКБ-системы видны (в т.ч. дефицит), не-ЕКБ схлопнуты.
assert "Екатеринбург №54" in html
assert "Екатеринбург №7" in html # дефицит НЕ спрятан
assert "и ещё" in html # областной шум схлопнут
# Не-ЕКБ город не должен фигурировать строкой.
assert "Ирбит" not in html
def test_heat_deficit_not_hidden_under_cap() -> None:
"""Честность: отрицательный резерв ЕКБ виден даже при переполнении капа (FIX 3)."""
systems = [{"org": "ЕТК", "system_name": "СТ Екатеринбург дефицит", "reserve_gcal_h": -5.0}]
# Забиваем кап 20 профицитными ЕКБ-системами.
for i in range(20):
systems.append(
{"org": "ЕТК", "system_name": f"СТ Екатеринбург плюс {i}", "reserve_gcal_h": float(i)}
)
html = build_full_report_html_part_a(
_synthetic_result(), cad="X", connection_capacity={"heat": {"systems": systems}}
)
assert "Екатеринбург дефицит" in html
def test_long_system_name_truncated() -> None:
"""Длинное бюрократическое имя системы усечено до ~120 символов с «…» (FIX 3)."""
long_name = "Централизованная система теплоснабжения " * 6 # >200 симв.
cap = {"heat": {"systems": [{"org": "ЕТК Екатеринбург", "system_name": long_name}]}}
html = build_full_report_html_part_a(_synthetic_result(), cad="X", connection_capacity=cap)
assert "" in html
assert long_name not in html # полное имя не попало целиком
# ── 9. Каркас документа: титул / оглавление / опциональный Part B ────────────────

View file

@ -567,3 +567,113 @@ def test_document_frame_with_both_parts() -> None:
assert f'href="#{_ANCHOR_S7}"' in doc
# Ровно один общий <style> (не дублируется на Part B).
assert doc.count("<style>") == 1
# ── FIX 4: конкуренты — дубли-корпуса + безымянные с 0 лотов ─────────────────────
def _forecast_with_competitors(competitors: list[dict[str, Any]]) -> dict[str, Any]:
"""Форсайт только с блоком конкурентов (остальное пустое → graceful)."""
return {"market_now": {"competitors": competitors}}
def test_competitors_group_corpuses_and_sum_lots() -> None:
"""6 корпусов «ЖК ОК: Премиум» → одна строка «(6 корпусов)», лоты суммой (FIX 4)."""
corpuses = [
{
"comm_name": "ЖК ОК: Премиум",
"dev_name": "Девелопер X",
"obj_class": "Премиум",
"distance_m": 1000 + i * 10,
"flat_count": 50,
}
for i in range(6)
]
html = build_full_report_html_part_b(_forecast_with_competitors(corpuses), None, cad="X")
assert "ЖК ОК: Премиум (6 корпусов)" in html
assert "300" in html # 6 × 50 лотов суммой
# Ближайшая дистанция группы — минимум (1000), с пробелом-разрядом «1 000».
assert "1 000" in html
def test_competitors_skip_nameless_zero_lots() -> None:
"""Безымянные строки с 0 лотов («—» Эфес) скипаются (FIX 4)."""
competitors = [
{"comm_name": None, "dev_name": "Эфес", "flat_count": 0, "distance_m": 500},
{"comm_name": None, "dev_name": "Эфес", "flat_count": 0, "distance_m": 600},
{"comm_name": "ЖК Реальный", "dev_name": "Дев", "flat_count": 120, "distance_m": 800},
]
html = build_full_report_html_part_b(_forecast_with_competitors(competitors), None, cad="X")
assert "ЖК Реальный" in html
assert "Эфес" not in html # безымянные с 0 лотов выброшены
def test_competitors_single_corpus_no_suffix() -> None:
"""Один корпус → без суффикса «(K корпусов)» (FIX 4)."""
competitors = [
{"comm_name": "ЖК Одиночка", "dev_name": "Дев", "flat_count": 40, "distance_m": 300}
]
html = build_full_report_html_part_b(_forecast_with_competitors(competitors), None, cad="X")
assert "ЖК Одиночка" in html
assert "корпусов" not in html
# ── FIX 6b: confidence-факторы — схлопывание дублирующей колонки «Комментарий» ────
def test_confidence_factors_collapse_duplicate_note_column() -> None:
"""note дублирует label у всех факторов → колонка «Комментарий» скрыта (FIX 6b)."""
forecast = {
"confidence": {
"level": "medium",
"factors": {
"a": {"label": "Сделки", "level": "high", "note": "Сделки"},
"b": {"label": "История", "level": "low", "note": "История"},
},
}
}
html = build_full_report_html_part_b(forecast, None, cad="X")
assert "Сделки" in html
assert "Комментарий" not in html # дублирующая колонка схлопнута
def test_confidence_factors_keep_note_when_informative() -> None:
"""note несёт данные сверх label → колонка «Комментарий» остаётся (FIX 6b)."""
forecast = {
"confidence": {
"level": "low",
"factors": {
"a": {"label": "Сделки", "level": "high", "note": "1388 сделок за 6 мес"},
},
}
}
html = build_full_report_html_part_b(forecast, None, cad="X")
assert "Комментарий" in html
assert "1388 сделок за 6 мес" in html
# ── FIX 6c/6d: абсорбция как %, средний чек без «+» ──────────────────────────────
def test_absorption_rate_shown_as_percent() -> None:
"""absorption_rate 0.0112 → «1.1%», не сырое «0.01» (FIX 6c)."""
forecast = {"market_now": {"market_metrics": {"absorption_rate": 0.0112}}}
html = build_full_report_html_part_b(forecast, None, cad="X")
assert "1.1%" in html
assert ">0.01<" not in html
def test_avg_ticket_has_no_leading_plus() -> None:
"""Средний чек лота — цена, не дельта: «8 млн ₽» без ведущего «+» (FIX 6d)."""
forecast = {
"scoring": {
"special_indices": {
"indices": {
"cost_of_error": {"detail": {"avg_ticket_rub": 8_000_000.0}},
}
}
}
}
html = build_full_report_html_part_b(forecast, None, cad="X")
assert "8 млн ₽" in html
assert "+8 млн ₽" not in html

View file

@ -258,6 +258,73 @@ def test_connection_capacity_failure_is_best_effort(monkeypatch: pytest.MonkeyPa
assert mod._get_connection_capacity(MagicMock(), _CAD) is None
_MULTIPOLYGON = {
"type": "MultiPolygon",
"coordinates": [
# Мелкий контур (площадь ~0.0001 в градусах²).
[[[60.60, 56.838], [60.61, 56.838], [60.61, 56.839], [60.60, 56.839], [60.60, 56.838]]],
# Крупный контур (площадь ~0.00007*0.007 — намеренно шире по обеим осям).
[[[60.62, 56.840], [60.70, 56.840], [60.70, 56.850], [60.62, 56.850], [60.62, 56.840]]],
],
}
def test_largest_polygon_passthrough_for_polygon() -> None:
"""Обычный Polygon → возвращается без изменений (FIX 1: только MultiPolygon трогаем)."""
assert mod._largest_polygon_geojson(_GEOJSON) is _GEOJSON
def test_largest_polygon_extracts_biggest_contour() -> None:
"""MultiPolygon → GeoJSON Polygon КРУПНЕЙШЕГО контура (FIX 1, эпик #2259)."""
out = mod._largest_polygon_geojson(_MULTIPOLYGON)
assert out["type"] == "Polygon"
# Крупнейший контур — второй (60.62..60.70 × 56.840..56.850): его первая вершина x≈60.62.
xs = [pt[0] for pt in out["coordinates"][0]]
assert min(xs) >= 60.62
def test_largest_polygon_unwraps_feature() -> None:
"""Feature-обёртка над MultiPolygon → крупнейший Polygon (FIX 1)."""
feature = {"type": "Feature", "geometry": _MULTIPOLYGON, "properties": {}}
out = mod._largest_polygon_geojson(feature)
assert out["type"] == "Polygon"
def test_largest_polygon_bad_geometry_returns_input() -> None:
"""Битая MultiPolygon-геометрия → best-effort вернуть вход как есть (FIX 1)."""
broken = {"type": "MultiPolygon", "coordinates": "не координаты"}
assert mod._largest_polygon_geojson(broken) is broken
def test_concept_multipolygon_does_not_crash(monkeypatch: pytest.MonkeyPatch) -> None:
"""Многоконтурный участок (MultiPolygon) НЕ роняет генерацию концепции (FIX 1).
Корень прод-бага: concept-стек (_parse_polygon / parse_parcel) принимал только
Polygon и падал `ParcelGeometryError: expected Polygon, got MultiPolygon`. После
FIX 1 geom нормализуется к крупнейшему контуру ДО передачи в стек generate
получает Polygon-payload и отрабатывает.
"""
analyze = {"geom_geojson": _MULTIPOLYGON, "egrn": {}}
captured: dict[str, Any] = {}
def _fake_generate(payload: Any, **kwargs: Any) -> list[Any]:
captured["geom_type"] = payload.parcel_geojson.get("type")
return []
monkeypatch.setattr("app.services.generative.geometry.generate", _fake_generate)
monkeypatch.setattr(
"app.api.v1.concepts._parcel_centroid_wkt", lambda payload: "POINT(60.66 56.845)"
)
monkeypatch.setattr(
"app.api.v1.concepts._lookup_market_price", lambda db, wkt: (None, "class_norm")
)
result = mod._generate_concept_result(MagicMock(), analyze)
# generate вызвана с уже-нормализованным Polygon (не MultiPolygon) → не упала.
assert captured["geom_type"] == "Polygon"
assert result == {"variants": []}
def test_cad_safe_whitelist() -> None:
"""_cad_safe: только `[0-9:]`, `:`→`_`; path-traversal/спецсимволы вырезаны (п.4)."""
assert mod._cad_safe("66:41:0204016:10") == "66_41_0204016_10"