bot-backend
47b3fffc99
All checks were successful
CI Trade-In / changes (pull_request) Successful in 9s
CI / changes (pull_request) Successful in 9s
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) Has been skipped
CI / openapi-codegen-check (pull_request) Successful in 2m5s
CI / backend-tests (pull_request) Successful in 15m17s
market_pulse.competitors_total and neighbors_summary.count_buildings_100m were silently capped by their SQL LIMITs (20/30) and reported as true totals; add unbounded COUNT(*) for both (coverage_pct now uses the true denominator), matching the features_truncated pattern from epic #2445 A2. VelocityResult gains competitors_truncated for its LIMIT-200 query. The permits-nearby table in both the HTML and DOCX full-report exporters now discloses "… и ещё N записей" instead of silently hard-slicing to 10. neighbors_truncated compares against the actually-returned (20) list, catching a second Python-side [:20] slice beyond the SQL LIMIT 30. Refs #2464 (cluster B silent-caps).
1277 lines
65 KiB
Python
1277 lines
65 KiB
Python
"""DOCX-вариант полного отчёта ПТИЦА — зеркало PDF-конвейера (эпик #2259 PR-F).
|
||
|
||
Второй формат выгрузки полного отчёта `/site-finder/analysis/{cad}` (рядом с PDF-
|
||
конвейером `full_report_pdf`): рендерит те же §1–§7 + альтернативы в редактируемый
|
||
Word-документ (.docx) и возвращает БАЙТЫ, готовые для FileResponse / записи файла.
|
||
|
||
DRY / SINGLE SOURCE: НЕ парсим HTML и НЕ дублируем нормализацию — переиспользуем
|
||
ИСХОДНЫЕ словари (analyze / forecast / concept / connection-capacity) и PURE-хелперы
|
||
форматирования из `full_report_html` (`_fmt`/`_fmt_int_ru`/`_fmt_money_signed`/…). Все
|
||
эти хелперы — приватные (`_`-префикс), но пакет внутренний, импорт по имени легитимен.
|
||
Отличие от HTML-ветки: здесь значения кладём В ЯЧЕЙКИ python-docx как есть (`_fmt`),
|
||
БЕЗ `html.escape` (docx — не HTML, XML-экранирование делает сам lxml внутри python-docx).
|
||
|
||
СТРУКТУРА (зеркало PDF, `build_full_report_pdf` → `build_full_report_html`):
|
||
титул (cad / адрес / дата) → §1 «Участок» (карта + факты + регламент + ЗОУИТ + gate)
|
||
→ §2 «Окружение» → §3 «Сети» (+ connection-capacity) → альтернативы (если есть)
|
||
→ §4 «Рынок» → §5 «Финмодель» → §6 «Риски» → §7 «Концепция» (карта + варианты).
|
||
|
||
КАРТЫ: PNG-байты участка (§1) и footprint-плана концепции (§7) прокидываются
|
||
аргументами (те же рендеры `report_maps`, что и у PDF — НЕ рендерим дважды). None →
|
||
плашка-абзац «карта недоступна» (graceful, как `embed_map_png` в PDF-ветке).
|
||
|
||
ДЕГРАДАЦИИ (те же, что у PDF/HTML): нет концепции → §7 честная плашка + §5 только
|
||
рыночный контекст; нет forecast → §4–§6 «нет данных»; нет connection-capacity → §3
|
||
без резервов; пустая секция → «нет данных». НИКОГДА не падает на частичном payload.
|
||
|
||
`python-docx` (PyPI `python-docx`, import `docx`) — чистый Python (lxml), без нативных
|
||
системных библиотек: Dockerfile не трогаем, тесты без skip. Импортируется ЛОКАЛЬНО
|
||
внутри `build_full_report_docx` (зеркало WeasyPrint / `report_docx.render_report_docx`).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import io
|
||
import logging
|
||
from typing import TYPE_CHECKING, Any
|
||
|
||
# Единый источник форматтеров/нормализации — HTML-билдеры полного отчёта. Импортируем
|
||
# приватные PURE-хелперы по имени (внутренний пакет): числа/деньги/проценты/RU-метки
|
||
# причёсываются ТАК ЖЕ, как в PDF-ветке → цифры в DOCX и PDF совпадают байт-в-байт.
|
||
# `_fc_*`-хелперы (нормализация forecast-словаря) — реэкспорт из report_pdf через
|
||
# full_report_html, тянем оттуда же (одна точка импорта).
|
||
from app.services.exporters.full_report_html import (
|
||
_as_dict,
|
||
_as_list,
|
||
_development_type_ru,
|
||
_fc_as_dict,
|
||
_fc_future_supply_pairs,
|
||
_fc_level_ru,
|
||
_fc_normalize,
|
||
_fmt,
|
||
_fmt_int_ru,
|
||
_fmt_money,
|
||
_fmt_money_signed,
|
||
_fmt_pct,
|
||
_fmt_pct_raw,
|
||
_housing_class_ru,
|
||
_price_source_fin_ru,
|
||
_price_source_ru,
|
||
_reserve_num,
|
||
_strategy_ru,
|
||
_truncate_name,
|
||
_variant_strategy,
|
||
)
|
||
from app.services.exporters.full_report_html import (
|
||
_best_variant as _best_concept_variant,
|
||
)
|
||
from app.services.exporters.full_report_html import (
|
||
_capacity_rows_capped as _cap_rows,
|
||
)
|
||
from app.services.exporters.full_report_html import (
|
||
_competitor_lots as _competitor_lots,
|
||
)
|
||
from app.services.exporters.full_report_html import (
|
||
_is_non_ekb as _is_non_ekb,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
# Только для аннотаций — тяжёлый python-docx импортируем ЛОКАЛЬНО в
|
||
# build_full_report_docx (как WeasyPrint в report_pdf / Document в report_docx).
|
||
from docx.document import Document as _DocxDocument
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Микрокопия / заглушки (зеркало full_report_html) ────────────────────────────
|
||
_DASH = "—"
|
||
_NO_DATA = "нет данных"
|
||
|
||
# Кап видимых строк тепло/вода-таблиц (зеркало _HEAT_ROW_CAP / _WATER_ROW_CAP в HTML).
|
||
_HEAT_ROW_CAP = 15
|
||
_WATER_ROW_CAP = 25
|
||
# Кап видимых строк таблицы разрешений §6 (зеркало _PERMITS_ROW_CAP в HTML).
|
||
_PERMITS_ROW_CAP = 10
|
||
|
||
# Титул документа + заголовки секций (зеркало _TITLE_* в full_report_html).
|
||
_TITLE_DOC = "Отчёт по участку — Site Finder ПТИЦА"
|
||
_TITLE_S1 = "§1. Участок"
|
||
_TITLE_S2 = "§2. Окружение"
|
||
_TITLE_S3 = "§3. Инженерные сети"
|
||
_TITLE_ALT = "Как участок сходится (альтернативы программы)"
|
||
_TITLE_S4 = "§4. Рынок"
|
||
_TITLE_S5 = "§5. Финансовая модель"
|
||
_TITLE_S6 = "§6. Риски и дефицит"
|
||
_TITLE_S7 = "§7. Концепция застройки"
|
||
|
||
# Встроенный стиль таблиц python-docx (рисует видимые границы; есть в любом дефолтном
|
||
# Document, не требует шаблона). Зеркало _TABLE_STYLE в report_docx.
|
||
_TABLE_STYLE = "Table Grid"
|
||
|
||
# Ширина карт-картинок (дюймы) — под ширину печатной A4-колонки при полях документа.
|
||
# python-docx масштабирует пропорционально при задании только ширины.
|
||
_MAP_WIDTH_INCHES = 6.3
|
||
|
||
# Плашка недоступной карты (зеркало embed_map_png-плейсхолдера PDF-ветки).
|
||
_MAP_UNAVAILABLE = "Карта недоступна."
|
||
|
||
# Дисклеймер-футер (зеркало .footer в build_full_report_html).
|
||
_FOOTER = (
|
||
"GenDesign · Site Finder ПТИЦА · отчёт носит советующий характер и не является "
|
||
"основанием для инвестиционного решения."
|
||
)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# docx-специфичные микро-билдеры. Все мутируют переданный `doc`/контейнер. Значения
|
||
# причёсываются импортированными PURE-хелперами и кладутся как есть (без html.escape —
|
||
# XML-экранирование делает python-docx/lxml). Пустая секция → «нет данных» (graceful).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _add_heading(doc: _DocxDocument, text: str, level: int) -> None:
|
||
"""Заголовок секции (python-docx Heading level). Тонкая обёртка для единообразия."""
|
||
doc.add_heading(text, level=level)
|
||
|
||
|
||
def _add_no_data(doc: _DocxDocument) -> None:
|
||
"""Абзац-заглушка «нет данных» (graceful для пустой секции)."""
|
||
doc.add_paragraph(_NO_DATA)
|
||
|
||
|
||
def _add_kv_table(doc: _DocxDocument, pairs: list[tuple[str, Any]]) -> None:
|
||
"""Таблица «метка → значение» из списка пар. Пустой список → «нет данных».
|
||
|
||
`label` — статичная RU-метка (bold-run), значение — через `_fmt` (None → «—»).
|
||
Зеркало `_kv_table` в full_report_html (там HTML-таблица). Значения ПРЕД-
|
||
форматированные (`_fmt_int_ru`/`_fmt_money_signed`/…) билдер оставляет строкой.
|
||
"""
|
||
if not pairs:
|
||
_add_no_data(doc)
|
||
return
|
||
table = doc.add_table(rows=0, cols=2)
|
||
table.style = _TABLE_STYLE
|
||
for label, value in pairs:
|
||
cells = table.add_row().cells
|
||
# Метка — bold-run (первый абзац ячейки); значение — plain через _fmt.
|
||
cells[0].text = ""
|
||
cells[0].paragraphs[0].add_run(str(label)).bold = True
|
||
cells[1].text = _fmt(value)
|
||
|
||
|
||
def _add_data_table(doc: _DocxDocument, headers: list[str], rows: list[list[Any]]) -> None:
|
||
"""Таблица: шапка (bold RU-метки) + строки данных (через `_fmt`). Graceful.
|
||
|
||
Пустой `rows` → одна строка-заглушка «нет данных» под шапкой (таблица всё равно
|
||
валидна). Зеркало `_data_table` в full_report_html (там HTML `<table class="data">`).
|
||
"""
|
||
table = doc.add_table(rows=1, cols=len(headers))
|
||
table.style = _TABLE_STYLE
|
||
header_cells = table.rows[0].cells
|
||
for idx, header in enumerate(headers):
|
||
header_cells[idx].text = ""
|
||
header_cells[idx].paragraphs[0].add_run(str(header)).bold = True
|
||
|
||
if not rows:
|
||
empty_cells = table.add_row().cells
|
||
empty_cells[0].text = _NO_DATA
|
||
return
|
||
|
||
for row in rows:
|
||
cells = table.add_row().cells
|
||
for idx in range(len(headers)):
|
||
# Строка короче шапки (defensive) → недостающее «—».
|
||
value = row[idx] if idx < len(row) else None
|
||
cells[idx].text = _fmt(value)
|
||
|
||
|
||
def _add_caveat(doc: _DocxDocument, text: str) -> None:
|
||
"""Честная заметка/оговорка курсивом (зеркало `.caveat` HTML-ветки)."""
|
||
para = doc.add_paragraph()
|
||
para.add_run(str(text)).italic = True
|
||
|
||
|
||
def _add_map(doc: _DocxDocument, png: bytes | None) -> None:
|
||
"""Вставить карту-PNG (`add_picture` из BytesIO) или плашку «карта недоступна».
|
||
|
||
None / пустые байты / сбой декодирования → абзац-плашка (graceful, как embed_map_png
|
||
в PDF-ветке ставит SVG-плашку). Ширина фиксирована (`_MAP_WIDTH_INCHES`) — высота
|
||
масштабируется пропорционально самим python-docx.
|
||
"""
|
||
if not png:
|
||
_add_caveat(doc, _MAP_UNAVAILABLE)
|
||
return
|
||
try:
|
||
from docx.shared import Inches
|
||
|
||
doc.add_picture(io.BytesIO(png), width=Inches(_MAP_WIDTH_INCHES))
|
||
except Exception:
|
||
# Битый PNG / несовместимый формат — не роняем отчёт, ставим плашку.
|
||
logger.exception("full_report_docx: вставка карты упала → плашка «недоступна»")
|
||
_add_caveat(doc, _MAP_UNAVAILABLE)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# §1 «Участок» — зеркало _build_section_1 (HTML). Источник: analyze-payload.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_parcel_facts(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Кадастровые факты: адрес / площадь / категория / ВРИ / статус / стоимость."""
|
||
egrn = _as_dict(result.get("egrn"))
|
||
parcel_meta = _as_dict(result.get("parcel_meta"))
|
||
geometry = _as_dict(result.get("geometry_suitability"))
|
||
|
||
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]] = [
|
||
("Адрес", egrn.get("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, "")]
|
||
_add_kv_table(doc, pairs)
|
||
|
||
|
||
def _build_zoning(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Градрегламент ПЗЗ: зона + ТЭП (FAR / этажность / высота) + список ВРИ зоны."""
|
||
nspd_zoning = _as_dict(result.get("nspd_zoning"))
|
||
if not nspd_zoning:
|
||
zoning = _as_dict(result.get("zoning"))
|
||
if not zoning or not zoning.get("zone_code"):
|
||
_add_no_data(doc)
|
||
note = zoning.get("note")
|
||
if note:
|
||
doc.add_paragraph(_fmt(note))
|
||
return
|
||
nspd_zoning = zoning
|
||
|
||
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, "")]
|
||
_add_kv_table(doc, pairs)
|
||
|
||
vri_rows = [[v] for v in _as_list(nspd_zoning.get("main_vri")) if v not in (None, "")]
|
||
if vri_rows:
|
||
_add_heading(doc, "Разрешённые ВРИ зоны", level=3)
|
||
_add_data_table(doc, ["Вид разрешённого использования"], vri_rows)
|
||
|
||
|
||
def _build_zouit(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""ЗОУИТ-ограничения: сводка + список пересечений (тип / № границы / покрытие)."""
|
||
encumbrance = _as_dict(result.get("encumbrance"))
|
||
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"))
|
||
|
||
# То же разрешение противоречия «нет ЗОУИТ / но НСПД-пересечения есть», что в HTML.
|
||
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]] = [
|
||
("Есть ЗОУИТ", has_zouit),
|
||
("Кол-во типов ЗОУИТ", zouit_count),
|
||
]
|
||
if zouit_types:
|
||
summary_pairs.append(("Типы", ", ".join(str(t) for t in zouit_types)))
|
||
_add_kv_table(doc, summary_pairs)
|
||
|
||
rows: list[list[Any]] = []
|
||
for ov in overlaps:
|
||
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]
|
||
)
|
||
_add_heading(doc, "Пересечения ЗОУИТ", level=3)
|
||
_add_data_table(doc, ["Тип зоны", "№ границы", "Покрытие участка"], rows)
|
||
|
||
|
||
def _build_gate_verdict(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""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 "Нельзя")
|
||
para = doc.add_paragraph()
|
||
para.add_run("Строительство МКД: ").bold = True
|
||
para.add_run(_fmt(label)).bold = True
|
||
|
||
def _issues(items: Any, title: str) -> None:
|
||
rows = [
|
||
[it.get("code"), it.get("detail")] for it in _as_list(items) if isinstance(it, dict)
|
||
]
|
||
if not rows:
|
||
return
|
||
_add_heading(doc, title, level=3)
|
||
_add_data_table(doc, ["Код", "Детали"], rows)
|
||
|
||
_issues(gate.get("blockers"), "Блокеры")
|
||
_issues(gate.get("warnings"), "Предупреждения")
|
||
|
||
|
||
def _build_section_1(
|
||
doc: _DocxDocument, result: dict[str, Any], parcel_map_png: bytes | None
|
||
) -> None:
|
||
"""§1 «Участок»: карта + кадастровые факты + регламент + ЗОУИТ + gate-вердикт."""
|
||
_add_heading(doc, _TITLE_S1, level=1)
|
||
_add_map(doc, parcel_map_png)
|
||
_add_heading(doc, "Кадастровые факты", level=2)
|
||
_build_parcel_facts(doc, result)
|
||
_add_heading(doc, "Градостроительный регламент", level=2)
|
||
_build_zoning(doc, result)
|
||
_add_heading(doc, "Ограничения (ЗОУИТ)", level=2)
|
||
_build_zouit(doc, result)
|
||
_build_gate_verdict(doc, result)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# §2 «Окружение» — зеркало _build_section_2 (HTML).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_transport(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Транспорт: удалённость от центра ЕКБ + ближайшие станции метро."""
|
||
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))
|
||
_add_kv_table(doc, pairs)
|
||
|
||
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)
|
||
]
|
||
if metro_rows:
|
||
_add_heading(doc, "Ближайшее метро", level=3)
|
||
_add_data_table(doc, ["Станция", "Расстояние, м"], metro_rows)
|
||
|
||
|
||
def _build_noise_air(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Шум + качество воздуха 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, "")]
|
||
_add_kv_table(doc, pairs)
|
||
|
||
|
||
def _build_geotech_hydro(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Геотехника (сейсмика / промышленность) + гидрология (пойма / водные объекты)."""
|
||
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, "")]
|
||
_add_kv_table(doc, 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)
|
||
]
|
||
if water_rows:
|
||
_add_heading(doc, "Ближайшие водные объекты", level=3)
|
||
_add_data_table(doc, ["Объект", "Расстояние, м"], water_rows)
|
||
|
||
|
||
def _build_neighbors(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Соседняя застройка (cad_buildings в радиусе): назначение / этажность / расстояние."""
|
||
summary = _as_dict(result.get("neighbors_summary"))
|
||
rows: list[list[Any]] = []
|
||
for nb in _as_list(summary.get("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")),
|
||
]
|
||
)
|
||
_add_data_table(doc, ["Объект", "Назначение", "Этажей", "Расстояние, м"], rows)
|
||
|
||
|
||
def _build_section_2(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""§2 «Окружение»: транспорт + шум/воздух + геотехника/гидрология + соседи."""
|
||
_add_heading(doc, _TITLE_S2, level=1)
|
||
_add_heading(doc, "Транспортная доступность", level=2)
|
||
_build_transport(doc, result)
|
||
_add_heading(doc, "Шум и качество воздуха", level=2)
|
||
_build_noise_air(doc, result)
|
||
_add_heading(doc, "Геотехника и гидрология", level=2)
|
||
_build_geotech_hydro(doc, result)
|
||
_add_heading(doc, "Соседняя застройка", level=2)
|
||
_build_neighbors(doc, result)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# §3 «Инженерные сети» + connection-capacity + альтернативы.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_utilities(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Инженерные сети рядом (OSM `utilities`): ближайшие сети + сводка по видам."""
|
||
utilities = _as_dict(result.get("utilities"))
|
||
if not utilities:
|
||
_add_no_data(doc)
|
||
return
|
||
|
||
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)]
|
||
_add_kv_table(doc, 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)
|
||
]
|
||
if summary_rows:
|
||
_add_heading(doc, "Сети рядом с участком (OSM)", level=3)
|
||
_add_data_table(doc, ["Вид сети", "Ближайшая, м", "В радиусе 2 км"], summary_rows)
|
||
|
||
|
||
def _build_engineering_nearby(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Инженерные сооружения НСПД рядом (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")
|
||
if name in (None, "") and purpose in (None, ""):
|
||
continue
|
||
rows.append([name, purpose, _fmt_int_ru(it.get("distance_m"))])
|
||
if not any_item:
|
||
return
|
||
_add_heading(doc, "Инженерные сооружения рядом (НСПД)", level=3)
|
||
_add_data_table(doc, ["Сооружение", "Назначение", "Расстояние, м"], rows)
|
||
|
||
|
||
def _build_connection_capacity(doc: _DocxDocument, cap: dict[str, Any] | None) -> None:
|
||
"""Ресурсные резервы (ЦП/вода/газ/тепло) + сети рядом. Зеркало HTML-версии.
|
||
|
||
Переиспользует общий `_capacity_rows_capped` (фильтр областного шума + кап строк) из
|
||
HTML-модуля — та же логика честности (дефицит всегда виден, не-ЕКБ схлопывается).
|
||
None / пустой → блок не рисуется (§3 деградирует к OSM-payload).
|
||
"""
|
||
cap = _as_dict(cap)
|
||
if not cap:
|
||
return
|
||
|
||
# Электрика: сводка + ближайший ЦП с резервом.
|
||
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:
|
||
_add_heading(doc, "Электроснабжение — свободная мощность", level=3)
|
||
_add_kv_table(doc, pairs)
|
||
|
||
# Вода: резервы ЦСВ/ЦСК (ЕКБ-системы + кап ~25).
|
||
water_rows, water_hidden = _cap_rows(
|
||
_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:
|
||
_add_heading(doc, "Водоснабжение/водоотведение — резервы", level=3)
|
||
_add_data_table(doc, ["Система", "Резерв, тыс. м³/сут", "Период"], 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:
|
||
_add_heading(doc, "Газоснабжение — свободная мощность ГРС", level=3)
|
||
_add_data_table(doc, ["ГРС", "Свободно, тыс. м³/ч", "Свободно, %"], gas_rows)
|
||
|
||
# Тепло: агрегат ЕКБ + топ-15 + кап (зеркало HTML).
|
||
heat = _as_dict(cap.get("heat"))
|
||
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 = _cap_rows(
|
||
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,
|
||
]
|
||
)
|
||
_add_heading(doc, "Теплоснабжение — резервы систем", level=3)
|
||
_add_data_table(doc, ["Организация", "Система", "Резерв, Гкал/ч", "Период"], 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:
|
||
_add_heading(doc, "Сети рядом (охранные зоны)", level=3)
|
||
_add_data_table(doc, ["Сеть", "Расстояние, м"], zone_rows)
|
||
|
||
|
||
def _build_section_3(
|
||
doc: _DocxDocument, result: dict[str, Any], connection_capacity: dict[str, Any] | None
|
||
) -> None:
|
||
"""§3 «Инженерные сети»: OSM-сети + НСПД-сооружения + ресурсные резервы."""
|
||
_add_heading(doc, _TITLE_S3, level=1)
|
||
_add_heading(doc, "Инженерные сети рядом", level=2)
|
||
_build_utilities(doc, result)
|
||
_build_engineering_nearby(doc, result)
|
||
_build_connection_capacity(doc, connection_capacity)
|
||
|
||
|
||
def _build_alternatives(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""Блок альтернатив программы (`program_alternatives`) — топ-N по NPV + caveat.
|
||
|
||
Присутствует ТОЛЬКО при отрицательном базовом фин-вердикте; иначе поле null → блок не
|
||
рисуется вовсе (зеркало HTML). PDF рисует карточки — здесь эквивалентная таблица.
|
||
"""
|
||
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)} вариантов программы (класс × этажность) не "
|
||
"выходит в плюс. Ниже — наименее убыточные:"
|
||
)
|
||
)
|
||
_add_heading(doc, _TITLE_ALT, level=1)
|
||
doc.add_paragraph(subtitle)
|
||
|
||
rows: list[list[Any]] = []
|
||
for v in variants:
|
||
if not isinstance(v, dict):
|
||
continue
|
||
program = (
|
||
f"{_housing_class_ru(v.get('housing_class'))} · "
|
||
f"{_fmt(v.get('floors'))} эт · {_development_type_ru(v.get('development_type'))}"
|
||
)
|
||
price = (
|
||
f"{_fmt_int_ru(v.get('price_per_sqm_used'))} ₽/м² "
|
||
f"({_price_source_ru(v.get('price_source'))})"
|
||
)
|
||
rows.append(
|
||
[
|
||
program,
|
||
_fmt_money_signed(v.get("npv_rub")),
|
||
_fmt_pct(v.get("roi")),
|
||
_fmt_pct(v.get("irr")),
|
||
_fmt_int_ru(v.get("residential_sqm")),
|
||
_fmt_int_ru(v.get("apartments_count")),
|
||
price,
|
||
]
|
||
)
|
||
_add_data_table(
|
||
doc,
|
||
["Программа", "NPV", "ROI", "IRR", "Жилая, м²", "Квартир", "Цена"],
|
||
rows,
|
||
)
|
||
|
||
caveat = alt.get("caveat")
|
||
if caveat:
|
||
_add_caveat(doc, caveat)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# §4–§7 (Part B): forecast-ран (нормализован _fc_normalize) + концепция.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_market_metrics(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Метрики рынка сейчас: темп/абсорбция, ликвидность/затоварка, цена, покрытие."""
|
||
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")),
|
||
("Ставка абсорбции (в мес.)", _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")),
|
||
]
|
||
pairs = [(k, v) for k, v in pairs if v not in (None, "", _DASH)]
|
||
_add_kv_table(doc, pairs)
|
||
|
||
|
||
def _build_market_competitors(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Конкуренты рынка сейчас: группировка по (ЖК, девелопер), лоты суммой, ближайшая дист.
|
||
|
||
Тот же схлопывающий алгоритм, что в HTML `_build_market_competitors` (по строке НА
|
||
КОРПУС в прод-payload — группируем в один ЖК с «(K корпусов)»).
|
||
"""
|
||
market_now = _fc_as_dict(forecast.get("market_now"))
|
||
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 key in order:
|
||
g = groups[key]
|
||
name = g["name"]
|
||
if g["corpus"] > 1:
|
||
name = f"{_fmt(name)} ({g['corpus']} корпусов)"
|
||
rows.append(
|
||
[
|
||
name,
|
||
g["dev_name"],
|
||
g["obj_class"],
|
||
_fmt_int_ru(g["distance_m"]),
|
||
_fmt_int_ru(g["lots"]),
|
||
]
|
||
)
|
||
_add_data_table(doc, ["ЖК", "Девелопер", "Класс", "Расстояние, м", "Лотов"], rows)
|
||
|
||
|
||
def _build_market_coverage(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Покрытие данными: итоговая уверенность + факторы (сделки / история / ДОМ.РФ)."""
|
||
confidence = _fc_as_dict(forecast.get("confidence"))
|
||
factors = _fc_as_dict(confidence.get("factors"))
|
||
|
||
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)]
|
||
_add_kv_table(doc, level_pairs)
|
||
|
||
parsed: list[tuple[Any, Any, Any]] = []
|
||
for _key, payload in factors.items():
|
||
data = _fc_as_dict(payload)
|
||
if not data:
|
||
continue
|
||
label = data.get("label") or data.get("note")
|
||
parsed.append((label, _fc_level_ru(data.get("level")), data.get("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]
|
||
_add_data_table(doc, ["Фактор", "Уровень", "Комментарий"], rows)
|
||
else:
|
||
rows = [[label, level] for label, level, _note in parsed]
|
||
_add_data_table(doc, ["Фактор", "Уровень"], rows)
|
||
|
||
|
||
def _build_section_4(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""§4 «Рынок»: резюме + метрики + конкуренты + покрытие данными."""
|
||
_add_heading(doc, _TITLE_S4, level=1)
|
||
market_now = _fc_as_dict(forecast.get("market_now"))
|
||
summary = market_now.get("summary")
|
||
if summary:
|
||
doc.add_paragraph(_fmt(summary))
|
||
_add_heading(doc, "Метрики рынка (темп, ликвидность, цена)", level=2)
|
||
_build_market_metrics(doc, forecast)
|
||
_add_heading(doc, "Конкуренты рядом", level=2)
|
||
_build_market_competitors(doc, forecast)
|
||
_add_heading(doc, "Покрытие данными и уверенность", level=2)
|
||
_build_market_coverage(doc, forecast)
|
||
|
||
|
||
def _build_financial_headline(doc: _DocxDocument, financial: dict[str, Any]) -> None:
|
||
"""Ключевые числа финмодели: DCF (NPV / IRR / окупаемость) + ROI / прибыль + caveats."""
|
||
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)]
|
||
_add_kv_table(doc, pairs)
|
||
|
||
if financial.get("irr_is_proxy"):
|
||
_add_caveat(
|
||
doc,
|
||
"IRR помечен как оценочный: денежный поток вырожденный (нет смены знака), "
|
||
"показан аннуализированный ROI вместо DCF-IRR.",
|
||
)
|
||
if financial.get("schedule_is_default"):
|
||
_add_caveat(
|
||
doc,
|
||
"График фаз и темп продаж — типовые нормативные допущения (ПИР → СМР → "
|
||
"распродажа); точность метрик зависит от графика конкретного проекта.",
|
||
)
|
||
|
||
|
||
def _build_financial_cascade(doc: _DocxDocument, financial: dict[str, Any]) -> None:
|
||
"""Каскад затрат и БДР: выручка по статьям → затраты → НДС/налог → чистая прибыль."""
|
||
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"))],
|
||
]
|
||
)
|
||
_add_data_table(doc, ["Статья", "Значение"], rows)
|
||
|
||
|
||
def _build_market_affordability(doc: _DocxDocument, forecast: dict[str, Any]) -> bool:
|
||
"""Рыночный контекст цены §5 (cost_of_error-индекс). Возвращает: нарисовано ли что-то."""
|
||
indices = _fc_as_dict(
|
||
_fc_as_dict(_fc_as_dict(forecast.get("scoring")).get("special_indices")).get("indices")
|
||
)
|
||
detail = _fc_as_dict(_fc_as_dict(indices.get("cost_of_error")).get("detail"))
|
||
if not detail:
|
||
return False
|
||
pairs: list[tuple[str, Any]] = [
|
||
("Рыночная цена, ₽/м²", _fmt_int_ru(detail.get("price_per_m2"))),
|
||
("Средний чек лота, ₽", _fmt_money(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 False
|
||
_add_heading(doc, "Рыночный контекст цены", level=2)
|
||
_add_kv_table(doc, pairs)
|
||
return True
|
||
|
||
|
||
def _build_section_5(doc: _DocxDocument, forecast: dict[str, Any], concept: dict[str, Any]) -> None:
|
||
"""§5 «Финмодель»: DCF/NPV/IRR/ROI + каскад затрат (из концепции) + рыночный контекст.
|
||
|
||
Финмодель живёт в опорном варианте концепции (макс. NPV). Концепция не рассчитана →
|
||
только рыночный контекст цены + честная заметка (зеркало HTML). Без падения.
|
||
"""
|
||
_add_heading(doc, _TITLE_S5, level=1)
|
||
variant = _best_concept_variant(concept)
|
||
financial = _as_dict(variant.get("financial"))
|
||
|
||
if not financial:
|
||
_add_caveat(
|
||
doc,
|
||
"Финансовая модель (DCF, NPV/IRR/ROI, каскад затрат) считается по концепции "
|
||
"застройки — она ещё не рассчитана. Ниже — только рыночный контекст цены из "
|
||
"прогноза.",
|
||
)
|
||
if not _build_market_affordability(doc, forecast):
|
||
_add_no_data(doc)
|
||
return
|
||
|
||
strategy_label = _strategy_ru(_variant_strategy(variant))
|
||
doc.add_paragraph(f"Опорный вариант концепции (макс. NPV): {strategy_label}.")
|
||
_add_heading(doc, "Ключевые показатели (DCF)", level=2)
|
||
_build_financial_headline(doc, financial)
|
||
_add_heading(doc, "Каскад затрат и расчёт прибыли", level=2)
|
||
_build_financial_cascade(doc, financial)
|
||
_build_market_affordability(doc, forecast)
|
||
|
||
|
||
def _build_deficit_by_horizon(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Прогноз дефицита/затоварки по горизонтам: спрос / предложение / индекс / months."""
|
||
future = _fc_as_dict(forecast.get("future_market"))
|
||
rows: list[list[Any]] = []
|
||
for f in _as_list(future.get("forecasts_by_horizon")):
|
||
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 = ["Горизонт, мес", "Спрос", "Предложение", "Индекс дефицита", "Мес. запаса", "Увер."]
|
||
_add_data_table(doc, headers, rows)
|
||
|
||
|
||
def _build_supply_pressure(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Давление будущего предложения (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)]
|
||
_add_kv_table(doc, kv)
|
||
|
||
|
||
def _build_risk_indices(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Риск-индексы дефицита/ошибки: белые пятна, цена ошибки, окно запуска, каннибализация."""
|
||
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")])
|
||
_add_data_table(doc, ["Индекс", "Метка", "Значение"], rows)
|
||
|
||
|
||
def _build_scenarios_honesty(doc: _DocxDocument, forecast: dict[str, Any]) -> None:
|
||
"""Честный блок сценариев: разброс дефицита ИЛИ плашка-объяснение при коллапсе."""
|
||
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 "Сценарная дифференциация недоступна на текущих данных."
|
||
)
|
||
_add_caveat(doc, f"Сценарная дифференциация недоступна: {reason_text}")
|
||
return
|
||
|
||
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()]
|
||
_add_kv_table(doc, pairs)
|
||
|
||
|
||
def _build_permits_nearby(doc: _DocxDocument, result: dict[str, Any]) -> None:
|
||
"""РНС/РВЭ в радиусе 500 м участка (ГИСОГД-66) — короткая сводка + список до 10.
|
||
|
||
`result` — analyze-payload (НЕ forecast): читает `permits_nearby` (см.
|
||
`permits_nearby.get_permits_nearby`). Пусто / total_count=0 → честная плашка-фраза
|
||
(полное предложение, не аббревиатура). Зеркало `_build_permits_nearby` в
|
||
full_report_html.
|
||
|
||
#2464 cluster B: `total_count` — честный total апстрима (get_permits_nearby
|
||
считает его COUNT'ом БЕЗ SQL LIMIT), `items` уже капнут апстримом на 30. Здесь
|
||
список дополнительно режется до `_PERMITS_ROW_CAP` (10) для компактности — раньше
|
||
это резалось МОЛЧА. Дисклоузим разницу `total_count - показано` строкой
|
||
«и ещё N …», как тепло/вода-таблицы выше (`_build_connection_capacity`).
|
||
"""
|
||
nearby = _as_dict(result.get("permits_nearby"))
|
||
total = nearby.get("total_count")
|
||
if not isinstance(total, int) or total <= 0:
|
||
_add_caveat(
|
||
doc,
|
||
"В радиусе 500 м участка новых разрешений на строительство не найдено "
|
||
"(по данным ГИСОГД Свердловской области).",
|
||
)
|
||
return
|
||
|
||
rs_count = nearby.get("rs_count") or 0
|
||
rv_count = nearby.get("rv_count") or 0
|
||
nearest = nearby.get("nearest_distance_m")
|
||
kv: list[tuple[str, Any]] = [
|
||
("Разрешений на строительство (РНС)", _fmt_int_ru(rs_count)),
|
||
("Разрешений на ввод (РВЭ)", _fmt_int_ru(rv_count)),
|
||
("Ближайшее, м", _fmt_int_ru(nearest) if nearest is not None else _DASH),
|
||
]
|
||
_add_kv_table(doc, kv)
|
||
|
||
rows: list[list[Any]] = []
|
||
for item in _as_list(nearby.get("items"))[:_PERMITS_ROW_CAP]:
|
||
data = _as_dict(item)
|
||
rows.append(
|
||
[
|
||
data.get("doc_name"),
|
||
data.get("date_doc"),
|
||
data.get("approved_organization"),
|
||
data.get("distance_m"),
|
||
]
|
||
)
|
||
hidden = total - len(rows)
|
||
if hidden > 0:
|
||
rows.append(
|
||
[f"… и ещё {hidden} записей (полный список в веб-версии §6)", _DASH, _DASH, _DASH]
|
||
)
|
||
headers = ["Документ", "Дата", "Согласующий орган", "Дистанция, м"]
|
||
_add_data_table(doc, headers, rows)
|
||
|
||
|
||
def _build_section_6(doc: _DocxDocument, forecast: dict[str, Any], result: dict[str, Any]) -> None:
|
||
"""§6 «Риски и дефицит»: дефицит по горизонтам + давление предложения + риск-индексы.
|
||
|
||
`result` — analyze-payload (для блока разрешений рядом, `permits_nearby`); `forecast` —
|
||
форсайт-ран (дефицит/сценарии). Разные источники — §6 читает оба (зеркало HTML-версии).
|
||
"""
|
||
_add_heading(doc, _TITLE_S6, level=1)
|
||
future = _fc_as_dict(forecast.get("future_market"))
|
||
summary = future.get("summary")
|
||
if summary:
|
||
doc.add_paragraph(_fmt(summary))
|
||
_add_heading(doc, "Дефицит / затоварка по горизонтам", level=2)
|
||
_build_deficit_by_horizon(doc, forecast)
|
||
_add_heading(doc, "Давление будущего предложения", level=2)
|
||
_build_supply_pressure(doc, forecast)
|
||
_add_heading(doc, "Риск-индексы", level=2)
|
||
_build_risk_indices(doc, forecast)
|
||
_add_heading(doc, "Сценарии", level=2)
|
||
_build_scenarios_honesty(doc, forecast)
|
||
_add_heading(doc, "Разрешения на строительство рядом (500 м)", level=2)
|
||
_build_permits_nearby(doc, result)
|
||
|
||
|
||
def _build_concept_program(doc: _DocxDocument, variant: dict[str, Any]) -> None:
|
||
"""Программа застройки варианта из размещённых фич (тип дома × этажность → секций)."""
|
||
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]
|
||
_add_heading(doc, "Программа застройки", level=3)
|
||
_add_data_table(doc, ["Тип дома", "Этажность", "Секций"], rows)
|
||
|
||
|
||
def _build_concept_variant(doc: _DocxDocument, variant: dict[str, Any]) -> None:
|
||
"""Карточка одного варианта концепции: ТЭП + финитог + программа."""
|
||
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)
|
||
|
||
_add_heading(doc, strategy_label, level=2)
|
||
|
||
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
|
||
):
|
||
_add_caveat(
|
||
doc,
|
||
f"Размещено {placed} из {requested} секций — участок вмещает меньше, чем в "
|
||
"заданной программе; ТЭП и финмодель — по фактически размещённым корпусам.",
|
||
)
|
||
|
||
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)]
|
||
_add_heading(doc, "ТЭП", level=3)
|
||
_add_kv_table(doc, teap_pairs)
|
||
|
||
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)]
|
||
_add_heading(doc, "Финансовый итог", level=3)
|
||
_add_kv_table(doc, fin_pairs)
|
||
|
||
_build_concept_program(doc, variant)
|
||
|
||
|
||
def _build_section_7(
|
||
doc: _DocxDocument, concept: dict[str, Any] | None, concept_map_png: bytes | None
|
||
) -> None:
|
||
"""§7 «Концепция»: footprint-план (карта) + ТЭП/финитог/программа вариантов.
|
||
|
||
`concept` = сериализованный `ConceptOutput` или None. None / пустые варианты →
|
||
честная плашка «концепция не рассчитана» (зеркало HTML, без падения).
|
||
"""
|
||
_add_heading(doc, _TITLE_S7, level=1)
|
||
concept_dict = _as_dict(concept)
|
||
variants = [v for v in _as_list(concept_dict.get("variants")) if isinstance(v, dict)]
|
||
|
||
if not variants:
|
||
_add_caveat(
|
||
doc,
|
||
"Концепция застройки не рассчитана — раздел появится после генерации вариантов "
|
||
"концепции для участка.",
|
||
)
|
||
return
|
||
|
||
_add_map(doc, concept_map_png)
|
||
for v in variants:
|
||
_build_concept_variant(doc, v)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Публичный API PR-F.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def build_full_report_docx(
|
||
analyze_result: dict[str, Any],
|
||
forecast_result: dict[str, Any] | None,
|
||
concept_result: dict[str, Any] | None,
|
||
connection_capacity: dict[str, Any] | None,
|
||
*,
|
||
cad: str,
|
||
address: str | None,
|
||
generated_at: str,
|
||
parcel_map_png: bytes | None,
|
||
concept_map_png: bytes | None,
|
||
) -> bytes:
|
||
"""Собрать DOCX-вариант полного отчёта ПТИЦА (зеркало PDF) → байты. #2259 PR-F.
|
||
|
||
Структура зеркалит PDF-конвейер (`build_full_report_pdf` → `build_full_report_html`):
|
||
титул → §1–§3 (+ connection-capacity в §3, + альтернативы) → §4–§7. Источники — те же
|
||
ИСХОДНЫЕ словари, что кормят HTML-билдеры (НЕ парсим HTML). Карты — PNG-байты тех же
|
||
рендеров `report_maps` (прокинуты аргументами, НЕ рендерим дважды).
|
||
|
||
GRACEFUL: forecast=None → §4–§6 «нет данных»; concept=None → §7 плашка + §5 только
|
||
рыночный контекст; connection_capacity=None → §3 без резервов; карта None → плашка
|
||
«недоступна». НИКОГДА не падает на частичном payload.
|
||
|
||
Args:
|
||
analyze_result: `analysis_runs.result` (schema `analyze-1.0`) — §1–§3.
|
||
forecast_result: forecast-ран (schema §22-форсайта "1.0") или None — §4–§6.
|
||
concept_result: сериализованный `ConceptOutput` или None — §5/§7.
|
||
connection_capacity: результат `get_connection_capacity` или None — §3-резервы.
|
||
cad: кадастровый номер участка (титул).
|
||
address: человекочитаемый адрес или None (титул).
|
||
generated_at: строка даты формирования (уже отформатирована вызывающим).
|
||
parcel_map_png: PNG-байты карты участка (§1) или None (плашка).
|
||
concept_map_png: PNG-байты footprint-плана концепции (§7) или None (плашка).
|
||
|
||
Returns:
|
||
Непустые DOCX-байты (OOXML zip, начинаются с `b"PK"`), готовые для FileResponse /
|
||
записи файла.
|
||
"""
|
||
# python-docx импортируем локально — тяжёлый (lxml); не нужен при импорте модуля
|
||
# (зеркало WeasyPrint в full_report_pdf / Document в report_docx).
|
||
try:
|
||
from docx import Document
|
||
except ImportError as exc:
|
||
raise RuntimeError(
|
||
"python-docx не установлен. Добавь 'python-docx>=1.1.0' в pyproject.toml."
|
||
) from exc
|
||
|
||
result = _as_dict(analyze_result)
|
||
forecast = _fc_normalize(forecast_result)
|
||
concept = _as_dict(concept_result)
|
||
|
||
doc = Document()
|
||
|
||
# ── Титул (зеркало cover в build_full_report_html) ──────────────────────────
|
||
doc.add_heading(_TITLE_DOC, level=0)
|
||
doc.add_paragraph(f"Кадастровый номер: {cad}")
|
||
if address:
|
||
doc.add_paragraph(f"Адрес: {address}")
|
||
doc.add_paragraph(f"Дата формирования: {generated_at}")
|
||
|
||
# ── §1–§3 (Part A) ──────────────────────────────────────────────────────────
|
||
_build_section_1(doc, result, parcel_map_png)
|
||
_build_section_2(doc, result)
|
||
_build_section_3(doc, result, connection_capacity)
|
||
_build_alternatives(doc, result)
|
||
|
||
# ── §4–§7 (Part B) ──────────────────────────────────────────────────────────
|
||
_build_section_4(doc, forecast)
|
||
_build_section_5(doc, forecast, concept)
|
||
_build_section_6(doc, forecast, result)
|
||
_build_section_7(doc, concept_result, concept_map_png)
|
||
|
||
# ── Футер-дисклеймер ────────────────────────────────────────────────────────
|
||
_add_caveat(doc, _FOOTER)
|
||
|
||
buffer = io.BytesIO()
|
||
doc.save(buffer)
|
||
docx_bytes = buffer.getvalue()
|
||
logger.info(
|
||
"build_full_report_docx: cad=%s size=%d bytes forecast_keys=%d concept_variants=%d "
|
||
"parcel_map=%s concept_map=%s",
|
||
cad,
|
||
len(docx_bytes),
|
||
len(forecast),
|
||
len(_as_list(concept.get("variants"))),
|
||
bool(parcel_map_png),
|
||
bool(concept_map_png),
|
||
)
|
||
return docx_bytes
|