Когда demand_normalization honesty-gate отбивает β rate-sensitivity (`sensitivity.confidence=='low'`, ЕКБ-регрессия не проходит n≥30/R²≥0.1/slope<0), все 393 прод-отчёта получают coefficient=1.0 для всех трёх сценариев conservative/base/aggressive. По дизайну backend честно деградирует, но фронт рисует 3 разноцветные карточки/линии одинаковых чисел без caveat. Backend (scenarios.py, report.py, report_assembler.py, orchestrator.py): - compute_scenarios теперь возвращает (list, collapsed: bool, reason: str|None) - _detect_collapsed(): math.isclose(rel_tol=1e-9, abs_tol=1e-6) сравнивает projected_demand_units И deficit_index на всех горизонтах между conservative и aggressive — расхождение в любой метрике на любом h → False - _COLLAPSE_REASON_LOW_BETA — единственный источник истины каноничного русского предложения - ReportScenarios получает поля scenarios_collapsed + scenarios_collapse_reason - demand_normalization.py НЕ ТРОНУТ — поведение корректно по контракту Backend экспортёры (report_pdf.py, excel.py): - PDF: при collapse — тёмная плашка-параграф вместо таблицы трёх столбцов - Excel: ОДНА строка «base» + caveat-ячейка вместо трёх идентичных столбцов Frontend (forecast.ts, ScenariosBlock, ScenarioCards, ScenarioCompareTable, ForecastChart, Section6Forecast, ptica.module.css): - Тип ReportScenarios расширен двумя optional-полями - ScenariosBlock (light): headline-bar + одна grid-карточка base + caveat - ScenarioCards (ptica dark): sub-component ScenarioBaseCard, одна карточка + collapse-note, ScenarioCompareTable return null - ForecastChart: effectiveScenarios filter — одна линия viz-1 «Базовый» вместо трёх перекрывающихся - Section6Forecast: caveat выше графика, читается ТОЛЬКО из scenarios_collapse_reason (источник истины — backend constant) Тесты: +18 новых (TestMetricsEqual×6, TestDetectCollapsed×7, TestComputeScenariosCollapseDetection×4 + 1 проверка graceful + corner). 102 forecasting passed, 88 cross-module passed. Refs #1871
708 lines
33 KiB
Python
708 lines
33 KiB
Python
"""§13 PDF-экспортёр итогового советующего отчёта Site Finder v2 (WeasyPrint).
|
||
|
||
#989 (955-A3, Site Finder v2 / «GG-форсайт» ТЗ §13), EPIC 11 «Отчёт». Рендерит
|
||
`SiteFinderReport` (#987) в PDF-документ (по одной секции §13 на блок) и возвращает
|
||
БАЙТЫ — готовые для `Response(media_type="application/pdf")` / записи. Это АДДИТИВНЫЙ
|
||
двойник `excel.export_report_xlsx` (#991): зеркалит его house-style ОДИН-В-ОДИН —
|
||
module-level `logger`, `_normalize` (приём инстанса ИЛИ `as_dict()`-словаря), named-
|
||
константы заголовков/микрокопии/CSS, RU-метки, graceful-None, чистый возврат `bytes`.
|
||
|
||
ДЕТЕРМИНИРОВАННЫЙ, БЕЗ LLM, БЕЗ БД/сети: только ПОТРЕБЛЯЕТ уже-собранный отчёт (его
|
||
наполняет сборщик #988) и раскладывает по секциям. Принимает КАК `SiteFinderReport`-
|
||
инстанс, ТАК и его `as_dict()`-словарь (нормализуется через `_normalize`).
|
||
|
||
GRACEFUL (дух всего форсайт-стека): частичный/пустой отчёт ВАЛИДЕН — пустая секция
|
||
рисует «нет данных», экспортёр НИКОГДА не падает. Отчёт СОВЕТУЮЩИЙ: на блоке «Сводка»
|
||
проставляется заметный ADVISORY-маркер (оценка не основание для инвест-решения).
|
||
|
||
БЕЗОПАСНОСТЬ: ВСЕ динамические строковые поля экранируются `html.escape` (микрокопия
|
||
RU-шаблонная, но трактуем вход как недоверенный — отчёт могут наполнить из внешних
|
||
источников). WeasyPrint импортируется ЛОКАЛЬНО внутри функции (тяжёлая зависимость —
|
||
не нужна при импорте модуля; зеркало `snapshot_pdf`).
|
||
|
||
Подключение к эндпоинту экспорта — #959 (вне scope: здесь только сам экспортёр).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import html
|
||
import logging
|
||
from typing import Any
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Named-константы: заголовки секций (по одной на содержательную секцию §13) ──
|
||
# Тот же набор из шести содержательных секций, что рисует excel.py (#991).
|
||
|
||
_TITLE_DOC: str = "Site Finder v2 — советующий отчёт §13"
|
||
_TITLE_SUMMARY: str = "Сводка"
|
||
_TITLE_MARKET_NOW: str = "Рынок сейчас"
|
||
_TITLE_FUTURE_MARKET: str = "Будущий рынок"
|
||
_TITLE_PRODUCT_TZ: str = "Продукт ТЗ"
|
||
_TITLE_SCENARIOS: str = "Сценарии"
|
||
_TITLE_SCORING: str = "Скоринг"
|
||
|
||
# ── Named-константы: микрокопия / плейсхолдеры (зеркало excel.py) ──────────────
|
||
|
||
# Заглушка пустой ячейки/поля (зеркало `_DASH` у excel.py).
|
||
_DASH: str = "—"
|
||
# Текст для полностью пустой секции (graceful: блок есть, данных нет).
|
||
_NO_DATA: str = "нет данных"
|
||
# ADVISORY-маркер на блоке «Сводка» — отчёт советующий, не основание для решения.
|
||
_ADVISORY_MARKER: str = (
|
||
"ADVISORY: отчёт носит советующий характер и не является основанием "
|
||
"для инвестиционного решения."
|
||
)
|
||
# Округление чисел по умолчанию (скоры/индексы ∈ [0,1] и доли — 3 знака читаемо).
|
||
_ROUND_DIGITS: int = 3
|
||
# Основной продуктовый горизонт (мес) — из него тянем сводный deficit_index сценария
|
||
# (зеркало report_assembler._PRIMARY_HORIZON_MONTHS). ScenarioForecast.as_dict() несёт
|
||
# список forecasts по горизонтам, скалярного «overall» у сценария НЕТ — берём дефицит
|
||
# основного горизонта как сводный показатель сценария.
|
||
_PRIMARY_HORIZON_MONTHS: int = 12
|
||
# Сноска-футер на каждой странице (зеркало advisory-духа форсайт-стека).
|
||
_FOOTER_NOTE: str = "GenDesign · Site Finder v2 · советующий отчёт §13 (не оферта)"
|
||
|
||
# RU-метки уровней отчётной уверенности (§15) — зеркало `_LEVEL_RU` у excel.py.
|
||
_LEVEL_RU: dict[str, str] = {
|
||
"high": "высокая",
|
||
"medium": "средняя",
|
||
"low": "низкая",
|
||
}
|
||
|
||
# ── Named-константы: CSS (house-accent синий, янтарный ADVISORY — как у соседей) ──
|
||
|
||
_CSS: str = """
|
||
@page {
|
||
size: A4;
|
||
margin: 18mm 16mm 18mm 16mm;
|
||
@bottom-center {
|
||
content: "Site Finder v2 · советующий отчёт §13 · стр. " 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: 18pt; margin: 0 0 4pt 0; line-height: 1.2; color: #fff; }
|
||
h2 {
|
||
font-size: 13pt; 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; }
|
||
.section { margin-bottom: 16pt; }
|
||
.section:last-child { margin-bottom: 0; }
|
||
|
||
/* Cover / summary header */
|
||
.cover-header {
|
||
background: #1d4ed8; color: #fff;
|
||
padding: 16pt 18pt; border-radius: 6pt; margin-bottom: 14pt;
|
||
}
|
||
.cover-meta { font-size: 9pt; color: rgba(255,255,255,0.85); margin-top: 4pt; }
|
||
.advisory {
|
||
margin: 0 0 14pt 0; padding: 10pt 12pt;
|
||
background: #fffbeb; border-left: 4px solid #b45309;
|
||
border-radius: 0 4pt 4pt 0; font-size: 9.5pt; color: #92400e; font-weight: 700;
|
||
}
|
||
.headline { font-size: 13pt; font-weight: 700; margin: 0 0 6pt 0; color: #111827; }
|
||
.verdict { font-size: 10pt; color: #374151; margin: 0 0 8pt 0; }
|
||
|
||
/* Key-value rows */
|
||
.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: 38%; }
|
||
.kv td.v { color: #1a1d23; }
|
||
|
||
/* Tables */
|
||
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: 12pt; }
|
||
.no-data { padding: 6pt 0; }
|
||
|
||
/* Confidence badge */
|
||
.badge {
|
||
display: inline-block; padding: 2pt 8pt;
|
||
border-radius: 6pt; font-size: 9pt; font-weight: 700;
|
||
background: #eff6ff; color: #1d4ed8; border: 1px solid #bfdbfe;
|
||
}
|
||
|
||
/* Footer */
|
||
.footer {
|
||
margin-top: 18pt; padding-top: 8pt;
|
||
border-top: 1px solid #e6e8ec; font-size: 8pt; color: #9ca3af;
|
||
}
|
||
|
||
/* §22 audit-issue #1871 P1: collapse-плашка вместо таблицы трёх сценариев. */
|
||
.scenarios-collapse {
|
||
margin: 0 0 10pt 0; padding: 12pt 14pt;
|
||
background: #1f2937; color: #f9fafb;
|
||
border-radius: 4pt;
|
||
}
|
||
.scenarios-collapse-title {
|
||
font-size: 11pt; font-weight: 700; margin-bottom: 6pt; color: #fef3c7;
|
||
}
|
||
.scenarios-collapse-body {
|
||
font-size: 9.5pt; line-height: 1.4; color: #e5e7eb;
|
||
}
|
||
"""
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Нормализация входа + микро-хелперы форматирования. Все PURE, graceful. Зеркало
|
||
# `_normalize`/`_fmt`/`_as_dict`/`_as_list`/`_level_ru` у excel.py (#991).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _normalize(report: Any) -> dict[str, Any]:
|
||
"""Привести вход к `as_dict()`-словарю отчёта (dataclass | dict). PURE, graceful.
|
||
|
||
Принимаем КАК `SiteFinderReport`-инстанс (есть метод `as_dict`), ТАК и уже его
|
||
`as_dict()`-словарь — вызывающий (#959) волен передать любое. Любой иной тип
|
||
(мусор/None) → пустой dict: экспортёр отрисует пустой отчёт, а не упадёт.
|
||
"""
|
||
as_dict = getattr(report, "as_dict", None)
|
||
if callable(as_dict):
|
||
result = as_dict()
|
||
return result if isinstance(result, dict) else {}
|
||
if isinstance(report, dict):
|
||
return report
|
||
return {}
|
||
|
||
|
||
def _fmt(value: Any) -> str:
|
||
"""Привести значение к ТЕКСТУ ячейки PDF: None → "—", float → округление. PURE.
|
||
|
||
Числа округляем до `_ROUND_DIGITS` (но bool → «да»/«нет» — это не число для
|
||
округления). Строки/прочие скаляры — как есть (str). Контейнеры (dict/list)
|
||
сворачиваем в компактную строку (на случай неожиданно-вложенного значения — не
|
||
падаем). Возврат — ВСЕГДА str (его экранирует `_esc` перед вставкой в HTML).
|
||
"""
|
||
if value is None:
|
||
return _DASH
|
||
if isinstance(value, bool):
|
||
return "да" if value else "нет"
|
||
if isinstance(value, float):
|
||
return str(round(value, _ROUND_DIGITS))
|
||
if isinstance(value, str):
|
||
return value
|
||
return str(value)
|
||
|
||
|
||
def _esc(value: Any) -> str:
|
||
"""Привести значение к ячейке (`_fmt`) и ЭКРАНИРОВАТЬ для HTML. PURE.
|
||
|
||
Единая точка вставки динамических данных в разметку: всё, что попадает в HTML,
|
||
проходит `html.escape` (вход трактуем как недоверенный, см. модульный docstring).
|
||
"""
|
||
return html.escape(_fmt(value))
|
||
|
||
|
||
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 _level_ru(level: Any) -> str:
|
||
"""RU-метка уровня уверенности (high/medium/low) или "—". PURE."""
|
||
if isinstance(level, str) and level in _LEVEL_RU:
|
||
return _LEVEL_RU[level]
|
||
return _DASH
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Низкоуровневые HTML-помощники (kv-строки / таблицы / заглушки). Все PURE: строят
|
||
# строку HTML из уже-экранированных кусочков, в БД/сеть не ходят.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
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:
|
||
"""Таблица «метка → значение» из списка пар. Пустой список → «нет данных»."""
|
||
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 _dict_kv_table(data: dict[str, Any]) -> str:
|
||
"""Таблица «ключ → значение» из плоского dict (ключи — данные, тоже экранируем)."""
|
||
if not data:
|
||
return _no_data()
|
||
rows = "".join(_kv_row(str(key), value) for key, value in data.items())
|
||
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>'
|
||
|
||
|
||
def _no_data() -> str:
|
||
"""Блок-заглушка «нет данных» для пустой секции (graceful). PURE."""
|
||
return f'<div class="no-data">{_NO_DATA}</div>'
|
||
|
||
|
||
def _join_ints(values: list[Any]) -> Any:
|
||
"""Свернуть список горизонтов в строку «6, 12, 18» или None (для `_fmt` → "—")."""
|
||
return ", ".join(str(v) for v in values) if values else None
|
||
|
||
|
||
def _future_supply_pairs(future_supply: Any) -> dict[str, Any]:
|
||
"""Плоская карта «метка → значение» по FutureSupplyPressure.as_dict() (§9.3). PURE.
|
||
|
||
`future_supply` = `FutureSupplyPressure.as_dict()`: верхнеуровневый `index`/контекст +
|
||
вложенный `breakdown` (объёмы по слоям, поглощение, месяцы давления). Разворачиваем
|
||
breakdown на верхний уровень с RU-метками. Пустой/None → {} (graceful → «нет данных»).
|
||
"""
|
||
data = _as_dict(future_supply)
|
||
if not data:
|
||
return {}
|
||
breakdown = _as_dict(data.get("breakdown"))
|
||
return {
|
||
"Индекс давления": data.get("index"),
|
||
"Уверенность": _level_ru(data.get("confidence")),
|
||
"Открытый сток (ед.)": breakdown.get("open_units"),
|
||
"Скрытый запас (ед.)": breakdown.get("hidden_units"),
|
||
"Будущий слой в горизонте (ед.)": breakdown.get("future_units_by_horizon"),
|
||
"Поглощение (ед./мес)": breakdown.get("monthly_absorption_units"),
|
||
"Месяцев давления": breakdown.get("months_of_pressure"),
|
||
}
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Построители секций — по одной на содержательную секцию §13. Каждый graceful:
|
||
# пустая секция → «нет данных», НИКОГДА не падает. Возвращают HTML-фрагмент.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_summary(report: dict[str, Any]) -> str:
|
||
"""Блок «Сводка»: cover + ADVISORY-маркер + вердикт + ключевые числа + контекст."""
|
||
exec_summary = _as_dict(report.get("exec_summary"))
|
||
meta = _as_dict(report.get("meta"))
|
||
confidence = _as_dict(report.get("confidence"))
|
||
|
||
headline = exec_summary.get("headline")
|
||
verdict = exec_summary.get("verdict")
|
||
key_numbers = _as_dict(exec_summary.get("key_numbers"))
|
||
factors = _as_dict(confidence.get("factors"))
|
||
|
||
cad = _esc(meta.get("cad_num"))
|
||
district = _esc(meta.get("district"))
|
||
horizons = _join_ints(_as_list(meta.get("horizons")))
|
||
|
||
context_pairs: list[tuple[str, Any]] = [
|
||
("Кадастровый номер", meta.get("cad_num")),
|
||
("Район", meta.get("district")),
|
||
("Горизонты (мес)", horizons),
|
||
("Сформировано", meta.get("generated_at")),
|
||
("Версия схемы", meta.get("schema_version")),
|
||
]
|
||
confidence_pairs: list[tuple[str, Any]] = [
|
||
("Уровень", _level_ru(confidence.get("level"))),
|
||
("Обоснование", confidence.get("rationale")),
|
||
]
|
||
overall_conf = _esc(_level_ru(exec_summary.get("overall_confidence")))
|
||
|
||
return f"""
|
||
<div class="section" id="summary">
|
||
<div class="cover-header">
|
||
<h1>{html.escape(_TITLE_DOC)}</h1>
|
||
<div class="cover-meta">Кадастровый номер: {cad} · Район: {district}</div>
|
||
</div>
|
||
<div class="advisory">{html.escape(_ADVISORY_MARKER)}</div>
|
||
|
||
<h2>{html.escape(_TITLE_SUMMARY)}</h2>
|
||
<p class="headline">{_esc(headline)}</p>
|
||
<p class="verdict">{_esc(verdict)}</p>
|
||
<span class="badge">Общая уверенность: {overall_conf}</span>
|
||
|
||
<h3>Ключевые числа</h3>
|
||
{_dict_kv_table(key_numbers)}
|
||
|
||
<h3>Уверенность отчёта</h3>
|
||
{_kv_table(confidence_pairs)}
|
||
|
||
<h3>Факторы уверенности</h3>
|
||
{_dict_kv_table(factors)}
|
||
|
||
<h3>Контекст</h3>
|
||
{_kv_table(context_pairs)}
|
||
</div>
|
||
"""
|
||
|
||
|
||
def _build_market_now(report: dict[str, Any]) -> str:
|
||
"""Блок «Рынок сейчас»: резюме + метрики + слои предложения + конкуренты."""
|
||
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
|
||
if isinstance(c, dict)
|
||
]
|
||
|
||
return f"""
|
||
<div class="section" id="market-now">
|
||
<h2>{html.escape(_TITLE_MARKET_NOW)}</h2>
|
||
<p class="verdict">{_esc(market_now.get("summary"))}</p>
|
||
|
||
<h3>Метрики рынка</h3>
|
||
{_dict_kv_table(metrics)}
|
||
|
||
<h3>Слои предложения</h3>
|
||
{_dict_kv_table(supply)}
|
||
|
||
<h3>Конкуренты</h3>
|
||
{_data_table(["ЖК", "ID", "Релевантность"], comp_rows)}
|
||
</div>
|
||
"""
|
||
|
||
|
||
def _build_future_market(report: dict[str, Any]) -> str:
|
||
"""Блок «Будущий рынок»: резюме + прогноз по горизонтам + разброс сценариев."""
|
||
future = _as_dict(report.get("future_market"))
|
||
forecasts = _as_list(future.get("forecasts_by_horizon"))
|
||
scenarios_summary = _as_dict(future.get("scenarios_summary"))
|
||
|
||
forecast_rows = [
|
||
[
|
||
f.get("horizon_months"),
|
||
f.get("projected_demand_units"),
|
||
f.get("projected_supply_units"),
|
||
f.get("deficit_index"),
|
||
_level_ru(f.get("confidence")),
|
||
]
|
||
for f in forecasts
|
||
if isinstance(f, dict)
|
||
]
|
||
forecast_headers = ["Горизонт, мес", "Спрос", "Предложение", "Индекс дефицита", "Уверенность"]
|
||
|
||
return f"""
|
||
<div class="section" id="future-market">
|
||
<h2>{html.escape(_TITLE_FUTURE_MARKET)}</h2>
|
||
<p class="verdict">{_esc(future.get("summary"))}</p>
|
||
|
||
<h3>Прогноз по горизонтам</h3>
|
||
{_data_table(forecast_headers, forecast_rows)}
|
||
|
||
<h3>Давление будущего предложения</h3>
|
||
{_dict_kv_table(_future_supply_pairs(future.get("future_supply")))}
|
||
|
||
<h3>Разброс сценариев (индекс дефицита)</h3>
|
||
{_dict_kv_table(scenarios_summary)}
|
||
</div>
|
||
"""
|
||
|
||
|
||
def _build_product_tz(report: dict[str, Any]) -> str:
|
||
"""Блок «Продукт ТЗ»: класс + квартирография (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"))
|
||
|
||
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)]
|
||
|
||
head_pairs: list[tuple[str, Any]] = [
|
||
("Резюме", product.get("summary")),
|
||
("Рекомендованный класс", product.get("obj_class")),
|
||
]
|
||
|
||
return f"""
|
||
<div class="section" id="product-tz">
|
||
<h2>{html.escape(_TITLE_PRODUCT_TZ)}</h2>
|
||
{_kv_table(head_pairs)}
|
||
|
||
<h3>Квартирография (mix)</h3>
|
||
{_data_table(["Формат", "Доля, %", "Класс", "Индекс дефицита"], mix_rows)}
|
||
|
||
<h3>Коммерция</h3>
|
||
{_dict_kv_table(commercial)}
|
||
|
||
<h3>USP-ниши</h3>
|
||
{_data_table(["Сегмент", "USP"], usp_rows)}
|
||
</div>
|
||
"""
|
||
|
||
|
||
def _scenario_deficit_index(payload: dict[str, Any]) -> Any:
|
||
"""Сводный deficit_index сценария = дефицит основного горизонта (12 мес). PURE.
|
||
|
||
`payload` = `ScenarioForecast.as_dict()`: у сценария НЕТ скалярного «overall» — есть
|
||
список `forecasts` по горизонтам, каждый с `deficit_index`. Берём горизонт
|
||
`_PRIMARY_HORIZON_MONTHS`, иначе первый с не-None дефицитом. Нет → None (→ "—").
|
||
|
||
NB: эту функцию импортируют report_md/docx/pptx — НЕ менять сигнатуру (вернёт скаляр).
|
||
Для подписи фактического горизонта используй `_scenario_deficit_horizon`.
|
||
"""
|
||
forecasts = _as_list(payload.get("forecasts"))
|
||
primary = next(
|
||
(
|
||
f
|
||
for f in forecasts
|
||
if isinstance(f, dict) and f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS
|
||
),
|
||
None,
|
||
)
|
||
if primary is not None and primary.get("deficit_index") is not None:
|
||
return primary.get("deficit_index")
|
||
for f in forecasts:
|
||
if isinstance(f, dict) and f.get("deficit_index") is not None:
|
||
return f.get("deficit_index")
|
||
return None
|
||
|
||
|
||
def _scenario_deficit_horizon(payload: dict[str, Any]) -> int | None:
|
||
"""Фактический горизонт (мес) дефицита, выбранного `_scenario_deficit_index`. PURE.
|
||
|
||
Зеркалит порядок выбора `_scenario_deficit_index` (primary 12 мес → первый с не-None
|
||
дефицитом), но возвращает `horizon_months` выбранного прогноза, чтобы подпись не лгала
|
||
при fallback на чужой горизонт. Нет дефицита → None. NB: сигнатура
|
||
`_scenario_deficit_index` (импортируется в report_md/docx/pptx) НЕ меняется — это
|
||
отдельная чистая функция-зеркало.
|
||
"""
|
||
forecasts = _as_list(payload.get("forecasts"))
|
||
primary = next(
|
||
(
|
||
f
|
||
for f in forecasts
|
||
if isinstance(f, dict) and f.get("horizon_months") == _PRIMARY_HORIZON_MONTHS
|
||
),
|
||
None,
|
||
)
|
||
if primary is not None and primary.get("deficit_index") is not None:
|
||
return _PRIMARY_HORIZON_MONTHS
|
||
for f in forecasts:
|
||
if isinstance(f, dict) and f.get("deficit_index") is not None:
|
||
h = f.get("horizon_months")
|
||
return h if isinstance(h, int) else None
|
||
return None
|
||
|
||
|
||
def _scenario_deficit_cell(payload: dict[str, Any]) -> Any:
|
||
"""Ячейка дефицита сценария: значение + «(гор. N мес)» при N != основного горизонта.
|
||
|
||
PURE. При отсутствии 12-мес прогноза `_scenario_deficit_index` делает fallback на чужой
|
||
горизонт — помечаем ячейку реальным горизонтом, чтобы число под общей шапкой
|
||
«(12 мес)» не читалось как 12-месячное. Дефицит None → отдаём как есть (→ "—").
|
||
"""
|
||
deficit = _scenario_deficit_index(payload)
|
||
if deficit is None:
|
||
return deficit
|
||
horizon = _scenario_deficit_horizon(payload)
|
||
if horizon is not None and horizon != _PRIMARY_HORIZON_MONTHS:
|
||
return f"{_fmt(deficit)} (гор. {horizon} мес)"
|
||
return deficit
|
||
|
||
|
||
def _build_scenarios(report: dict[str, Any]) -> str:
|
||
"""Блок «Сценарии»: conservative/base/aggressive (таблица). Graceful.
|
||
|
||
§22 audit-issue #1871 P1 (collapse-плашка): если `scenarios_collapsed=True`
|
||
(failed β-gate в §9.6 → §9.4 coefficient=1.0 → три сценария идентичны),
|
||
рисуем тёмную плашку-объяснение ВМЕСТО таблицы трёх одинаковых столбцов.
|
||
Сохраняем deficit_index базы как информационный фон (без сравнения с
|
||
conservative/aggressive — они идентичны). Поля `scenarios_collapsed`/
|
||
`scenarios_collapse_reason` гарантированы в схеме (см. report.ReportScenarios).
|
||
"""
|
||
scenarios = _as_dict(report.get("scenarios"))
|
||
by_scenario = _as_dict(scenarios.get("by_scenario"))
|
||
|
||
collapsed = bool(scenarios.get("scenarios_collapsed"))
|
||
collapse_reason = scenarios.get("scenarios_collapse_reason")
|
||
|
||
if collapsed:
|
||
base_payload = _as_dict(by_scenario.get("base"))
|
||
base_deficit = _scenario_deficit_cell(base_payload) if base_payload else None
|
||
base_horizon = _scenario_deficit_horizon(base_payload) if base_payload else None
|
||
info_label = "Базовый сценарий"
|
||
if base_horizon is not None:
|
||
info_label += f" (индекс дефицита, гор. {base_horizon} мес)"
|
||
info_row = (
|
||
f'<table class="kv"><tr>'
|
||
f'<td class="k">{html.escape(info_label)}</td>'
|
||
f'<td class="v">{_esc(base_deficit)}</td>'
|
||
f"</tr></table>"
|
||
if base_deficit is not None
|
||
else ""
|
||
)
|
||
reason_text = (
|
||
collapse_reason
|
||
if isinstance(collapse_reason, str) and collapse_reason
|
||
else "Сценарная дифференциация недоступна на текущих данных."
|
||
)
|
||
return f"""
|
||
<div class="section" id="scenarios">
|
||
<h2>{html.escape(_TITLE_SCENARIOS)}</h2>
|
||
<div class="scenarios-collapse">
|
||
<div class="scenarios-collapse-title">Сценарная дифференциация недоступна</div>
|
||
<div class="scenarios-collapse-body">{html.escape(reason_text)}</div>
|
||
</div>
|
||
{info_row}
|
||
</div>
|
||
"""
|
||
|
||
rows: list[list[Any]] = []
|
||
for name, payload in by_scenario.items():
|
||
data = _as_dict(payload)
|
||
rows.append([name, _scenario_deficit_cell(data), data.get("advisory")])
|
||
|
||
headers = ["Сценарий", f"Индекс дефицита ({_PRIMARY_HORIZON_MONTHS} мес)", "Advisory"]
|
||
return f"""
|
||
<div class="section" id="scenarios">
|
||
<h2>{html.escape(_TITLE_SCENARIOS)}</h2>
|
||
<p class="verdict">{_esc(scenarios.get("summary"))}</p>
|
||
|
||
<h3>Сводка по сценариям</h3>
|
||
{_data_table(headers, rows)}
|
||
</div>
|
||
"""
|
||
|
||
|
||
def _build_scoring(report: dict[str, Any]) -> str:
|
||
"""Блок «Скоринг»: продуктовые скоры (#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()
|
||
]
|
||
|
||
return f"""
|
||
<div class="section" id="scoring">
|
||
<h2>{html.escape(_TITLE_SCORING)}</h2>
|
||
<span class="badge">Итоговый скор (overall): {_esc(scoring.get("overall"))}</span>
|
||
|
||
<h3>Продуктовые скоры</h3>
|
||
{_data_table(["Скор", "Значение"], score_rows)}
|
||
|
||
<h3>Специальные индексы</h3>
|
||
{_data_table(["Индекс", "Значение", "Метка"], index_rows)}
|
||
|
||
<div class="footer">{html.escape(_FOOTER_NOTE)}</div>
|
||
</div>
|
||
"""
|
||
|
||
|
||
# Реестр построителей секций. Порядок = порядок блоков в документе (зеркало
|
||
# `_SHEET_BUILDERS` у excel.py — тот же набор из шести содержательных секций §13).
|
||
_SECTION_BUILDERS: tuple[Any, ...] = (
|
||
_build_summary,
|
||
_build_market_now,
|
||
_build_future_market,
|
||
_build_product_tz,
|
||
_build_scenarios,
|
||
_build_scoring,
|
||
)
|
||
|
||
|
||
def _build_html(report: dict[str, Any]) -> str:
|
||
"""Склеить HTML-документ из шести секций §13. PURE (только строки). Graceful."""
|
||
sections = "".join(builder(report) for builder in _SECTION_BUILDERS)
|
||
return f"""<!DOCTYPE html>
|
||
<html lang="ru">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<title>{html.escape(_TITLE_DOC)}</title>
|
||
</head>
|
||
<body>
|
||
{sections}
|
||
</body>
|
||
</html>"""
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Публичный API — рендер отчёта в PDF-байты (без файлового I/O на диск).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def export_report_pdf(report: Any) -> bytes:
|
||
"""§13 Отрендерить `SiteFinderReport` (#987) в PDF-документ и вернуть БАЙТЫ.
|
||
|
||
По одному блоку на содержательную секцию §13 (Сводка / Рынок сейчас / Будущий
|
||
рынок / Продукт ТЗ / Сценарии / Скоринг — тот же набор, что и `export_report_xlsx`).
|
||
Шапки таблиц с заливкой, RU-метки, числа округлены, None → "—". На блоке «Сводка» —
|
||
заметный ADVISORY-маркер (отчёт советующий). ВСЕ динамические строки экранируются
|
||
`html.escape`.
|
||
|
||
ДЕТЕРМИНИРОВАННО, БЕЗ LLM/БД/сети. Принимает КАК `SiteFinderReport`-инстанс, ТАК и
|
||
его `as_dict()`-словарь (нормализуется через `_normalize`). GRACEFUL: частичный/
|
||
пустой отчёт → блоки с «нет данных», НИКОГДА не падает.
|
||
|
||
WeasyPrint импортируется ЛОКАЛЬНО (тяжёлая зависимость — не нужна при импорте
|
||
модуля; зеркало `snapshot_pdf`).
|
||
|
||
Args:
|
||
report: `SiteFinderReport`-инстанс или его `as_dict()`-словарь.
|
||
|
||
Returns:
|
||
Непустые PDF-байты (начинаются с `b"%PDF"`), готовые для
|
||
`Response(media_type="application/pdf")`.
|
||
"""
|
||
# WeasyPrint импортируем локально — тяжёлый; не нужен при импорте модуля.
|
||
try:
|
||
from weasyprint import CSS, HTML
|
||
except ImportError as exc:
|
||
raise RuntimeError(
|
||
"WeasyPrint не установлен. Добавь 'weasyprint>=62.0' в pyproject.toml."
|
||
) from exc
|
||
|
||
data = _normalize(report)
|
||
html_str = _build_html(data)
|
||
pdf_bytes: bytes = HTML(string=html_str).write_pdf(stylesheets=[CSS(string=_CSS)])
|
||
|
||
meta = _as_dict(data.get("meta"))
|
||
logger.info(
|
||
"export_report_pdf: cad_num=%s sections=%d size=%d bytes advisory=%s",
|
||
meta.get("cad_num"),
|
||
len(_SECTION_BUILDERS),
|
||
len(pdf_bytes),
|
||
data.get("advisory"),
|
||
)
|
||
return pdf_bytes
|