gendesign/backend/app/services/exporters/report_docx.py
Light1YT 764db61734
All checks were successful
CI / changes (pull_request) Successful in 7s
CI / changes (push) Successful in 7s
CI / frontend-tests (push) Successful in 1m1s
CI / frontend-tests (pull_request) Successful in 1m17s
CI / openapi-codegen-check (push) Successful in 2m19s
CI / backend-tests (pull_request) Successful in 9m11s
CI / backend-tests (push) Successful in 9m15s
CI / openapi-codegen-check (pull_request) Successful in 1m43s
fix(exporters): annotate scenario deficit horizon in docx/pptx (#1590 follow-up)
report_docx и report_pptx рендерили bare _scenario_deficit_index под
захардкоженной шапкой «(12 мес)» — при fallback на другой горизонт подпись
лгала. Переключил оба на _scenario_deficit_cell (как в report_md): per-cell
аннотация «(гор. N мес)» при N != 12. Закрывает code-review major по #1590.
2026-06-17 14:35:48 +05:00

442 lines
23 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.

"""§22 DOCX-экспортёр итогового советующего отчёта Site Finder v2 (python-docx).
#959 (EPIC export) — завершает набор форматов выгрузки §22-форсайта (md/json/tg уже
есть): рендерит `SiteFinderReport` (#987) в редактируемый Word-документ (.docx) и
возвращает БАЙТЫ — готовые для
`Response(media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document")`.
Аддитивный двойник PDF/Markdown-экспортёров (`report_pdf` / `report_md`): то же
содержание и ТОТ ЖЕ порядок секций (зеркало `report_md`).
DRY: PURE-хелперы нормализации/форматирования НЕ дублируем — импортируем из
`report_pdf` (там они уже есть: `_normalize` — приём инстанса ИЛИ `as_dict()`-словаря,
`_fmt` — строковое форматирование ячейки/None→"", `_level_ru`, `_as_dict`/`_as_list`,
`_future_supply_pairs`, `_scenario_deficit_cell`) + переиспользуем его named-константы
(`_DASH`/`_NO_DATA`/`_ADVISORY_MARKER`). Здесь добавлены ТОЛЬКО docx-специфичные
микро-билдеры (заголовки/абзацы/таблицы python-docx).
ДЕТЕРМИНИРОВАННЫЙ, БЕЗ LLM, БЕЗ БД/сети: только ПОТРЕБЛЯЕТ уже-собранный отчёт (его
наполняет сборщик #988) и раскладывает по секциям. Принимает КАК `SiteFinderReport`-
инстанс, ТАК и его `as_dict()`-словарь (нормализуется через `_normalize`).
GRACEFUL (дух всего форсайт-стека): частичный/пустой/мусорный отчёт ВАЛИДЕН — пустая
секция рисует «нет данных», экспортёр НИКОГДА не падает (нет KeyError на тонком отчёте)
и возвращает минимальный валидный .docx. Отчёт СОВЕТУЮЩИЙ: после титула — заметный
advisory-дисклеймер (оценка не основание для инвест-решения).
`python-docx` (PyPI `python-docx`, import `docx`) — чистый Python (lxml), без нативных
системных библиотек: Dockerfile не трогаем. Импортируется ЛОКАЛЬНО внутри функции
(зеркало WeasyPrint в `report_pdf` — не нужен при импорте модуля).
"""
from __future__ import annotations
import io
import logging
from typing import TYPE_CHECKING, Any
from app.services.exporters.report_pdf import (
_ADVISORY_MARKER,
_DASH,
_NO_DATA,
_as_dict,
_as_list,
_fmt,
_future_supply_pairs,
_level_ru,
_normalize,
_scenario_deficit_cell,
)
if TYPE_CHECKING:
# Только для аннотаций: тяжёлый python-docx импортируем ЛОКАЛЬНО в render_report_docx
# (как WeasyPrint в report_pdf), чтобы импорт модуля был дешёвым и не требовал dep.
from docx.document import Document as _DocxDocument
logger = logging.getLogger(__name__)
# ── Named-константы: заголовки (зеркало report_md — те же восемь секций §22) ──
_TITLE_DOC: str = "Site Finder v2 — советующий отчёт"
_TITLE_SUMMARY: str = "Сводка"
_TITLE_MARKET_NOW: str = "Рынок сейчас"
_TITLE_FUTURE_MARKET: str = "Будущий рынок"
_TITLE_SCENARIOS: str = "Сценарии"
_TITLE_PRODUCT_TZ: str = "Продукт ТЗ"
_TITLE_SCORING: str = "Скоринг"
_TITLE_CONFIDENCE: str = "Уверенность"
# Краткий RU-дисклеймер (формулировка из ТЗ #959) — после титула, виден сразу.
_ADVISORY_DISCLAIMER: str = "Оценка advisory — не основание для инвест-решения."
# Стиль таблиц python-docx (встроенный «Table Grid» — есть в любом дефолтном Document,
# рисует видимые границы; не требует кастомного шаблона). Зеркало house-style соседей.
_TABLE_STYLE: str = "Table Grid"
# Сколько конкурентов выводить в таблицу (top-N — зеркало report_md._COMPETITORS_TOP_N).
_COMPETITORS_TOP_N: int = 5
# Основной продуктовый горизонт (мес) — подпись столбца сводного дефицита сценария
# (зеркало report_md._PRIMARY_HORIZON_MONTHS; значение тянет _scenario_deficit_cell —
# при fallback на чужой горизонт ячейка несёт «(гор. N мес)», #1590).
_PRIMARY_HORIZON_MONTHS: int = 12
# ──────────────────────────────────────────────────────────────────────────────
# docx-специфичные микро-билдеры. Все принимают `doc`/контейнер python-docx и пишут
# в него (мутируют документ). Хелперы нормализации/форматирования — импортированы из
# report_pdf, не дублируются (DRY). Числа/None уже причёсаны `_fmt` → str.
# ──────────────────────────────────────────────────────────────────────────────
def _add_kv_lines(doc: _DocxDocument, pairs: list[tuple[str, Any]]) -> None:
"""Карточка «**метка:** значение» построчно (bullet-абзацы). Пустой → «нет данных».
Метка — статичная RU-строка (bold-run), значение — через `_fmt` (None → "").
Зеркало `report_md._md_kv_lines` (там Markdown-буллеты; здесь — docx-абзацы).
"""
if not pairs:
doc.add_paragraph(_NO_DATA)
return
for label, value in pairs:
para = doc.add_paragraph(style="List Bullet")
run = para.add_run(f"{label}: ")
run.bold = True
para.add_run(_fmt(value))
def _add_table(doc: _DocxDocument, headers: list[str], rows: list[list[Any]]) -> None:
"""Таблица: шапка (статичные RU-метки, bold) + строки данных (через `_fmt`). Graceful.
Пустой `rows` → одна строка-заглушка «нет данных» под шапкой (таблица всё равно
валидна). Все ячейки данных проходят `_fmt` (str, None → ""). Зеркало
`report_md._md_table` (GFM-таблица) — здесь python-docx Table со стилем «Table Grid».
"""
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):
cell = header_cells[idx]
cell.text = ""
run = cell.paragraphs[0].add_run(header)
run.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_kv_table(doc: _DocxDocument, data: dict[str, Any]) -> None:
"""Таблица «Показатель → Значение» из плоского dict. Пустой → «нет данных».
Ключи — данные (тоже через `_fmt`). Зеркало `report_md._md_kv_table`.
"""
rows = [[str(key), value] for key, value in data.items()]
_add_table(doc, ["Показатель", "Значение"], rows)
def _add_heading(doc: _DocxDocument, text: str, level: int) -> None:
"""Заголовок секции (python-docx Heading level). Тонкая обёртка для единообразия."""
doc.add_heading(text, level=level)
def _join_horizons(values: list[Any]) -> Any:
"""Свернуть список горизонтов в «6, 12, 18» или None (→ `_fmt` отдаст ""). PURE."""
return ", ".join(str(v) for v in values) if values else None
# ──────────────────────────────────────────────────────────────────────────────
# Построители секций — по одной на содержательную секцию §22, тот же порядок/набор,
# что и в report_md/report_pdf. Каждый graceful (пустая секция → «нет данных», не
# падает) и ПИШЕТ в переданный `doc` (мутирует документ).
# ──────────────────────────────────────────────────────────────────────────────
def _build_header(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""Титул + advisory-дисклеймер + карточка meta (cad/район/горизонты/дата/схема)."""
meta = _as_dict(report.get("meta"))
schema_version = report.get("schema_version") or meta.get("schema_version")
doc.add_heading(_TITLE_DOC, level=0)
disclaimer = doc.add_paragraph(style="Intense Quote")
run = disclaimer.add_run(f"{_ADVISORY_DISCLAIMER} {_ADVISORY_MARKER}")
run.bold = True
meta_pairs: list[tuple[str, Any]] = [
("Кадастровый номер", meta.get("cad_num")),
("Район", meta.get("district")),
("Сегмент", meta.get("segment")),
("Горизонты (мес)", _join_horizons(_as_list(meta.get("horizons")))),
("Сформировано", meta.get("generated_at")),
("Версия схемы", schema_version),
]
_add_kv_lines(doc, meta_pairs)
def _build_summary(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.1 «Сводка»: headline (bold) + вердикт + общая уверенность + ключевые числа."""
exec_summary = _as_dict(report.get("exec_summary"))
headline = exec_summary.get("headline")
verdict = exec_summary.get("verdict")
key_numbers = _as_dict(exec_summary.get("key_numbers"))
overall_conf = _level_ru(exec_summary.get("overall_confidence"))
_add_heading(doc, _TITLE_SUMMARY, level=1)
headline_para = doc.add_paragraph()
headline_para.add_run(_fmt(headline)).bold = True
doc.add_paragraph(_fmt(verdict))
conf_para = doc.add_paragraph()
conf_para.add_run(f"Общая уверенность: {overall_conf}").italic = True
_add_heading(doc, "Ключевые числа", level=2)
_add_kv_table(doc, key_numbers)
def _build_market_now(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.2 «Рынок сейчас»: резюме + метрики + слои предложения + конкуренты (top-N)."""
market_now = _as_dict(report.get("market_now"))
metrics = _as_dict(market_now.get("market_metrics"))
supply = _as_dict(market_now.get("supply_layers"))
competitors = _as_list(market_now.get("competitors"))
comp_rows = [
[c.get("comm_name"), c.get("obj_id"), c.get("relevance_weight")]
for c in competitors[:_COMPETITORS_TOP_N]
if isinstance(c, dict)
]
_add_heading(doc, _TITLE_MARKET_NOW, level=1)
doc.add_paragraph(_fmt(market_now.get("summary")))
_add_heading(doc, "Метрики рынка", level=2)
_add_kv_table(doc, metrics)
_add_heading(doc, "Слои предложения", level=2)
_add_kv_table(doc, supply)
_add_heading(doc, "Конкуренты", level=2)
_add_table(doc, ["ЖК", "ID", "Релевантность"], comp_rows)
def _build_future_market(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.3 «Будущий рынок»: таблица по горизонтам + давление предложения + сценарии."""
future = _as_dict(report.get("future_market"))
forecasts = _as_list(future.get("forecasts_by_horizon"))
scenarios_summary = _as_dict(future.get("scenarios_summary"))
forecast_headers = [
"Горизонт, мес",
"Индекс дефицита",
"Месяцев запаса",
"Спрос",
"Предложение",
"Ставка",
"Уверенность",
]
forecast_rows = [
[
f.get("horizon_months"),
f.get("deficit_index"),
f.get("months_of_inventory"),
f.get("projected_demand_units"),
f.get("projected_supply_units"),
f.get("rate_future"),
_level_ru(f.get("confidence")),
]
for f in forecasts
if isinstance(f, dict)
]
_add_heading(doc, _TITLE_FUTURE_MARKET, level=1)
doc.add_paragraph(_fmt(future.get("summary")))
_add_heading(doc, "Прогноз по горизонтам", level=2)
_add_table(doc, forecast_headers, forecast_rows)
_add_heading(doc, "Давление будущего предложения", level=2)
_add_kv_table(doc, _future_supply_pairs(future.get("future_supply")))
_add_heading(doc, "Разброс сценариев (индекс дефицита)", level=2)
_add_kv_table(doc, scenarios_summary)
def _build_scenarios(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.5 «Сценарии»: base/aggressive/conservative — дефицит (12 мес) + rate_path."""
scenarios = _as_dict(report.get("scenarios"))
by_scenario = _as_dict(scenarios.get("by_scenario"))
rows: list[list[Any]] = []
for name, payload in by_scenario.items():
data = _as_dict(payload)
rate_path = _as_dict(data.get("rate_path"))
rate_str = (
", ".join(f"{k}: {_fmt(v)}" for k, v in rate_path.items()) if rate_path else None
)
rows.append([name, _scenario_deficit_cell(data), rate_str, data.get("advisory")])
headers = [
"Сценарий",
f"Индекс дефицита ({_PRIMARY_HORIZON_MONTHS} мес)",
"Траектория ставки",
"Advisory",
]
_add_heading(doc, _TITLE_SCENARIOS, level=1)
doc.add_paragraph(_fmt(scenarios.get("summary")))
_add_table(doc, headers, rows)
def _build_product_tz(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.4 «Продукт ТЗ»: класс + квартирография (mix) + коммерция + USP + причины."""
product = _as_dict(report.get("product_tz"))
mix = _as_list(product.get("mix"))
commercial = _as_dict(product.get("commercial"))
usp = _as_list(product.get("usp"))
reasons = _as_list(product.get("reasons"))
head_pairs: list[tuple[str, Any]] = [
("Резюме", product.get("summary")),
("Рекомендованный класс", product.get("obj_class")),
]
mix_rows = [
[m.get("bucket"), m.get("pct"), m.get("obj_class"), m.get("deficit_index")]
for m in mix
if isinstance(m, dict)
]
usp_rows = [[u.get("segment"), u.get("usp_text")] for u in usp if isinstance(u, dict)]
reason_rows = [[r.get("why"), r.get("advisory")] for r in reasons if isinstance(r, dict)]
_add_heading(doc, _TITLE_PRODUCT_TZ, level=1)
_add_kv_lines(doc, head_pairs)
_add_heading(doc, "Квартирография (mix)", level=2)
_add_table(doc, ["Формат", "Доля, %", "Класс", "Индекс дефицита"], mix_rows)
_add_heading(doc, "Коммерция", level=2)
_add_kv_table(doc, commercial)
_add_heading(doc, "USP-ниши", level=2)
_add_table(doc, ["Сегмент", "USP"], usp_rows)
_add_heading(doc, "Причины (§16)", level=2)
_add_table(doc, ["Почему", "Advisory"], reason_rows)
def _build_scoring(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.6 «Скоринг»: продуктовые скоры (#985) + спец-индексы (#986) + overall."""
scoring = _as_dict(report.get("scoring"))
product_scores = _as_dict(scoring.get("product_scores"))
special_indices = _as_dict(scoring.get("special_indices"))
scores = _as_dict(product_scores.get("scores"))
score_rows = [[name, _as_dict(payload).get("value")] for name, payload in scores.items()]
indices = _as_dict(special_indices.get("indices"))
index_rows = [
[name, _as_dict(payload).get("value"), _as_dict(payload).get("label")]
for name, payload in indices.items()
]
_add_heading(doc, _TITLE_SCORING, level=1)
overall_para = doc.add_paragraph()
overall_para.add_run(f"Итоговый скор (overall): {_fmt(scoring.get('overall'))}").italic = True
_add_heading(doc, "Продуктовые скоры", level=2)
_add_table(doc, ["Скор", "Значение"], score_rows)
_add_heading(doc, "Специальные индексы", level=2)
_add_table(doc, ["Индекс", "Значение", "Метка"], index_rows)
def _build_confidence(doc: _DocxDocument, report: dict[str, Any]) -> None:
"""§22.7 «Уверенность»: уровень + обоснование + факторы-драйверы (таблица)."""
confidence = _as_dict(report.get("confidence"))
level = _level_ru(confidence.get("level"))
rationale = confidence.get("rationale")
factors = _as_dict(confidence.get("factors"))
# Факторы #990: {name: {value, level, note}} ИЛИ плоское {name: value}. Defensive:
# если значение — dict, раскладываем на value/level/note; иначе кладём как есть.
factor_rows: list[list[Any]] = []
for name, payload in factors.items():
if isinstance(payload, dict):
factor_rows.append(
[name, payload.get("value"), _level_ru(payload.get("level")), payload.get("note")]
)
else:
factor_rows.append([name, payload, _DASH, _DASH])
_add_heading(doc, _TITLE_CONFIDENCE, level=1)
level_para = doc.add_paragraph()
level_para.add_run(f"Уровень: {level}").italic = True
doc.add_paragraph(_fmt(rationale))
_add_heading(doc, "Факторы уверенности", level=2)
_add_table(doc, ["Фактор", "Значение", "Уровень", "Комментарий"], factor_rows)
# Реестр построителей. Порядок = порядок блоков в документе (зеркало report_md:
# Сводка → Рынок сейчас → Будущий рынок → Сценарии → Продукт ТЗ → Скоринг →
# Уверенность; header идёт отдельно первым).
_SECTION_BUILDERS: tuple[Any, ...] = (
_build_summary,
_build_market_now,
_build_future_market,
_build_scenarios,
_build_product_tz,
_build_scoring,
_build_confidence,
)
# ──────────────────────────────────────────────────────────────────────────────
# Публичный API — рендер отчёта в DOCX-байты (без файлового I/O на диск).
# ──────────────────────────────────────────────────────────────────────────────
def render_report_docx(report: Any) -> bytes:
"""§22 Отрендерить `SiteFinderReport` (#987) в Word-документ (.docx) и вернуть БАЙТЫ.
Титул + advisory-дисклеймер + карточка meta, далее семь содержательных секций §22
(Сводка / Рынок сейчас / Будущий рынок / Сценарии / Продукт ТЗ / Скоринг /
Уверенность) — содержание и порядок ЗЕРКАЛЯТ `report_md`/`report_pdf`. Числа
форматируются через импортированный `_fmt` (None → ""), уровни уверенности — RU-
метками. Заголовки python-docx (Heading 0-2), таблицы со стилем «Table Grid».
ДЕТЕРМИНИРОВАННО, БЕЗ LLM/БД/сети. Принимает КАК `SiteFinderReport`-инстанс, ТАК и
его `as_dict()`-словарь (нормализуется через импортированный `_normalize`). GRACEFUL:
частичный/пустой/мусорный отчёт → секции с «нет данных», НИКОГДА не падает (нет
KeyError на тонком отчёте) — возвращает минимальный валидный .docx.
`python-docx` импортируется ЛОКАЛЬНО (как WeasyPrint в `report_pdf`): не нужен при
импорте модуля. Документ собирается в памяти (`io.BytesIO`) — без файлового I/O.
Args:
report: `SiteFinderReport`-инстанс или его `as_dict()`-словарь (или мусор → {}).
Returns:
Непустые DOCX-байты (OOXML — это zip, начинаются с `b"PK"`), готовые для
`Response(media_type=
"application/vnd.openxmlformats-officedocument.wordprocessingml.document")`.
"""
# python-docx импортируем локально — тяжёлый (lxml); не нужен при импорте модуля.
try:
from docx import Document
except ImportError as exc:
raise RuntimeError(
"python-docx не установлен. Добавь 'python-docx>=1.1.0' в pyproject.toml."
) from exc
data = _normalize(report)
doc = Document()
_build_header(doc, data)
for builder in _SECTION_BUILDERS:
builder(doc, data)
buffer = io.BytesIO()
doc.save(buffer)
docx_bytes = buffer.getvalue()
meta = _as_dict(data.get("meta"))
logger.info(
"render_report_docx: cad_num=%s sections=%d size=%d bytes advisory=%s",
meta.get("cad_num"),
len(_SECTION_BUILDERS),
len(docx_bytes),
data.get("advisory"),
)
return docx_bytes