gendesign/backend/app/services/exporters/full_report_html.py
bot-backend 880bac7738
All checks were successful
CI / changes (pull_request) Successful in 8s
CI Trade-In / changes (pull_request) Successful in 8s
CI Trade-In / backend-tests (pull_request) Has been skipped
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / frontend-tests (pull_request) Successful in 1m15s
CI / openapi-codegen-check (pull_request) Successful in 2m7s
CI / backend-tests (pull_request) Successful in 14m57s
feat(reports): Celery-джоба полного PDF-отчёта + кэш + 3 endpoint'а (#2259 PR-D)
Оркестратор build_full_report собирает §1-§7 + карты из analyze/forecast
ранов (реальная schema форсайта = "1.0"), best-effort дотягивает
connection-capacity (§3-резервы + сети рядом) и server-side концепцию
(поток POST /concepts: ConceptInput comfort → generate), рендерит
WeasyPrint'ом (локальный import — web-процесс native не тянет) на volume
/app/reports с метадата-раном report-pdf-1.0. Кэш по (analyze_run_id,
forecast_run_id) с проверкой существования файла (сирота → ре-рендер).
Запись атомарная (tmp+os.replace), имя файла из whitelist-санитизации
кад.номера. Celery task on-demand (time_limit 600), endpoint'ы POST
/{cad}/report (202/200) + status + download (дата из generated_at).

Deploy: bind-mount ./reports:/app/reports (backend+worker) + mkdir/chown
uid1000 в deploy.yml (docker создал бы root:root → PermissionError →
вечный building). reports/ в .gitignore. api-types регенерён (+219).

Refs #2259
2026-07-03 15:56:58 +05:00

1537 lines
75 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""PURE HTML-агрегатор полного PDF-отчёта ПТИЦА §1§3 + альтернативы (эпик #2259 PR-A).
Foundation-слой полного PDF-экспорта отчёта `/site-finder/analysis/{cad}`. Собирает
из ПЕРСИСТНУТОГО `analysis_runs.result` (schema `analyze-1.0`, loose dict ~90 ключей —
`AnalyzeResponse`, `extra="allow"`) единый печатный HTML-документ:
• §1 «Участок» — адрес / площадь / категория / ВРИ / градрегламент (ПЗЗ-зона, ТЭП)
/ ЗОУИТ-ограничения / gate-вердикт «можно ли строить МКД».
• §2 «Окружение» — транспорт (центр / метро) / шум / воздух / геотехника / гидрология
/ соседняя застройка.
• §3 «Сети» — инженерные сети рядом (`utilities`) + инж.сооружения НСПД + блок
альтернатив программы (`program_alternatives`, при отрицательном фин-вердикте).
Архитектура — зеркало `layout_tz_pdf.build_layout_tz_html`: PURE-сборка HTML (без
WeasyPrint / БД / сети), тестируется юнит-тестами БЕЗ native GTK/Pango. Нормализация
loose-словарей — по образцу `report_pdf._normalize`/`_fmt`/`_as_dict`/`_as_list`.
БЕЗОПАСНОСТЬ: ВСЕ динамические строки payload проходят `html.escape` (вход трактуем
как недоверенный — часть данных приходит из внешних источников: НСПД / OSM / геопортал).
GRACEFUL: любой ключ читается через `.get()` с дефолтом; пустая секция рисует «нет
данных» по-русски; агрегатор НИКОГДА не падает на частичном / пустом payload.
КАРТА: место под карту участка — плейсхолдер `{{MAP_PARCEL}}` (реальная подстановка в
PR-C). PR-A оставляет только маркер.
ГРАНИЦЫ PR-A: §4§7 (PR-B), карты (PR-C), live-дозапросы отсутствующих в persist-payload
слоёв — «Ресурсные резервы» ЦП/Водоканала и `nearby_network_zones` берутся ленивым
`connection-capacity`-эндпоинтом, ИХ В persist-payload НЕТ (PR-D).
"""
from __future__ import annotations
import html
import logging
from typing import Any
# §4§7 нормализуют forecast-словарь (schema §22-форсайта "1.0" = SiteFinderReport.
# as_dict()) теми же PURE-хелперами, что уже гоняет PDF-рендерер форсайта — импортируем,
# НЕ копируем (ревью PR-A уже отметило дублирование). WeasyPrint внутри report_pdf
# импортируется ЛОКАЛЬНО (в export_report_pdf), поэтому импорт модуля тут дешёвый и
# native-libs не тянет. Алиасим под _fc_ («forecast»), чтобы не спутать с локальными
# каркас-хелперами §1§3 (у них своя семантика _fmt: округление ₽ до млн против 3 знаков).
from app.services.exporters.report_pdf import _as_dict as _fc_as_dict
from app.services.exporters.report_pdf import _future_supply_pairs as _fc_future_supply_pairs
from app.services.exporters.report_pdf import _level_ru as _fc_level_ru
from app.services.exporters.report_pdf import _normalize as _fc_normalize
logger = logging.getLogger(__name__)
# ── Плейсхолдер карты (подставится в PR-C) ─────────────────────────────────────
MAP_PARCEL_PLACEHOLDER = "{{MAP_PARCEL}}"
# Плейсхолдер footprint-плана концепции §7 (реальная подстановка карты — PR-C).
MAP_CONCEPT_PLACEHOLDER = "{{MAP_CONCEPT}}"
# ── Микрокопия / заглушки ──────────────────────────────────────────────────────
_DASH = ""
_NO_DATA = "нет данных"
# Заголовки секций (якоря — id внутри) и титул документа.
_TITLE_DOC = "Отчёт по участку — Site Finder ПТИЦА"
_TITLE_S1 = "§1. Участок"
_TITLE_S2 = "§2. Окружение"
_TITLE_S3 = "§3. Инженерные сети"
_TITLE_ALT = "Как участок сходится (альтернативы программы)"
# id-якоря секций (оглавление + внутренние ссылки).
_ANCHOR_S1 = "section-1"
_ANCHOR_S2 = "section-2"
_ANCHOR_S3 = "section-3"
_ANCHOR_ALT = "section-alternatives"
# ── §4§7 (Part B): заголовки секций + якоря ────────────────────────────────────
_TITLE_S4 = "§4. Рынок"
_TITLE_S5 = "§5. Финансовая модель"
_TITLE_S6 = "§6. Риски и дефицит"
_TITLE_S7 = "§7. Концепция застройки"
_ANCHOR_S4 = "section-4"
_ANCHOR_S5 = "section-5"
_ANCHOR_S6 = "section-6"
_ANCHOR_S7 = "section-7"
# RU-метка стратегии генерации концепции (зеркало STRATEGY_LABELS в concept-api.ts).
# «program»-вариант каркас распознаёт по флагу и подписывает «Ваша программа».
_STRATEGY_RU: dict[str, str] = {
"max_area": "Макс. площадь",
"max_insolation": "Макс. инсоляция",
"balanced": "Баланс",
"program": "Ваша программа",
}
# RU-фраза источника цены продажи финмодели (зеркало priceSourceCaption в concept-api.ts).
_PRICE_SOURCE_FIN_RU: dict[str, str] = {
"objective_district_median": "рынок: медиана объявлений Objective по району",
"objective_geo_radius": "рынок: медиана новостроек в радиусе 3 км",
"district_reference": "рынок: справочная медиана района (нет свежей выборки Objective)",
"class_norm": "норматив класса (нет рыночных данных по участку)",
}
# RU-метки источников цены варианта программы (зеркало SectionAlternatives.tsx).
_PRICE_SOURCE_RU: dict[str, str] = {
"objective_district_median": "рынок: медиана объявлений Objective по району",
"objective_geo_radius": "рынок: медиана новостроек в радиусе 3 км",
"district_reference": "рынок: справочная медиана района",
"class_norm": "норматив класса (нет рыночных данных по участку)",
}
# RU-метки класса жилья / типа застройки (зеркало concept-api-лейблов фронта).
_HOUSING_CLASS_RU: dict[str, str] = {
"econom": "Эконом",
"comfort": "Комфорт",
"business": "Бизнес",
}
_DEVELOPMENT_TYPE_RU: dict[str, str] = {
"spot": "точечная",
"mid_rise": "среднеэтажная",
"high_rise": "высотная",
}
# ── Печатная типографика + page-break-правила WeasyPrint ────────────────────────
_CSS = """
@page {
size: A4;
margin: 18mm 16mm 20mm 16mm;
@bottom-center {
content: "Site Finder ПТИЦА · стр. " counter(page);
font-size: 7.5pt;
color: #9ca3af;
}
}
* { box-sizing: border-box; }
body {
font-family: 'DejaVu Sans', 'Helvetica', 'Arial', sans-serif;
font-size: 10pt;
color: #1a1d23;
margin: 0;
padding: 0;
}
h1 { font-size: 20pt; margin: 0 0 6pt 0; line-height: 1.2; }
h2 {
font-size: 14pt; margin: 0 0 8pt 0;
border-bottom: 1.5px solid #e6e8ec; padding-bottom: 4pt; color: #1d4ed8;
}
h3 { font-size: 11pt; margin: 12pt 0 6pt 0; color: #374151; }
p { margin: 4pt 0; }
/* Page-break: каждая содержательная секция стартует со свежей страницы, но
не рвётся внутри заголовка (WeasyPrint поддерживает break-before/inside). */
.section { margin-bottom: 16pt; break-inside: avoid-page; }
.section + .section { break-before: page; }
h2, h3 { break-after: avoid-page; }
table { break-inside: avoid-page; }
/* Титул */
.cover { padding: 6pt 0 12pt 0; border-bottom: 2px solid #1d4ed8; margin-bottom: 14pt; }
.cover-meta { font-size: 9pt; color: #5b6066; margin-top: 4pt; }
/* Оглавление */
.toc { margin: 0 0 16pt 0; break-after: page; }
.toc ul { list-style: none; padding: 0; margin: 6pt 0 0 0; }
.toc li { padding: 3pt 0; font-size: 10.5pt; }
/* Карта-плейсхолдер */
.map-placeholder {
margin: 8pt 0; padding: 40pt 12pt; text-align: center;
background: #fafbfc; border: 1px dashed #d1d5db; border-radius: 6pt;
color: #9ca3af; font-size: 9pt;
}
/* KV-таблица «метка → значение» */
.kv { width: 100%; border-collapse: collapse; margin-top: 4pt; }
.kv td { padding: 4pt 7pt; border-bottom: 1px solid #eef0f3; vertical-align: top; }
.kv td.k { font-weight: 700; color: #374151; width: 40%; }
.kv td.v { color: #1a1d23; }
/* Таблицы данных */
table.data { width: 100%; border-collapse: collapse; margin-top: 8pt; font-size: 9pt; }
table.data thead tr { background: #1d4ed8; }
table.data th {
padding: 5pt 7pt; text-align: left; font-weight: 700;
color: #fff; border-bottom: 1.5px solid #1e40af;
}
table.data td { padding: 5pt 7pt; border-bottom: 1px solid #e6e8ec; vertical-align: top; }
table.data tr:nth-child(even) td { background: #f9fafb; }
td.empty, .no-data { text-align: center; color: #9ca3af; font-style: italic; padding: 10pt; }
.no-data { padding: 6pt 0; }
/* Плашки-вердикты */
.verdict-block {
margin: 8pt 0; padding: 8pt 12pt; border-radius: 0 4pt 4pt 0; font-size: 9.5pt;
}
.verdict-ok { background: #dcfce7; border-left: 4px solid #0a7a3a; color: #05612c; }
.verdict-no { background: #fee2e2; border-left: 4px solid #b3261e; color: #8a1b15; }
.caveat {
margin: 8pt 0; padding: 8pt 12pt; border-radius: 0 4pt 4pt 0;
background: #fef3c7; border-left: 4px solid #9a6700; color: #7a5209; font-size: 9pt;
}
/* Карточки альтернатив */
.alt-card {
border: 1px solid #e6e8ec; border-radius: 8pt; padding: 10pt 12pt; margin: 8pt 0;
}
.alt-title { font-size: 11pt; font-weight: 700; color: #111; }
.alt-npv { font-size: 15pt; font-weight: 700; margin: 4pt 0; }
.alt-npv.pos { color: #0a7a3a; }
.alt-npv.neg { color: #b3261e; }
.alt-meta { font-size: 9pt; color: #5b6066; }
.footer {
margin-top: 16pt; padding-top: 8pt;
border-top: 1px solid #e6e8ec; font-size: 8pt; color: #9ca3af;
}
"""
# ──────────────────────────────────────────────────────────────────────────────
# Нормализация входа + микро-хелперы форматирования. Все PURE, graceful.
# ──────────────────────────────────────────────────────────────────────────────
def _as_dict(value: Any) -> dict[str, Any]:
"""Вернуть dict как есть, иначе пустой dict (defensive чтение секций). PURE."""
return value if isinstance(value, dict) else {}
def _as_list(value: Any) -> list[Any]:
"""Вернуть list как есть, иначе пустой list (defensive чтение секций). PURE."""
return value if isinstance(value, list) else []
def _fmt(value: Any) -> str:
"""Привести значение к тексту ячейки: None → «—», float → аккуратно, str — как есть.
Float с целым значением (78414.0) печатаем без хвоста «.0». Прочие float —
до 2 знаков, лишние нули убираем. Bool → «да»/«нет». Возврат — ВСЕГДА str
(его экранирует `_esc`). PURE.
"""
if value is None:
return _DASH
if isinstance(value, bool):
return "да" if value else "нет"
if isinstance(value, float):
if value.is_integer():
return str(int(value))
return f"{value:.2f}".rstrip("0").rstrip(".")
if isinstance(value, str):
return value if value != "" else _DASH
return str(value)
def _esc(value: Any) -> str:
"""Привести значение к ячейке (`_fmt`) и ЭКРАНИРОВАТЬ для HTML. PURE.
Единая точка вставки динамических данных в разметку: всё, что попадает в HTML,
проходит `html.escape` (вход недоверенный — см. модульный docstring).
"""
return html.escape(_fmt(value))
def _fmt_int_ru(value: Any) -> str:
"""Целое число с пробелами-разрядами («77 656»), иначе «—». PURE."""
if isinstance(value, bool) or value is None:
return _DASH
if isinstance(value, int | float):
return f"{int(value):,}".replace(",", " ")
return _DASH
def _fmt_money_signed(value: Any) -> str:
"""Деньги округлением до млн со знаком: «+42 млн ₽» / «216 млн ₽». PURE.
Ниже 1 млн — точная сумма с пробелами-разрядами. Минус — типографский U+2212.
None / не-число → «—».
"""
if isinstance(value, bool) or not isinstance(value, int | float):
return _DASH
num = float(value)
sign = "+" if num > 0 else "" if num < 0 else ""
magnitude = abs(num)
if magnitude >= 1e9:
return f"{sign}{magnitude / 1e9:.1f} млрд ₽"
if magnitude >= 1e6:
return f"{sign}{round(magnitude / 1e6)} млн ₽"
return f"{sign}{f'{round(magnitude):,}'.replace(',', ' ')}"
def _fmt_pct(fraction: Any) -> str:
"""Доля 0.184 → «18.4%». Не-число → «—». PURE."""
if isinstance(fraction, bool) or not isinstance(fraction, int | float):
return _DASH
value = f"{fraction * 100:.1f}"
if value.startswith("-"):
return "" + value[1:] + "%"
return value + "%"
def _fmt_pct_raw(value: Any) -> str:
"""Процент, УЖЕ выраженный в единицах 0100 (напр. free_capacity_pct газовой ГРС).
В отличие от `_fmt_pct` (принимает долю 0..1) — здесь значение уже проценты, НЕ
домножаем на 100. 18.4 → «18.4%». Не-число → «—». PURE.
"""
if isinstance(value, bool) or not isinstance(value, int | float):
return _DASH
text_val = f"{float(value):.1f}"
if text_val.startswith("-"):
return "" + text_val[1:] + "%"
return text_val + "%"
def _housing_class_ru(value: Any) -> str:
"""RU-метка класса жилья, иначе значение как есть (str). PURE."""
if isinstance(value, str) and value in _HOUSING_CLASS_RU:
return _HOUSING_CLASS_RU[value]
return _fmt(value)
def _development_type_ru(value: Any) -> str:
"""RU-метка типа застройки, иначе значение как есть (str). PURE."""
if isinstance(value, str) and value in _DEVELOPMENT_TYPE_RU:
return _DEVELOPMENT_TYPE_RU[value]
return _fmt(value)
def _price_source_ru(value: Any) -> str:
"""RU-фраза источника цены варианта, иначе значение как есть. PURE."""
if isinstance(value, str) and value in _PRICE_SOURCE_RU:
return _PRICE_SOURCE_RU[value]
return _fmt(value)
# ──────────────────────────────────────────────────────────────────────────────
# Низкоуровневые HTML-помощники (kv-строки / таблицы / заглушки). Все PURE.
# ──────────────────────────────────────────────────────────────────────────────
def _no_data() -> str:
"""Блок-заглушка «нет данных» для пустой секции (graceful). PURE."""
return f'<div class="no-data">{_NO_DATA}</div>'
def _kv_row(label: str, value: Any) -> str:
"""Строка «метка → значение» (оба экранированы). `label` — статичная RU-метка."""
return f'<tr><td class="k">{html.escape(label)}</td><td class="v">{_esc(value)}</td></tr>'
def _kv_table(pairs: list[tuple[str, Any]]) -> str:
"""Таблица «метка → значение» из списка пар. Пустой список → «нет данных». PURE."""
if not pairs:
return _no_data()
rows = "".join(_kv_row(label, value) for label, value in pairs)
return f'<table class="kv">{rows}</table>'
def _data_table(headers: list[str], rows: list[list[Any]]) -> str:
"""Таблица: шапка (статичные RU-метки) + строки данных (через `_esc`). PURE.
Пустой `rows` → строка-заглушка «нет данных» под шапкой (graceful). Все ячейки
данных экранируются.
"""
head = "".join(f"<th>{html.escape(h)}</th>" for h in headers)
if not rows:
body = f'<tr><td class="empty" colspan="{len(headers)}">{_NO_DATA}</td></tr>'
else:
body = "".join(
"<tr>" + "".join(f"<td>{_esc(cell)}</td>" for cell in row) + "</tr>" for row in rows
)
return f'<table class="data"><thead><tr>{head}</tr></thead><tbody>{body}</tbody></table>'
# ──────────────────────────────────────────────────────────────────────────────
# §1 «Участок»: адрес / площадь / категория / ВРИ / градрегламент / ЗОУИТ / gate.
# ──────────────────────────────────────────────────────────────────────────────
def _build_parcel_facts(result: dict[str, Any]) -> str:
"""KV-таблица кадастровых фактов: адрес, площадь, категория, ВРИ, статус, стоимость."""
egrn = _as_dict(result.get("egrn"))
parcel_meta = _as_dict(result.get("parcel_meta"))
geometry = _as_dict(result.get("geometry_suitability"))
address = egrn.get("address")
area_m2 = egrn.get("area_m2") or geometry.get("area_m2")
category = egrn.get("land_category") or parcel_meta.get("land_category")
vri = egrn.get("permitted_use_text") or parcel_meta.get("permitted_use")
cad_cost = egrn.get("cadastral_value_rub") or parcel_meta.get("cad_cost")
pairs: list[tuple[str, Any]] = [
("Адрес", address),
("Площадь, м²", _fmt_int_ru(area_m2) if area_m2 is not None else None),
("Категория земель", category),
("Разрешённое использование (ВРИ)", vri),
("Подвид", parcel_meta.get("land_subtype")),
("Статус", egrn.get("parcel_status")),
("Форма собственности", egrn.get("ownership_type")),
("Право", egrn.get("right_type")),
("Кадастровая стоимость, ₽", _fmt_int_ru(cad_cost) if cad_cost is not None else None),
]
# Убираем строки без значения, чтобы не заваливать таблицу «—».
pairs = [(k, v) for k, v in pairs if v not in (None, "")]
return _kv_table(pairs)
def _build_zoning(result: dict[str, Any]) -> str:
"""Градрегламент ПЗЗ: зона + ТЭП (FAR / этажность / высота / застройка) + ВРИ-список."""
nspd_zoning = _as_dict(result.get("nspd_zoning"))
if not nspd_zoning:
# Fallback: старый per-parcel zoning (обычно data_available=false).
zoning = _as_dict(result.get("zoning"))
if not zoning or not zoning.get("zone_code"):
note = zoning.get("note")
note_html = f'<p class="alt-meta">{_esc(note)}</p>' if note else ""
return _no_data() + note_html
zone_code = nspd_zoning.get("zone_code") or nspd_zoning.get("regulation_zone_index")
pairs: list[tuple[str, Any]] = [
("Зона ПЗЗ", zone_code),
("Наименование зоны", nspd_zoning.get("zone_name")),
("Макс. коэф. застройки (FAR)", nspd_zoning.get("max_far")),
("Макс. этажность", nspd_zoning.get("max_floors")),
("Макс. высота, м", nspd_zoning.get("max_height_m")),
("Макс. процент застройки", nspd_zoning.get("max_building_pct")),
("Мин. площадь участка, м²", nspd_zoning.get("min_parcel_area_m2")),
("Источник регламента", nspd_zoning.get("regulation_source")),
]
pairs = [(k, v) for k, v in pairs if v not in (None, "")]
zoning_table = _kv_table(pairs)
main_vri = _as_list(nspd_zoning.get("main_vri"))
vri_rows = [[v] for v in main_vri if v not in (None, "")]
vri_table = (
f"<h3>Разрешённые ВРИ зоны</h3>{_data_table(['Вид разрешённого использования'], vri_rows)}"
if vri_rows
else ""
)
return zoning_table + vri_table
def _build_zouit(result: dict[str, Any]) -> str:
"""ЗОУИТ-ограничения: сводка + список пересечений (тип / № границы / покрытие)."""
encumbrance = _as_dict(result.get("encumbrance"))
overlaps = _as_list(result.get("nspd_zouit_overlaps"))
summary_pairs: list[tuple[str, Any]] = [
("Есть ЗОУИТ", encumbrance.get("has_zouit")),
("Кол-во типов ЗОУИТ", encumbrance.get("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
coverage = ov.get("coverage_pct")
coverage_str = _fmt_pct(coverage) if isinstance(coverage, int | float) else _DASH
rows.append(
[
ov.get("type_zone") or ov.get("name"),
ov.get("reg_numb_border"),
coverage_str,
]
)
summary = _kv_table(summary_pairs)
table = _data_table(["Тип зоны", "№ границы", "Покрытие участка"], rows)
return summary + f"<h3>Пересечения ЗОУИТ</h3>{table}"
def _build_gate_verdict(result: dict[str, Any]) -> str:
"""Плашка gate-вердикта «можно ли строить МКД» + блокеры / предупреждения."""
gate = _as_dict(result.get("gate_verdict"))
if not gate:
return ""
can_build = gate.get("can_build_mkd")
label = gate.get("verdict_label") or ("Можно" if can_build else "Нельзя")
cls = "verdict-ok" if can_build else "verdict-no"
verdict = (
f'<div class="verdict-block {cls}">Строительство МКД: <strong>{_esc(label)}</strong></div>'
)
def _issues(items: Any, title: str) -> str:
rows = [
[it.get("code"), it.get("detail")] for it in _as_list(items) if isinstance(it, dict)
]
if not rows:
return ""
return f"<h3>{html.escape(title)}</h3>{_data_table(['Код', 'Детали'], rows)}"
blockers = _issues(gate.get("blockers"), "Блокеры")
warnings = _issues(gate.get("warnings"), "Предупреждения")
return verdict + blockers + warnings
def _build_section_1(result: dict[str, Any]) -> str:
"""§1 «Участок»: карта-плейсхолдер + факты + градрегламент + ЗОУИТ + gate-вердикт."""
return f"""
<div class="section" id="{_ANCHOR_S1}">
<h2>{html.escape(_TITLE_S1)}</h2>
<div class="map-placeholder">{MAP_PARCEL_PLACEHOLDER}</div>
<h3>Кадастровые факты</h3>
{_build_parcel_facts(result)}
<h3>Градостроительный регламент</h3>
{_build_zoning(result)}
<h3>Ограничения (ЗОУИТ)</h3>
{_build_zouit(result)}
{_build_gate_verdict(result)}
</div>
"""
# ──────────────────────────────────────────────────────────────────────────────
# §2 «Окружение»: транспорт / шум / воздух / геотехника / гидрология / соседи.
# ──────────────────────────────────────────────────────────────────────────────
def _build_transport(result: dict[str, Any]) -> str:
"""Транспорт: удалённость от центра ЕКБ + ближайшие станции метро."""
location = _as_dict(result.get("location"))
metro = _as_dict(result.get("metro"))
pairs: list[tuple[str, Any]] = []
dist = location.get("distance_to_center_km")
if dist is not None:
pairs.append(("До центра ЕКБ, км", dist))
metro_rows = [
[m.get("name"), _fmt_int_ru(m.get("distance_m"))]
for m in _as_list(metro.get("nearest_top3"))
if isinstance(m, dict)
]
metro_table = (
f"<h3>Ближайшее метро</h3>{_data_table(['Станция', 'Расстояние, м'], metro_rows)}"
if metro_rows
else ""
)
head = _kv_table(pairs) if pairs else _no_data()
return head + metro_table
def _build_noise_air(result: dict[str, Any]) -> str:
"""Шум + качество воздуха KV-таблицей."""
noise = _as_dict(result.get("noise"))
air = _as_dict(result.get("air_quality"))
pairs: list[tuple[str, Any]] = [
("Уровень шума", noise.get("level")),
("Оценка шума, дБ", noise.get("estimated_db")),
("NO₂", air.get("no2")),
("PM10", air.get("pm10")),
("PM2.5", air.get("pm2_5")),
("Источник воздуха", air.get("source")),
]
pairs = [(k, v) for k, v in pairs if v not in (None, "")]
return _kv_table(pairs)
def _build_geotech_hydro(result: dict[str, Any]) -> str:
"""Геотехника (сейсмика / промышленность) + гидрология (пойма / водные объекты)."""
geotech = _as_dict(result.get("geotech_risk"))
hydro = _as_dict(result.get("hydrology"))
pairs: list[tuple[str, Any]] = [
("Сейсмика", geotech.get("seismic_label")),
("Балльность", geotech.get("seismic_intensity_balls")),
("Многолетняя мерзлота", geotech.get("permafrost")),
("Промобъектов в 500 м", geotech.get("industrial_within_500m")),
("Риск подтопления", hydro.get("flood_risk_flag")),
]
pairs = [(k, v) for k, v in pairs if v not in (None, "")]
geotech_table = _kv_table(pairs)
water_rows = [
[w.get("name") or w.get("subtype"), _fmt_int_ru(w.get("distance_m"))]
for w in _as_list(hydro.get("nearest"))
if isinstance(w, dict)
]
water_table = (
f"<h3>Ближайшие водные объекты</h3>{_data_table(['Объект', 'Расстояние, м'], water_rows)}"
if water_rows
else ""
)
return geotech_table + water_table
def _build_neighbors(result: dict[str, Any]) -> str:
"""Соседняя застройка (cad_buildings в радиусе): назначение / этажность / расстояние."""
summary = _as_dict(result.get("neighbors_summary"))
neighbors = _as_list(summary.get("neighbors"))
rows: list[list[Any]] = []
for nb in neighbors:
if not isinstance(nb, dict):
continue
rows.append(
[
nb.get("building_name") or nb.get("cad_num"),
nb.get("purpose"),
nb.get("floors_parsed") or nb.get("floors"),
_fmt_int_ru(nb.get("distance_m")),
]
)
return _data_table(["Объект", "Назначение", "Этажей", "Расстояние, м"], rows)
def _build_section_2(result: dict[str, Any]) -> str:
"""§2 «Окружение»: транспорт + шум/воздух + геотехника/гидрология + соседи."""
return f"""
<div class="section" id="{_ANCHOR_S2}">
<h2>{html.escape(_TITLE_S2)}</h2>
<h3>Транспортная доступность</h3>
{_build_transport(result)}
<h3>Шум и качество воздуха</h3>
{_build_noise_air(result)}
<h3>Геотехника и гидрология</h3>
{_build_geotech_hydro(result)}
<h3>Соседняя застройка</h3>
{_build_neighbors(result)}
</div>
"""
# ──────────────────────────────────────────────────────────────────────────────
# §3 «Инженерные сети» + блок альтернатив программы.
# ──────────────────────────────────────────────────────────────────────────────
def _build_utilities(result: dict[str, Any]) -> str:
"""Инженерные сети рядом (OSM `utilities`): ближайшие сети + сводка по видам."""
utilities = _as_dict(result.get("utilities"))
if not utilities:
return _no_data()
pairs: list[tuple[str, Any]] = [
("Ближайшая ПС/подстанция, м", _fmt_int_ru(utilities.get("nearest_substation_m"))),
("Ближайший водовод, м", _fmt_int_ru(utilities.get("nearest_water_main_m"))),
("Ближайший газопровод, м", _fmt_int_ru(utilities.get("nearest_gas_m"))),
("Ближайшая теплотрасса, м", _fmt_int_ru(utilities.get("nearest_heat_m"))),
# ключ в persist-payload реально смешивает латиницу и кириллицу — НЕ «чинить»
("В охранной зоне ЛЭП", utilities.get("power_line_охранная_зона_flag")),
]
pairs = [(k, v) for k, v in pairs if v not in (None, _DASH)]
head = _kv_table(pairs)
summary_rows = [
[
s.get("subtype"),
_fmt_int_ru(s.get("nearest_m")),
s.get("count_within_2km"),
]
for s in _as_list(utilities.get("summary"))
if isinstance(s, dict)
]
summary_table = (
f"<h3>Сети рядом с участком (OSM)</h3>"
f"{_data_table(['Вид сети', 'Ближайшая, м', 'В радиусе 2 км'], summary_rows)}"
if summary_rows
else ""
)
return head + summary_table
def _build_engineering_nearby(result: dict[str, Any]) -> str:
"""Инженерные сооружения НСПД рядом (name / назначение / расстояние из raw_props)."""
items = _as_list(result.get("nspd_engineering_nearby"))
rows: list[list[Any]] = []
for it in items:
if not isinstance(it, dict):
continue
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")
rows.append([name, purpose, _fmt_int_ru(it.get("distance_m"))])
if not rows:
return ""
return (
f"<h3>Инженерные сооружения рядом (НСПД)</h3>"
f"{_data_table(['Сооружение', 'Назначение', 'Расстояние, м'], rows)}"
)
def _build_alternatives(result: dict[str, Any]) -> str:
"""Блок альтернатив программы (`program_alternatives`).
Присутствует ТОЛЬКО при отрицательном базовом фин-вердикте (program_optimizer.py);
иначе поле null → возвращаем пустую строку (блок не рисуется вовсе). Топ-N карточек
по NPV + честный caveat. Оценка по МАКС. застройке зоны, НЕ инвест-рекомендация.
"""
alt = _as_dict(result.get("program_alternatives"))
variants = _as_list(alt.get("variants"))
if not alt or not variants:
return ""
any_viable = bool(alt.get("any_viable"))
grid_size = alt.get("grid_size")
subtitle = (
"Базовая оценка отрицательна, но участок сходится при другой программе — "
"топ вариантов по NPV:"
if any_viable
else (
f"Ни один из {_fmt(grid_size)} вариантов программы (класс × этажность) не "
"выходит в плюс. Ниже — наименее убыточные:"
)
)
cards: list[str] = []
for v in variants:
if not isinstance(v, dict):
continue
npv = v.get("npv_rub")
npv_cls = "pos" if isinstance(npv, int | float) and npv > 0 else "neg"
title = (
f"{_housing_class_ru(v.get('housing_class'))} · "
f"{_fmt(v.get('floors'))} эт · {_development_type_ru(v.get('development_type'))}"
)
cards.append(
f"""
<div class="alt-card">
<div class="alt-title">{html.escape(title)}</div>
<div class="alt-npv {npv_cls}">NPV {html.escape(_fmt_money_signed(npv))}</div>
<div class="alt-meta">ROI {html.escape(_fmt_pct(v.get("roi")))} · """
f"""IRR {html.escape(_fmt_pct(v.get("irr")))}</div>
<div class="alt-meta">Жилая {html.escape(_fmt_int_ru(v.get("residential_sqm")))} м² · """
f"""{html.escape(_fmt_int_ru(v.get("apartments_count")))} квартир</div>
<div class="alt-meta">Цена {html.escape(_fmt_int_ru(v.get("price_per_sqm_used")))} ₽/м² """
f"""({html.escape(_price_source_ru(v.get("price_source")))})</div>
</div>"""
)
caveat = alt.get("caveat")
caveat_html = f'<div class="caveat">{_esc(caveat)}</div>' if caveat else ""
return f"""
<div class="section" id="{_ANCHOR_ALT}">
<h2>{html.escape(_TITLE_ALT)}</h2>
<p class="alt-meta">{html.escape(subtitle)}</p>
{"".join(cards)}
{caveat_html}
</div>
"""
def _build_connection_capacity(cap: dict[str, Any] | None) -> str:
"""Ресурсные резервы (ЦП/вода/газ/тепло) + сети рядом из connection-capacity (#2259 PR-D).
connection-capacity НЕ персистится в analyze-payload (ленивый эндпоинт), поэтому его
результат дотягивает оркестратор PR-D (`get_connection_capacity`) и передаёт сюда
отдельным аргументом. Ключи — из `ConnectionCapacityResponse` (schemas/parcel.py):
`power_summary`/`power_points`, `water`, `gas`, `heat`, `nearby_network_zones`.
None / пустой → блок не рисуется вовсе (пустая строка) — §3 деградирует к тому, что
есть в analyze-payload. PURE, graceful: каждый ключ через `.get()`, всё через `_esc`.
"""
cap = _as_dict(cap)
if not cap:
return ""
blocks: list[str] = []
# Электрика: сводка + ближайшие ЦП с резервом.
power_summary = _as_dict(cap.get("power_summary"))
nearest = _as_dict(power_summary.get("nearest_with_reserve"))
if power_summary or _as_list(cap.get("power_points")):
pairs: list[tuple[str, Any]] = [
("Центров питания в радиусе", power_summary.get("total_power_points")),
]
if nearest:
unit = nearest.get("reserve_unit") or "МВА"
pairs.append((f"Ближайший ЦП с резервом, {unit}", nearest.get("reserve_mva")))
pairs.append((" — класс напряжения", nearest.get("voltage_class")))
pairs.append((" — расстояние, м", _fmt_int_ru(nearest.get("distance_m"))))
pairs = [(k, v) for k, v in pairs if v not in (None, _DASH)]
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)
]
if water_rows:
blocks.append(
"<h3>Водоснабжение/водоотведение — резервы</h3>"
+ _data_table(["Система", "Резерв, тыс. м³/сут", "Период"], water_rows)
)
# Газ: городские ГРС со свободной мощностью.
gas = _as_dict(cap.get("gas"))
gas_rows = [
[
g.get("grs_name"),
_fmt(g.get("free_capacity_th_m3_h")),
_fmt_pct_raw(g.get("free_capacity_pct")),
]
for g in _as_list(gas.get("city_grs"))
if isinstance(g, dict)
]
if gas_rows:
blocks.append(
"<h3>Газоснабжение — свободная мощность ГРС</h3>"
+ _data_table(["ГРС", "Свободно, тыс. м³/ч", "Свободно, %"], gas_rows)
)
# Тепло: резервы систем теплоснабжения.
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:
blocks.append(
"<h3>Теплоснабжение — резервы систем</h3>"
+ _data_table(["Организация", "Система", "Резерв, Гкал/ч", "Период"], heat_rows)
)
# Позитив-разрез: сетевые охранные зоны рядом (где физически проходит сеть).
zone_rows = [
[z.get("human_label"), _fmt_int_ru(z.get("distance_m"))]
for z in _as_list(cap.get("nearby_network_zones"))
if isinstance(z, dict)
]
if zone_rows:
blocks.append(
"<h3>Сети рядом (охранные зоны)</h3>"
+ _data_table(["Сеть", "Расстояние, м"], zone_rows)
)
return "".join(blocks)
def _build_section_3(result: dict[str, Any], connection_capacity: dict[str, Any] | None) -> str:
"""§3 «Инженерные сети»: OSM-сети рядом + НСПД-сооружения + ресурсные резервы + альтернативы.
Ресурсные резервы (ЦП Россетей/ЕЭСК, вода, газ, тепло) и `nearby_network_zones` в
persist-payload ОТСУТСТВУЮТ (ленивый connection-capacity-эндпоинт) — их дотягивает
оркестратор PR-D и передаёт через `connection_capacity`. None → блок не рисуется.
"""
return f"""
<div class="section" id="{_ANCHOR_S3}">
<h2>{html.escape(_TITLE_S3)}</h2>
<h3>Инженерные сети рядом</h3>
{_build_utilities(result)}
{_build_engineering_nearby(result)}
{_build_connection_capacity(connection_capacity)}
</div>
{_build_alternatives(result)}
"""
# ──────────────────────────────────────────────────────────────────────────────
# §4§7 (Part B): forecast-ран (schema "1.0" §22-форсайта = SiteFinderReport.as_dict())
# + концепция (ConceptOutput.variants). PURE, graceful — как §1§3. Нормализация
# forecast-словаря — импортированными _fc_* хелперами report_pdf (НЕ дублируем).
# ──────────────────────────────────────────────────────────────────────────────
def _price_source_fin_ru(value: Any) -> str:
"""RU-фраза источника цены финмодели, иначе значение как есть. PURE."""
if isinstance(value, str) and value in _PRICE_SOURCE_FIN_RU:
return _PRICE_SOURCE_FIN_RU[value]
return _fmt(value)
def _strategy_ru(value: Any) -> str:
"""RU-метка стратегии генерации концепции, иначе значение как есть. PURE."""
if isinstance(value, str) and value in _STRATEGY_RU:
return _STRATEGY_RU[value]
return _fmt(value)
def _variant_strategy(variant: dict[str, Any]) -> Any:
"""Стратегия варианта: program-вариант помечен на первом footprint'е `strategy`.
Движок кладёт `properties.strategy == "program"` на пятна program-режима, но
сам вариант рапортует под слотом `balanced` — распознаём program по первой фиче,
чтобы не подписать пользовательскую программу «Балансом» (зеркало isProgramVariant
во фронте). Иначе — верхнеуровневый `strategy` варианта. PURE.
"""
features = _as_list(_as_dict(variant.get("buildings_geojson")).get("features"))
first = features[0] if features else None
if isinstance(first, dict) and _as_dict(first.get("properties")).get("strategy") == "program":
return "program"
return variant.get("strategy")
# ── §4 «Рынок» ──────────────────────────────────────────────────────────────────
def _build_market_metrics(forecast: dict[str, Any]) -> str:
"""Метрики рынка сейчас: темп/абсорбция, ликвидность/затоварка, цена, покрытие."""
market_now = _fc_as_dict(forecast.get("market_now"))
metrics = _fc_as_dict(market_now.get("market_metrics"))
pairs: list[tuple[str, Any]] = [
("Район", metrics.get("district")),
("Тип помещений", metrics.get("premise_kind")),
("Объектов-аналогов", metrics.get("obj_count")),
("Лотов в выборке", _fmt_int_ru(metrics.get("n_lots"))),
("Продано за период", _fmt_int_ru(metrics.get("n_sold"))),
("В продаже (доступно)", _fmt_int_ru(metrics.get("n_available"))),
("Темп продаж (velocity), ед./мес", metrics.get("unit_velocity")),
("Темп продаж (площадь), м²/мес", metrics.get("area_velocity")),
("Окно расчёта, мес", metrics.get("window_months")),
("Ставка абсорбции", metrics.get("absorption_rate")),
("Месяцев запаса (months of supply)", metrics.get("months_of_supply")),
("Индекс затоварки (overstock)", metrics.get("overstock_index")),
("Sell-through, %", metrics.get("sell_through_pct")),
]
pairs = [(k, v) for k, v in pairs if v not in (None, "", _DASH)]
return _kv_table(pairs)
def _build_market_competitors(forecast: dict[str, Any]) -> str:
"""Конкуренты рынка сейчас: ЖК / девелопер / класс / расстояние / лотов."""
market_now = _fc_as_dict(forecast.get("market_now"))
competitors = _as_list(market_now.get("competitors"))
rows: list[list[Any]] = []
for c in competitors:
if not isinstance(c, dict):
continue
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")),
]
)
return _data_table(["ЖК", "Девелопер", "Класс", "Расстояние, м", "Лотов"], rows)
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]] = []
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)
level_pairs: list[tuple[str, Any]] = [
("Итоговая уверенность отчёта", _fc_level_ru(confidence.get("level"))),
("Обоснование", confidence.get("rationale")),
]
level_pairs = [(k, v) for k, v in level_pairs if v not in (None, "", _DASH)]
return _kv_table(level_pairs) + coverage_table
def _build_section_4(forecast: dict[str, Any]) -> str:
"""§4 «Рынок»: резюме + метрики (velocity/тренд) + конкуренты + покрытие данными."""
market_now = _fc_as_dict(forecast.get("market_now"))
summary = market_now.get("summary")
summary_html = f'<p class="verdict">{_esc(summary)}</p>' if summary else ""
return f"""
<div class="section" id="{_ANCHOR_S4}">
<h2>{html.escape(_TITLE_S4)}</h2>
{summary_html}
<h3>Метрики рынка (темп, ликвидность, цена)</h3>
{_build_market_metrics(forecast)}
<h3>Конкуренты рядом</h3>
{_build_market_competitors(forecast)}
<h3>Покрытие данными и уверенность</h3>
{_build_market_coverage(forecast)}
</div>
"""
# ── §5 «Финансовая модель» ──────────────────────────────────────────────────────
def _best_variant(concept: dict[str, Any]) -> dict[str, Any]:
"""Опорный вариант концепции для §5-финмодели: с максимальным NPV, иначе первый.
§5 показывает ОДНУ финмодель — берём вариант с наибольшим NPV (лучший экономически),
при равенстве / отсутствии NPV — первый валидный. Пустой concept → {}. PURE.
"""
variants = [v for v in _as_list(concept.get("variants")) if isinstance(v, dict)]
if not variants:
return {}
def _npv(v: dict[str, Any]) -> float:
npv = _as_dict(v.get("financial")).get("npv_rub")
if isinstance(npv, bool) or not isinstance(npv, int | float):
return float("-inf")
return float(npv)
return max(variants, key=_npv)
def _build_financial_headline(financial: dict[str, Any]) -> str:
"""Ключевые числа финмодели: DCF (NPV / IRR / окупаемость) + ROI / прибыль."""
payback = financial.get("payback_months")
payback_str = f"{_fmt(payback)} мес" if payback is not None else "не окупается"
pairs: list[tuple[str, Any]] = [
("Выручка (GDV), ₽", _fmt_money_signed(financial.get("revenue_rub"))),
("Итого затраты, ₽", _fmt_money_signed(financial.get("cost_rub"))),
("Чистая прибыль, ₽", _fmt_money_signed(financial.get("net_profit_rub"))),
("ROI на затраты", _fmt_pct(financial.get("roi"))),
("Чистая маржа на выручку", _fmt_pct(financial.get("margin_pct"))),
("NPV (DCF), ₽", _fmt_money_signed(financial.get("npv_rub"))),
("IRR (DCF, годовой)", _fmt_pct(financial.get("irr"))),
("Ставка дисконтирования", _fmt_pct(financial.get("discount_rate_used"))),
("Окупаемость (PBP)", payback_str),
(
"Цена продажи жилья, ₽/м²",
_fmt_int_ru(financial.get("price_per_sqm_used")),
),
("Источник цены", _price_source_fin_ru(financial.get("price_source"))),
]
pairs = [(k, v) for k, v in pairs if v not in (None, "", _DASH)]
caveats: list[str] = []
if financial.get("irr_is_proxy"):
caveats.append(
"IRR помечен как оценочный: денежный поток вырожденный (нет смены знака), "
"показан аннуализированный ROI вместо DCF-IRR."
)
if financial.get("schedule_is_default"):
caveats.append(
"График фаз и темп продаж — типовые нормативные допущения (ПИР → СМР"
"распродажа); точность метрик зависит от графика конкретного проекта."
)
caveat_html = "".join(f'<div class="caveat">{html.escape(c)}</div>' for c in caveats)
return _kv_table(pairs) + caveat_html
def _build_financial_cascade(financial: dict[str, Any]) -> str:
"""Каскад затрат и БДР: выручка по статьям → затраты → НДС/налог → чистая прибыль."""
rows: list[list[Any]] = [
["Выручка — жильё", _fmt_money_signed(financial.get("revenue_residential_rub"))],
["Выручка — паркинг", _fmt_money_signed(financial.get("revenue_parking_rub"))],
["Выручка — нежилое (1-й этаж)", _fmt_money_signed(financial.get("revenue_office_rub"))],
["Выручка (GDV)", _fmt_money_signed(financial.get("revenue_rub"))],
["СМР", _fmt_money_signed(financial.get("construction_rub"))],
["ПИР (проектирование)", _fmt_money_signed(financial.get("pir_rub"))],
["Сети (ТУ)", _fmt_money_signed(financial.get("networks_rub"))],
["Услуги заказчика", _fmt_money_signed(financial.get("developer_services_rub"))],
["Резерв", _fmt_money_signed(financial.get("contingency_rub"))],
["Маркетинг и риэлтор", _fmt_money_signed(financial.get("marketing_rub"))],
["Земля", _fmt_money_signed(financial.get("land_rub"))],
["Итого затраты", _fmt_money_signed(financial.get("cost_rub"))],
["Валовая маржа", _fmt_money_signed(financial.get("gross_margin_rub"))],
["НДС (паркинг)", _fmt_money_signed(financial.get("vat_rub"))],
["Прибыль до налога", _fmt_money_signed(financial.get("profit_before_tax_rub"))],
["Налог на прибыль", _fmt_money_signed(financial.get("profit_tax_rub"))],
["Чистая прибыль", _fmt_money_signed(financial.get("net_profit_rub"))],
]
if financial.get("financing_enabled"):
rows.extend(
[
["Пиковый долг", _fmt_money_signed(financial.get("peak_debt_rub"))],
["Проценты по кредиту", _fmt_money_signed(financial.get("total_interest_rub"))],
[
"Чистая прибыль после финансирования",
_fmt_money_signed(financial.get("net_profit_after_financing_rub")),
],
["IRR на собственные средства (LTC)", _fmt_pct(financial.get("levered_irr"))],
]
)
return _data_table(["Статья", "Значение"], rows)
def _build_market_affordability(forecast: dict[str, Any]) -> str:
"""Рыночный контекст цены/платёжеспособности §5: цена/ticket из cost_of_error-индекса."""
indices = _fc_as_dict(
_fc_as_dict(_fc_as_dict(forecast.get("scoring")).get("special_indices")).get("indices")
)
cost_of_error = _fc_as_dict(indices.get("cost_of_error"))
detail = _fc_as_dict(cost_of_error.get("detail"))
if not detail:
return ""
pairs: list[tuple[str, Any]] = [
("Рыночная цена, ₽/м²", _fmt_int_ru(detail.get("price_per_m2"))),
("Средний чек лота, ₽", _fmt_money_signed(detail.get("avg_ticket_rub"))),
("Референс-площадь, м²", detail.get("ref_area_m2")),
("Индекс избытка предложения", detail.get("oversupply_risk")),
]
pairs = [(k, v) for k, v in pairs if v not in (None, "", _DASH)]
if not pairs:
return ""
return f"<h3>Рыночный контекст цены</h3>{_kv_table(pairs)}"
def _build_section_5(forecast: dict[str, Any], concept: dict[str, Any]) -> str:
"""§5 «Финмодель»: DCF/NPV/IRR/ROI + каскад затрат (из концепции) + рыночный контекст.
Финмодель (DCF) живёт в КОНЦЕПЦИИ (ConceptVariant.financial) — берём опорный вариант
(макс. NPV). Концепция не рассчитана → показываем только рыночный контекст цены из
forecast (cost_of_error) + честную заметку, без падения.
"""
variant = _best_variant(concept)
financial = _as_dict(variant.get("financial"))
affordability = _build_market_affordability(forecast)
if not financial:
note = (
'<div class="caveat">Финансовая модель (DCF, NPV/IRR/ROI, каскад затрат) '
"считается по концепции застройки — она ещё не рассчитана. Ниже — только "
"рыночный контекст цены из прогноза.</div>"
)
return f"""
<div class="section" id="{_ANCHOR_S5}">
<h2>{html.escape(_TITLE_S5)}</h2>
{note}
{affordability or _no_data()}
</div>
"""
strategy_label = _strategy_ru(_variant_strategy(variant))
return f"""
<div class="section" id="{_ANCHOR_S5}">
<h2>{html.escape(_TITLE_S5)}</h2>
<p class="alt-meta">Опорный вариант концепции (макс. NPV): {html.escape(strategy_label)}.</p>
<h3>Ключевые показатели (DCF)</h3>
{_build_financial_headline(financial)}
<h3>Каскад затрат и расчёт прибыли</h3>
{_build_financial_cascade(financial)}
{affordability}
</div>
"""
# ── §6 «Риски и дефицит» ────────────────────────────────────────────────────────
def _build_deficit_by_horizon(forecast: dict[str, Any]) -> str:
"""Прогноз дефицита/затоварки по горизонтам: спрос / предложение / индекс / months."""
future = _fc_as_dict(forecast.get("future_market"))
forecasts = _as_list(future.get("forecasts_by_horizon"))
rows: list[list[Any]] = []
for f in forecasts:
if not isinstance(f, dict):
continue
rows.append(
[
f.get("horizon_months"),
_fmt_int_ru(f.get("projected_demand_units")),
_fmt_int_ru(f.get("projected_supply_units")),
f.get("deficit_index"),
f.get("months_of_inventory"),
_fc_level_ru(f.get("confidence")),
]
)
headers = ["Горизонт, мес", "Спрос", "Предложение", "Индекс дефицита", "Мес. запаса", "Увер."]
return _data_table(headers, rows)
def _build_supply_pressure(forecast: dict[str, Any]) -> str:
"""Давление будущего предложения (future_supply): открытый/скрытый сток, поглощение."""
future = _fc_as_dict(forecast.get("future_market"))
pairs = _fc_future_supply_pairs(future.get("future_supply"))
kv = [(k, v) for k, v in pairs.items() if v not in (None, "", _DASH)]
return _kv_table(kv)
def _build_risk_indices(forecast: dict[str, Any]) -> str:
"""Риск-индексы дефицита/ошибки: белые пятна, цена ошибки, окно запуска, каннибализация."""
special = _fc_as_dict(_fc_as_dict(forecast.get("scoring")).get("special_indices"))
indices = _fc_as_dict(special.get("indices"))
rows: list[list[Any]] = []
for _key, payload in indices.items():
data = _fc_as_dict(payload)
if not data:
continue
rows.append([data.get("key"), data.get("label"), data.get("value")])
return _data_table(["Индекс", "Метка", "Значение"], rows)
def _build_scenarios_honesty(forecast: dict[str, Any]) -> str:
"""Честный блок сценариев: разброс дефицита ИЛИ плашка-объяснение при коллапсе."""
scenarios = _fc_as_dict(forecast.get("scenarios"))
if scenarios.get("scenarios_collapsed"):
reason = scenarios.get("scenarios_collapse_reason")
reason_text = (
reason
if isinstance(reason, str) and reason
else "Сценарная дифференциация недоступна на текущих данных."
)
return (
'<div class="caveat">Сценарная дифференциация недоступна: '
f"{html.escape(reason_text)}</div>"
)
summary = _fc_as_dict(scenarios.get("scenarios_summary"))
if not summary:
summary = _fc_as_dict(_fc_as_dict(forecast.get("future_market")).get("scenarios_summary"))
pairs = [(str(name), value) for name, value in summary.items()]
return _kv_table(pairs)
def _build_section_6(forecast: dict[str, Any]) -> str:
"""§6 «Риски и дефицит»: дефицит по горизонтам + давление предложения + риск-индексы."""
future = _fc_as_dict(forecast.get("future_market"))
summary = future.get("summary")
summary_html = f'<p class="verdict">{_esc(summary)}</p>' if summary else ""
return f"""
<div class="section" id="{_ANCHOR_S6}">
<h2>{html.escape(_TITLE_S6)}</h2>
{summary_html}
<h3>Дефицит / затоварка по горизонтам</h3>
{_build_deficit_by_horizon(forecast)}
<h3>Давление будущего предложения</h3>
{_build_supply_pressure(forecast)}
<h3>Риск-индексы</h3>
{_build_risk_indices(forecast)}
<h3>Сценарии</h3>
{_build_scenarios_honesty(forecast)}
</div>
"""
# ── §7 «Концепция застройки» ────────────────────────────────────────────────────
def _build_concept_program(variant: dict[str, Any]) -> str:
"""Программа застройки варианта из размещённых фич (тип дома × этажность → секций).
ВЫХОДНОЙ `ConceptVariant` (schemas/concept.py) НЕ несёт `building_program` — это поле
только на входном `ConceptInput`. Реальный состав программы восстанавливаем из
размещённых footprint'ов: в program-режиме `buildings_geojson.features[].properties`
несут `section_type` + `floors` (см. `_program_footprints_to_geojson` в placement.py).
1b-стратегии (max_area / max_insolation / balanced) кладут только `floors` без
`section_type` → таблица программы не рисуется (у них нет каталожного состава).
Группируем по (тип, этажность) → count; порядок групп — по первому появлению
(стабилен). PURE.
"""
features = _as_list(_as_dict(variant.get("buildings_geojson")).get("features"))
groups: dict[tuple[Any, Any], int] = {}
order: list[tuple[Any, Any]] = []
for feat in features:
props = _as_dict(_as_dict(feat).get("properties"))
section_type = props.get("section_type")
if section_type in (None, ""):
continue
key = (section_type, props.get("floors"))
if key not in groups:
order.append(key)
groups[key] = groups.get(key, 0) + 1
if not order:
return ""
rows = [[stype, floors, groups[(stype, floors)]] for stype, floors in order]
return (
"<h3>Программа застройки</h3>" f"{_data_table(['Тип дома', 'Этажность', 'Секций'], rows)}"
)
def _build_concept_variant(variant: dict[str, Any]) -> str:
"""Карточка одного варианта концепции: ТЭП + финитог + программа."""
teap = _as_dict(variant.get("teap"))
financial = _as_dict(variant.get("financial"))
strategy_label = _strategy_ru(_variant_strategy(variant))
features = _as_list(_as_dict(variant.get("buildings_geojson")).get("features"))
corpus_count = len(features)
teap_pairs: list[tuple[str, Any]] = [
("Корпусов", corpus_count if corpus_count else None),
("Площадь застройки, м²", _fmt_int_ru(teap.get("built_area_sqm"))),
("Общая площадь (GFA), м²", _fmt_int_ru(teap.get("total_floor_area_sqm"))),
("Жилая площадь, м²", _fmt_int_ru(teap.get("residential_area_sqm"))),
("Нежилое (1-й этаж), м²", _fmt_int_ru(teap.get("office_area_sqm"))),
("Квартир", _fmt_int_ru(teap.get("apartments_count"))),
("Плотность (FAR)", teap.get("density")),
("Машино-мест", _fmt_int_ru(teap.get("parking_spaces"))),
]
teap_pairs = [(k, v) for k, v in teap_pairs if v not in (None, "", _DASH)]
fin_pairs: list[tuple[str, Any]] = [
("Выручка (GDV), ₽", _fmt_money_signed(financial.get("revenue_rub"))),
("Итого затраты, ₽", _fmt_money_signed(financial.get("cost_rub"))),
("Чистая прибыль, ₽", _fmt_money_signed(financial.get("net_profit_rub"))),
("NPV (DCF), ₽", _fmt_money_signed(financial.get("npv_rub"))),
("ROI", _fmt_pct(financial.get("roi"))),
("IRR", _fmt_pct(financial.get("irr"))),
]
fin_pairs = [(k, v) for k, v in fin_pairs if v not in (None, "", _DASH)]
partial = ""
placed = variant.get("placed_count")
requested = variant.get("requested_count")
if (
isinstance(placed, int)
and isinstance(requested, int)
and not isinstance(placed, bool)
and not isinstance(requested, bool)
and placed < requested
):
partial = (
f'<div class="caveat">Размещено {placed} из {requested} секций — участок '
"вмещает меньше, чем в заданной программе; ТЭП и финмодель — по фактически "
"размещённым корпусам.</div>"
)
return f"""
<div class="alt-card">
<div class="alt-title">{html.escape(strategy_label)}</div>
{partial}
<h3>ТЭП</h3>
{_kv_table(teap_pairs)}
<h3>Финансовый итог</h3>
{_kv_table(fin_pairs)}
{_build_concept_program(variant)}
</div>
"""
def _build_section_7(concept: dict[str, Any] | None) -> str:
"""§7 «Концепция»: footprint-план (плейсхолдер PR-C) + ТЭП/финитог/программа вариантов.
`concept` = сериализованный `ConceptOutput` (POST /concepts) или None. None / пустые
варианты → честная заметка «концепция не рассчитана» (без падения).
"""
concept_dict = _as_dict(concept)
variants = [v for v in _as_list(concept_dict.get("variants")) if isinstance(v, dict)]
if not variants:
return f"""
<div class="section" id="{_ANCHOR_S7}">
<h2>{html.escape(_TITLE_S7)}</h2>
<div class="caveat">Концепция застройки не рассчитана — раздел появится после генерации
вариантов концепции для участка.</div>
</div>
"""
cards = "".join(_build_concept_variant(v) for v in variants)
return f"""
<div class="section" id="{_ANCHOR_S7}">
<h2>{html.escape(_TITLE_S7)}</h2>
<div class="map-placeholder">{MAP_CONCEPT_PLACEHOLDER}</div>
{cards}
</div>
"""
def build_full_report_html_part_b(
forecast_result: dict[str, Any],
concept_result: dict[str, Any] | None,
*,
cad: str,
) -> str:
"""Собрать HTML Part B полного отчёта: §4 «Рынок» + §5 «Финмодель» + §6 «Риски» + §7.
PURE (без WeasyPrint / БД / сети). §4§6 читают ПЕРСИСТНУТЫЙ форсайт-ран
(`analysis_runs.result`, schema §22-форсайта "1.0" = `SiteFinderReport.as_dict()`) —
нормализуется теми же `_fc_*`-хелперами, что и PDF-рендерер форсайта. §7 читает
сериализованный `ConceptOutput` (POST /concepts). Каждый ключ — через `.get()` с
дефолтом; отсутствующая секция → «нет данных»; ВСЕ строки payload проходят
`html.escape`. Карта footprint-плана §7 — плейсхолдер `{{MAP_CONCEPT}}` (PR-C).
Args:
forecast_result: `analysis_runs.result` (schema §22-форсайта "1.0") или любой
dict/инстанс с `as_dict()`. Не-dict / None → пустой форсайт (секции → «нет
данных»).
concept_result: сериализованный `ConceptOutput` или None. None / без вариантов →
§7 рисует честную заметку «концепция не рассчитана» (§5 — только рыночный
контекст цены).
cad: кадастровый номер участка (для логов; в HTML приходит через каркас).
Returns:
HTML-фрагмент Part B (четыре `<div class="section">`), готовый как `part_b_html`
для `build_full_report_html`.
"""
forecast = _fc_normalize(forecast_result)
part_b = (
_build_section_4(forecast)
+ _build_section_5(forecast, _as_dict(concept_result))
+ _build_section_6(forecast)
+ _build_section_7(concept_result)
)
logger.info(
"build_full_report_html_part_b: cad=%s forecast_keys=%d concept_variants=%d",
cad,
len(forecast),
len(_as_list(_as_dict(concept_result).get("variants"))),
)
return part_b
# ──────────────────────────────────────────────────────────────────────────────
# Публичный API PR-A: сборка Part A (§1§3) + общий каркас документа.
# ──────────────────────────────────────────────────────────────────────────────
def build_full_report_html_part_a(
analyze_result: dict[str, Any],
*,
cad: str,
connection_capacity: dict[str, Any] | None = None,
) -> str:
"""Собрать HTML Part A полного отчёта: §1 «Участок» + §2 «Окружение» + §3 «Сети».
PURE (без WeasyPrint / БД / сети). Читает ПЕРСИСТНУТЫЙ `analysis_runs.result`
(`analyze-1.0`, loose dict). Каждый ключ — через `.get()` с дефолтом; отсутствующая
секция → «нет данных»; ВСЕ строки payload проходят `html.escape`. Карта участка —
плейсхолдер `{{MAP_PARCEL}}` (PR-C). Блок альтернатив (`program_alternatives`)
рисуется только при его наличии в payload.
Args:
analyze_result: `analysis_runs.result` (schema `analyze-1.0`) или любой dict.
Не-dict / None → трактуется как пустой payload (все секции → «нет данных»).
cad: кадастровый номер участка (для логов; в HTML не встраивается напрямую —
он приходит в каркас через `build_full_report_html`).
connection_capacity: результат `get_connection_capacity` (ЦП/вода/газ/тепло/сети
рядом) — НЕ персистится в analyze-payload, дотягивается оркестратором PR-D
и рендерится блоком «Ресурсные резервы» в §3. None → блок пропускается.
Returns:
HTML-фрагмент Part A (три `<div class="section">` + опциональный блок альтернатив),
готовый как `part_a_html` для `build_full_report_html`.
"""
result = _as_dict(analyze_result)
part_a = (
_build_section_1(result)
+ _build_section_2(result)
+ _build_section_3(result, connection_capacity)
)
logger.info(
"build_full_report_html_part_a: cad=%s keys=%d has_alternatives=%s has_capacity=%s",
cad,
len(result),
bool(_as_dict(result.get("program_alternatives")).get("variants")),
bool(_as_dict(connection_capacity)),
)
return part_a
def _build_toc(has_part_b: bool) -> str:
"""Оглавление-якоря. Part B опционален (PR-B) → его пункты скрываем, если None. PURE."""
items = [
f'<li><a href="#{_ANCHOR_S1}">{html.escape(_TITLE_S1)}</a></li>',
f'<li><a href="#{_ANCHOR_S2}">{html.escape(_TITLE_S2)}</a></li>',
f'<li><a href="#{_ANCHOR_S3}">{html.escape(_TITLE_S3)}</a></li>',
]
if has_part_b:
items.extend(
[
f'<li><a href="#{_ANCHOR_S4}">{html.escape(_TITLE_S4)}</a></li>',
f'<li><a href="#{_ANCHOR_S5}">{html.escape(_TITLE_S5)}</a></li>',
f'<li><a href="#{_ANCHOR_S6}">{html.escape(_TITLE_S6)}</a></li>',
f'<li><a href="#{_ANCHOR_S7}">{html.escape(_TITLE_S7)}</a></li>',
]
)
return f'<nav class="toc"><h3>Содержание</h3><ul>{"".join(items)}</ul></nav>'
def build_full_report_html(
part_a_html: str,
part_b_html: str | None,
*,
cad: str,
address: str | None,
generated_at: str,
) -> str:
"""Собрать единый печатный HTML-документ полного отчёта (каркас + Part A [+ Part B]).
Общий `<style>` (печатная типографика A4, page-break-правила WeasyPrint), титул с
кад.номером / адресом / датой, оглавление-якоря. `part_b_html=None` → блок §4§7
(PR-B) пропускается, пункт оглавления скрывается. PURE (только строки); WeasyPrint
здесь НЕ вызывается — render-обёртка (PR-C) отдельно.
Args:
part_a_html: HTML §1§3 от `build_full_report_html_part_a`.
part_b_html: HTML §4§7 (PR-B) или None (пропуск).
cad: кадастровый номер участка (титул).
address: человекочитаемый адрес участка или None (титул).
generated_at: строка даты формирования (уже отформатированная вызывающим).
Returns:
Полный HTML-документ (готов для WeasyPrint в PR-C).
"""
has_part_b = bool(part_b_html)
addr_line = f'<div class="cover-meta">Адрес: {html.escape(address)}</div>' if address else ""
part_b_block = part_b_html or ""
return f"""<!DOCTYPE html>
<html lang="ru">
<head>
<meta charset="UTF-8">
<title>{html.escape(_TITLE_DOC)}{html.escape(cad)}</title>
<style>{_CSS}</style>
</head>
<body>
<header class="cover">
<h1>{html.escape(_TITLE_DOC)}</h1>
<div class="cover-meta">Кадастровый номер: {html.escape(cad)}</div>
{addr_line}
<div class="cover-meta">Дата формирования: {html.escape(generated_at)}</div>
</header>
{_build_toc(has_part_b)}
{part_a_html}
{part_b_block}
<div class="footer">
GenDesign · Site Finder ПТИЦА · отчёт носит советующий характер и не является
основанием для инвестиционного решения.
</div>
</body>
</html>"""