Когда 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
662 lines
34 KiB
Python
662 lines
34 KiB
Python
"""§13 Excel-экспортёр итогового советующего отчёта Site Finder v2 (openpyxl).
|
||
|
||
#991 (955-A5, Site Finder v2 / «GG-форсайт» ТЗ §13), EPIC 11 «Отчёт». Рендерит
|
||
`SiteFinderReport` (#987) в .xlsx-книгу (по одному листу на содержательную секцию §13)
|
||
и возвращает БАЙТЫ — готовые для `Response(media_type=...)` / записи. Зеркалит house-
|
||
style соседей-экспортёров (snapshot_pdf / trade_in_pdf): module-level `logger`, named-
|
||
константы стилей/заголовков, RU-микрокопия, graceful-None, чистый возврат `bytes` без
|
||
файлового I/O на диск.
|
||
|
||
ДЕТЕРМИНИРОВАННЫЙ, БЕЗ LLM, БЕЗ БД/сети: только ПОТРЕБЛЯЕТ уже-собранный отчёт (его
|
||
наполняет сборщик #988) и раскладывает по листам. Принимает КАК `SiteFinderReport`-
|
||
инстанс, ТАК и его `as_dict()`-словарь (нормализуется через `_normalize`).
|
||
|
||
GRACEFUL (дух всего форсайт-стека): частичный/пустой отчёт ВАЛИДЕН — пустая секция
|
||
рисует ячейку «нет данных», экспортёр НИКОГДА не падает. Отчёт СОВЕТУЮЩИЙ: на листе
|
||
«Сводка» проставляется ADVISORY-маркер (оценка не основание для инвест-решения).
|
||
|
||
Подключение к эндпоинту экспорта — #959 (вне scope: здесь только сам экспортёр).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from io import BytesIO
|
||
from typing import Any
|
||
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Alignment, Font, PatternFill
|
||
from openpyxl.utils import get_column_letter
|
||
from openpyxl.worksheet.worksheet import Worksheet
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ── Named-константы: заголовки листов (по одному на содержательную секцию §13) ──
|
||
|
||
_SHEET_SUMMARY: str = "Сводка"
|
||
_SHEET_MARKET_NOW: str = "Рынок сейчас"
|
||
_SHEET_FUTURE_MARKET: str = "Будущий рынок"
|
||
_SHEET_PRODUCT_TZ: str = "Продукт ТЗ"
|
||
_SHEET_SCENARIOS: str = "Сценарии"
|
||
_SHEET_SCORING: str = "Скоринг"
|
||
|
||
# ── Named-константы: микрокопия / плейсхолдеры ────────────────────────────────
|
||
|
||
# Заглушка пустой ячейки/секции (зеркало `_format_cost` → "—" у snapshot_pdf).
|
||
_DASH: str = "—"
|
||
# Текст для полностью пустой секции (graceful: лист есть, данных нет).
|
||
_NO_DATA: str = "нет данных"
|
||
# ADVISORY-маркер на листе «Сводка» — отчёт советующий, не основание для решения.
|
||
_ADVISORY_MARKER: str = (
|
||
"ADVISORY: отчёт носит советующий характер и не является основанием "
|
||
"для инвестиционного решения."
|
||
)
|
||
# Округление чисел по умолчанию (скоры/индексы ∈ [0,1] и доли — 3 знака читаемо).
|
||
_ROUND_DIGITS: int = 3
|
||
# OWASP CSV/formula-injection: лидирующие символы, с которых Excel/LibreOffice трактуют
|
||
# строку как ФОРМУЛУ/команду при открытии. Внешние/скрейпинг-данные (comm_name из DOM.РФ,
|
||
# headline/usp_text и пр.) с таким префиксом нейтрализуем (см. `_fmt`).
|
||
# https://owasp.org/www-community/attacks/CSV_Injection
|
||
_FORMULA_INJECTION_PREFIXES: tuple[str, ...] = ("=", "+", "-", "@", "\t", "\r")
|
||
# Основной продуктовый горизонт (мес) — из него тянем сводный deficit_index сценария
|
||
# (зеркало report_assembler._PRIMARY_HORIZON_MONTHS). ScenarioForecast.as_dict() несёт
|
||
# список forecasts по горизонтам, скалярного «overall» у сценария НЕТ — берём дефицит
|
||
# основного горизонта как сводный показатель. При fallback на иной горизонт ячейка
|
||
# несёт «(гор. N мес)» через _scenario_deficit_cell, чтобы шапка «(12 мес)» не лгала
|
||
# (#1590).
|
||
_PRIMARY_HORIZON_MONTHS: int = 12
|
||
|
||
# RU-метки уровней отчётной уверенности (§15) для листа «Сводка».
|
||
_LEVEL_RU: dict[str, str] = {
|
||
"high": "высокая",
|
||
"medium": "средняя",
|
||
"low": "низкая",
|
||
}
|
||
|
||
# ── Named-константы: стили (header bold + заливка, выравнивание) ───────────────
|
||
|
||
_HEADER_FONT: Font = Font(bold=True, color="FFFFFF")
|
||
_HEADER_FILL: PatternFill = PatternFill("solid", fgColor="1D4ED8") # синий house-accent
|
||
_TITLE_FONT: Font = Font(bold=True, size=14)
|
||
_SECTION_FONT: Font = Font(bold=True, size=11, color="374151")
|
||
_ADVISORY_FONT: Font = Font(italic=True, color="B45309") # янтарный — предупреждение
|
||
_WRAP_TOP: Alignment = Alignment(wrap_text=True, vertical="top")
|
||
_LEFT_TOP: Alignment = Alignment(horizontal="left", vertical="top")
|
||
|
||
# Ширины колонок по умолчанию (символы). Первая колонка-метка шире, далее значения.
|
||
_LABEL_COL_WIDTH: float = 32.0
|
||
_VALUE_COL_WIDTH: float = 28.0
|
||
_WIDE_COL_WIDTH: float = 44.0
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Нормализация входа + микро-хелперы форматирования. Все PURE, graceful.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
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 _sanitize_formula(text: str) -> str:
|
||
"""Нейтрализовать formula/DDE-injection: префикс `'` строкам-«формулам». PURE.
|
||
|
||
openpyxl сохраняет строку с лидирующим `= + - @ \\t \\r` как ФОРМУЛУ (data_type='f'),
|
||
и Excel/LibreOffice исполнит её при открытии. Внешние/скрейпинг-данные (comm_name из
|
||
DOM.РФ, headline/usp_text/advisory и пр.) попадают в ячейки через `_fmt`, поэтому
|
||
экранируем по OWASP-набору: ведём строку апострофом — Excel показывает литеральный
|
||
текст и трактует ячейку как текст, формула не исполняется. Безопасные строки — как есть.
|
||
"""
|
||
if text.startswith(_FORMULA_INJECTION_PREFIXES):
|
||
return "'" + text
|
||
return text
|
||
|
||
|
||
def _fmt(value: Any) -> Any:
|
||
"""Привести значение к ячейке Excel: None → "—", float → округление. PURE.
|
||
|
||
Числа округляем до `_ROUND_DIGITS` (но bool оставляем как есть — это не число для
|
||
округления). Строки экранируем от formula/DDE-injection (`_sanitize_formula`) —
|
||
внешние/скрейпинг-данные с лидирующим `= + - @` не должны стать формулой. Контейнеры
|
||
(dict/list) сворачиваем в строку и тоже экранируем (на случай вложенного значения).
|
||
"""
|
||
if value is None:
|
||
return _DASH
|
||
if isinstance(value, bool):
|
||
return "да" if value else "нет"
|
||
if isinstance(value, float):
|
||
return round(value, _ROUND_DIGITS)
|
||
if isinstance(value, str):
|
||
return _sanitize_formula(value)
|
||
if isinstance(value, int):
|
||
return value
|
||
if isinstance(value, dict | list):
|
||
return _sanitize_formula(str(value))
|
||
return 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
|
||
|
||
|
||
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"),
|
||
}
|
||
|
||
|
||
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 (→ "—").
|
||
"""
|
||
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 (#1590, зеркало report_pdf.py).
|
||
"""
|
||
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 → отдаём как есть (→ "—").
|
||
Зеркалит report_pdf._scenario_deficit_cell (#1590).
|
||
"""
|
||
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
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Низкоуровневые помощники записи листа (заголовки/строки/ширины). PURE-побочка
|
||
# (мутируют переданный лист). Не ходят в БД/сеть.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _write_title(ws: Worksheet, row: int, text: str) -> int:
|
||
"""Заголовок-секции жирным; вернуть номер следующей свободной строки. PURE-side."""
|
||
cell = ws.cell(row=row, column=1, value=text)
|
||
cell.font = _SECTION_FONT
|
||
return row + 1
|
||
|
||
|
||
def _write_kv(ws: Worksheet, row: int, label: str, value: Any) -> int:
|
||
"""Строка «метка → значение» (label жирный, value через `_fmt`). PURE-side.
|
||
|
||
Метка тоже экранируется от formula-injection: часть меток — это КЛЮЧИ из секций
|
||
отчёта (`key_numbers`/`market_metrics`/`factors`…), потенциально внешнего происхождения.
|
||
"""
|
||
label_cell = ws.cell(row=row, column=1, value=_sanitize_formula(label))
|
||
label_cell.font = Font(bold=True)
|
||
label_cell.alignment = _LEFT_TOP
|
||
value_cell = ws.cell(row=row, column=2, value=_fmt(value))
|
||
value_cell.alignment = _WRAP_TOP
|
||
return row + 1
|
||
|
||
|
||
def _write_no_data(ws: Worksheet, row: int) -> int:
|
||
"""Ячейка-заглушка «нет данных» для пустой секции (graceful). PURE-side."""
|
||
cell = ws.cell(row=row, column=1, value=_NO_DATA)
|
||
cell.font = Font(italic=True, color="9CA3AF")
|
||
return row + 1
|
||
|
||
|
||
def _write_table(ws: Worksheet, row: int, headers: list[str], rows: list[list[Any]]) -> int:
|
||
"""Таблица: жирная шапка (заливка) + строки данных через `_fmt`. PURE-side.
|
||
|
||
Пустой `rows` → одна строка-заглушка «нет данных» под шапкой (graceful — таблица
|
||
всё равно валидна). Вернуть номер следующей свободной строки.
|
||
"""
|
||
for col_idx, header in enumerate(headers, start=1):
|
||
cell = ws.cell(row=row, column=col_idx, value=header)
|
||
cell.font = _HEADER_FONT
|
||
cell.fill = _HEADER_FILL
|
||
cell.alignment = _WRAP_TOP
|
||
row += 1
|
||
if not rows:
|
||
ws.cell(row=row, column=1, value=_NO_DATA).font = Font(italic=True, color="9CA3AF")
|
||
return row + 1
|
||
for data_row in rows:
|
||
for col_idx, value in enumerate(data_row, start=1):
|
||
cell = ws.cell(row=row, column=col_idx, value=_fmt(value))
|
||
cell.alignment = _WRAP_TOP
|
||
row += 1
|
||
return row
|
||
|
||
|
||
def _set_widths(ws: Worksheet, widths: list[float]) -> None:
|
||
"""Проставить ширины колонок (символы) по порядку. PURE-side."""
|
||
for idx, width in enumerate(widths, start=1):
|
||
ws.column_dimensions[get_column_letter(idx)].width = width
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Построители листов — по одному на содержательную секцию §13. Каждый graceful:
|
||
# пустая секция → лист с «нет данных», НИКОГДА не падает.
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _build_summary_sheet(ws: Worksheet, report: dict[str, Any]) -> None:
|
||
"""Лист «Сводка»: exec_summary + meta + confidence + ADVISORY-маркер. Graceful."""
|
||
exec_summary = _as_dict(report.get("exec_summary"))
|
||
meta = _as_dict(report.get("meta"))
|
||
confidence = _as_dict(report.get("confidence"))
|
||
|
||
title = ws.cell(row=1, column=1, value="Site Finder v2 — советующий отчёт §13")
|
||
title.font = _TITLE_FONT
|
||
advisory = ws.cell(row=2, column=1, value=_ADVISORY_MARKER)
|
||
advisory.font = _ADVISORY_FONT
|
||
advisory.alignment = _WRAP_TOP
|
||
row = 4
|
||
|
||
# ── Вердикт (exec_summary) ──
|
||
row = _write_title(ws, row, "Вердикт")
|
||
row = _write_kv(ws, row, "Заголовок", exec_summary.get("headline"))
|
||
row = _write_kv(ws, row, "Развёрнутый вердикт", exec_summary.get("verdict"))
|
||
row = _write_kv(ws, row, "Общая уверенность", _level_ru(exec_summary.get("overall_confidence")))
|
||
row += 1
|
||
|
||
# ── Ключевые числа (плоский dict) ──
|
||
row = _write_title(ws, row, "Ключевые числа")
|
||
key_numbers = _as_dict(exec_summary.get("key_numbers"))
|
||
if key_numbers:
|
||
for key, value in key_numbers.items():
|
||
row = _write_kv(ws, row, str(key), value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
row += 1
|
||
|
||
# ── Уверенность (§15) ──
|
||
row = _write_title(ws, row, "Уверенность отчёта")
|
||
row = _write_kv(ws, row, "Уровень", _level_ru(confidence.get("level")))
|
||
row = _write_kv(ws, row, "Обоснование", confidence.get("rationale"))
|
||
row += 1
|
||
|
||
# ── Факторы уверенности (§15 #990 — что тянуло уверенность) ──
|
||
row = _write_title(ws, row, "Факторы уверенности")
|
||
factors = _as_dict(confidence.get("factors"))
|
||
if factors:
|
||
for key, value in factors.items():
|
||
row = _write_kv(ws, row, str(key), value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
row += 1
|
||
|
||
# ── Контекст (meta) ──
|
||
row = _write_title(ws, row, "Контекст")
|
||
row = _write_kv(ws, row, "Кадастровый номер", meta.get("cad_num"))
|
||
row = _write_kv(ws, row, "Район", meta.get("district"))
|
||
horizons = _as_list(meta.get("horizons"))
|
||
row = _write_kv(
|
||
ws, row, "Горизонты (мес)", ", ".join(str(h) for h in horizons) if horizons else None
|
||
)
|
||
row = _write_kv(ws, row, "Сформировано", meta.get("generated_at"))
|
||
row = _write_kv(ws, row, "Версия схемы", meta.get("schema_version"))
|
||
|
||
_set_widths(ws, [_LABEL_COL_WIDTH, _WIDE_COL_WIDTH])
|
||
|
||
|
||
def _build_market_now_sheet(ws: Worksheet, report: dict[str, Any]) -> None:
|
||
"""Лист «Рынок сейчас»: абсорбция/цены (market_metrics) + конкуренты. Graceful."""
|
||
market_now = _as_dict(report.get("market_now"))
|
||
metrics = _as_dict(market_now.get("market_metrics"))
|
||
competitors = _as_list(market_now.get("competitors"))
|
||
supply = _as_dict(market_now.get("supply_layers"))
|
||
|
||
row = _write_title(ws, 1, "Текущий рынок")
|
||
row = _write_kv(ws, row, "Резюме", market_now.get("summary"))
|
||
row += 1
|
||
|
||
# ── Метрики рынка §9.2 (плоский dict — рисуем как метка→значение) ──
|
||
row = _write_title(ws, row, "Метрики рынка")
|
||
if metrics:
|
||
for key, value in metrics.items():
|
||
row = _write_kv(ws, row, str(key), value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
row += 1
|
||
|
||
# ── Слои предложения §9.3 ──
|
||
row = _write_title(ws, row, "Слои предложения")
|
||
if supply:
|
||
for key, value in supply.items():
|
||
row = _write_kv(ws, row, str(key), value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
row += 1
|
||
|
||
# ── Конкуренты §9.1 (таблица) ──
|
||
row = _write_title(ws, row, "Конкуренты")
|
||
headers = ["ЖК", "ID", "Релевантность"]
|
||
table_rows = [
|
||
[c.get("comm_name"), c.get("obj_id"), c.get("relevance_weight")]
|
||
for c in competitors
|
||
if isinstance(c, dict)
|
||
]
|
||
_write_table(ws, row, headers, table_rows)
|
||
|
||
_set_widths(ws, [_LABEL_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH])
|
||
|
||
|
||
def _build_future_market_sheet(ws: Worksheet, report: dict[str, Any]) -> None:
|
||
"""Лист «Будущий рынок»: спрос/предложение/дефицит по горизонтам. Graceful."""
|
||
future = _as_dict(report.get("future_market"))
|
||
forecasts = _as_list(future.get("forecasts_by_horizon"))
|
||
scenarios_summary = _as_dict(future.get("scenarios_summary"))
|
||
|
||
row = _write_title(ws, 1, "Будущий рынок")
|
||
row = _write_kv(ws, row, "Резюме", future.get("summary"))
|
||
row += 1
|
||
|
||
# ── Прогноз по горизонтам (таблица) ──
|
||
row = _write_title(ws, row, "Прогноз по горизонтам")
|
||
headers = ["Горизонт, мес", "Спрос", "Предложение", "Индекс дефицита", "Уверенность"]
|
||
table_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)
|
||
]
|
||
row = _write_table(ws, row, headers, table_rows)
|
||
row += 1
|
||
|
||
# ── Давление будущего предложения §9.3 (FutureSupplyPressure.as_dict()) ──
|
||
row = _write_title(ws, row, "Давление будущего предложения")
|
||
future_supply_pairs = _future_supply_pairs(future.get("future_supply"))
|
||
if future_supply_pairs:
|
||
for label, value in future_supply_pairs.items():
|
||
row = _write_kv(ws, row, label, value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
row += 1
|
||
|
||
# ── Разброс сценариев по дефициту ──
|
||
row = _write_title(ws, row, "Разброс сценариев (индекс дефицита)")
|
||
if scenarios_summary:
|
||
for key, value in scenarios_summary.items():
|
||
row = _write_kv(ws, row, str(key), value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
|
||
_set_widths(
|
||
ws,
|
||
[_LABEL_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH],
|
||
)
|
||
|
||
|
||
def _build_product_tz_sheet(ws: Worksheet, report: dict[str, Any]) -> None:
|
||
"""Лист «Продукт ТЗ»: класс + квартирография (mix) + коммерция + USP. Graceful."""
|
||
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"))
|
||
|
||
row = _write_title(ws, 1, "Рекомендация продукта")
|
||
row = _write_kv(ws, row, "Резюме", product.get("summary"))
|
||
row = _write_kv(ws, row, "Рекомендованный класс", product.get("obj_class"))
|
||
row += 1
|
||
|
||
# ── Квартирография (таблица: формат + доля/класс/дефицит — defensive ключи) ──
|
||
row = _write_title(ws, row, "Квартирография (mix)")
|
||
headers = ["Формат", "Доля, %", "Класс", "Индекс дефицита"]
|
||
table_rows = [
|
||
[m.get("bucket"), m.get("pct"), m.get("obj_class"), m.get("deficit_index")]
|
||
for m in mix
|
||
if isinstance(m, dict)
|
||
]
|
||
row = _write_table(ws, row, headers, table_rows)
|
||
row += 1
|
||
|
||
# ── Коммерция §10.4 ──
|
||
row = _write_title(ws, row, "Коммерция")
|
||
if commercial:
|
||
for key, value in commercial.items():
|
||
row = _write_kv(ws, row, str(key), value)
|
||
else:
|
||
row = _write_no_data(ws, row)
|
||
row += 1
|
||
|
||
# ── USP-ниши §10.5 (таблица: сегмент + текст) ──
|
||
row = _write_title(ws, row, "USP-ниши")
|
||
headers = ["Сегмент", "USP"]
|
||
usp_rows = [[u.get("segment"), u.get("usp_text")] for u in usp if isinstance(u, dict)]
|
||
_write_table(ws, row, headers, usp_rows)
|
||
|
||
_set_widths(ws, [_LABEL_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH])
|
||
|
||
|
||
def _build_scenarios_sheet(ws: Worksheet, report: dict[str, Any]) -> None:
|
||
"""Лист «Сценарии»: conservative/base/aggressive (таблица). Graceful.
|
||
|
||
§22 audit-issue #1871 P1 (collapse): если `scenarios_collapsed=True` (failed
|
||
β-gate в §9.6 → §9.4 coefficient=1.0 → три сценария идентичны), выводим ОДНУ
|
||
строку «Базовый» + плашку-caveat с причиной collapse — НЕ три одинаковых
|
||
столбца (иначе таблица читается как баг). Поля `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")
|
||
|
||
row = _write_title(ws, 1, "Сценарии")
|
||
row = _write_kv(ws, row, "Резюме", scenarios.get("summary"))
|
||
row += 1
|
||
|
||
if collapsed:
|
||
# Плашка-caveat: причина collapse (italic, янтарный — зеркало _ADVISORY_FONT).
|
||
caveat_cell = ws.cell(row=row, column=1, value="Сценарная дифференциация недоступна")
|
||
caveat_cell.font = Font(bold=True, color="B45309")
|
||
row += 1
|
||
if isinstance(collapse_reason, str) and collapse_reason:
|
||
reason_cell = ws.cell(row=row, column=1, value=collapse_reason)
|
||
reason_cell.font = _ADVISORY_FONT
|
||
reason_cell.alignment = _WRAP_TOP
|
||
row += 1
|
||
row += 1
|
||
|
||
# ── ОДНА строка «Базовый» вместо трёх идентичных (collapse) ──
|
||
row = _write_title(ws, row, "Базовый сценарий")
|
||
headers = ["Сценарий", f"Индекс дефицита ({_PRIMARY_HORIZON_MONTHS} мес)", "Advisory"]
|
||
base_payload = _as_dict(by_scenario.get("base"))
|
||
table_rows: list[list[Any]] = (
|
||
[["base", _scenario_deficit_cell(base_payload), base_payload.get("advisory")]]
|
||
if base_payload
|
||
else []
|
||
)
|
||
_write_table(ws, row, headers, table_rows)
|
||
|
||
_set_widths(ws, [_LABEL_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH])
|
||
return
|
||
|
||
# ── Штатная таблица сценариев (имя + сводный дефицит 12 мес + advisory) ──
|
||
# _scenario_deficit_cell: при fallback на нестандартный горизонт ячейка несёт
|
||
# «(гор. N мес)» — шапка «(12 мес)» не лжёт (#1590).
|
||
row = _write_title(ws, row, "Сводка по сценариям")
|
||
headers = ["Сценарий", f"Индекс дефицита ({_PRIMARY_HORIZON_MONTHS} мес)", "Advisory"]
|
||
table_rows = []
|
||
for name, payload in by_scenario.items():
|
||
data = _as_dict(payload)
|
||
table_rows.append([name, _scenario_deficit_cell(data), data.get("advisory")])
|
||
_write_table(ws, row, headers, table_rows)
|
||
|
||
_set_widths(ws, [_LABEL_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH])
|
||
|
||
|
||
def _build_scoring_sheet(ws: Worksheet, report: dict[str, Any]) -> None:
|
||
"""Лист «Скоринг»: продуктовые скоры (#985) + спец-индексы (#986) + overall. Graceful."""
|
||
scoring = _as_dict(report.get("scoring"))
|
||
product_scores = _as_dict(scoring.get("product_scores"))
|
||
special_indices = _as_dict(scoring.get("special_indices"))
|
||
|
||
row = _write_title(ws, 1, "Скоринг")
|
||
row = _write_kv(ws, row, "Итоговый скор (overall)", scoring.get("overall"))
|
||
row += 1
|
||
|
||
# ── Продуктовые скоры (#985): scores = {name: {value: ...}} → таблица ──
|
||
row = _write_title(ws, row, "Продуктовые скоры")
|
||
scores = _as_dict(product_scores.get("scores"))
|
||
score_rows = [[name, _as_dict(payload).get("value")] for name, payload in scores.items()]
|
||
row = _write_table(ws, row, ["Скор", "Значение"], score_rows)
|
||
row += 1
|
||
|
||
# ── Спец-индексы (#986): indices = {name: {value, label}} → таблица ──
|
||
row = _write_title(ws, row, "Специальные индексы")
|
||
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()
|
||
]
|
||
_write_table(ws, row, ["Индекс", "Значение", "Метка"], index_rows)
|
||
|
||
_set_widths(ws, [_LABEL_COL_WIDTH, _VALUE_COL_WIDTH, _VALUE_COL_WIDTH])
|
||
|
||
|
||
# Реестр: (заголовок листа → построитель). Порядок = порядок листов в книге.
|
||
_SHEET_BUILDERS: tuple[tuple[str, Any], ...] = (
|
||
(_SHEET_SUMMARY, _build_summary_sheet),
|
||
(_SHEET_MARKET_NOW, _build_market_now_sheet),
|
||
(_SHEET_FUTURE_MARKET, _build_future_market_sheet),
|
||
(_SHEET_PRODUCT_TZ, _build_product_tz_sheet),
|
||
(_SHEET_SCENARIOS, _build_scenarios_sheet),
|
||
(_SHEET_SCORING, _build_scoring_sheet),
|
||
)
|
||
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
# Публичный API — рендер отчёта в .xlsx-байты (без файлового I/O на диск).
|
||
# ──────────────────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def export_report_xlsx(report: Any) -> bytes:
|
||
"""§13 Отрендерить `SiteFinderReport` (#987) в .xlsx-книгу и вернуть БАЙТЫ.
|
||
|
||
По одному листу на содержательную секцию §13 (Сводка / Рынок сейчас / Будущий
|
||
рынок / Продукт ТЗ / Сценарии / Скоринг). Шапки таблиц жирные с заливкой, RU-метки,
|
||
числа округлены, None → "—". На листе «Сводка» — ADVISORY-маркер (отчёт советующий).
|
||
|
||
ДЕТЕРМИНИРОВАННО, БЕЗ LLM/БД/сети. Принимает КАК `SiteFinderReport`-инстанс, ТАК и
|
||
его `as_dict()`-словарь (нормализуется через `_normalize`). GRACEFUL: частичный/
|
||
пустой отчёт → листы с «нет данных», НИКОГДА не падает.
|
||
|
||
Args:
|
||
report: `SiteFinderReport`-инстанс или его `as_dict()`-словарь.
|
||
|
||
Returns:
|
||
Непустые .xlsx-байты (можно перечитать `openpyxl.load_workbook(BytesIO(...))`),
|
||
готовые для `Response(media_type="application/vnd.openxmlformats-...")`.
|
||
"""
|
||
data = _normalize(report)
|
||
|
||
wb = Workbook()
|
||
# Workbook создаёт дефолтный лист — переиспользуем его под первый (Сводка).
|
||
default_ws = wb.active
|
||
for idx, (title, builder) in enumerate(_SHEET_BUILDERS):
|
||
ws = default_ws if idx == 0 else wb.create_sheet(title=title)
|
||
ws.title = title
|
||
builder(ws, data)
|
||
|
||
buffer = BytesIO()
|
||
wb.save(buffer)
|
||
payload = buffer.getvalue()
|
||
|
||
meta = _as_dict(data.get("meta"))
|
||
logger.info(
|
||
"export_report_xlsx: cad_num=%s sheets=%d size=%d bytes advisory=%s",
|
||
meta.get("cad_num"),
|
||
len(_SHEET_BUILDERS),
|
||
len(payload),
|
||
data.get("advisory"),
|
||
)
|
||
return payload
|