"""§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» у сценария НЕТ — берём дефицит # основного горизонта как сводный показатель сценария. _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 # ────────────────────────────────────────────────────────────────────────────── # Низкоуровневые помощники записи листа (заголовки/строки/ширины). 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.""" scenarios = _as_dict(report.get("scenarios")) by_scenario = _as_dict(scenarios.get("by_scenario")) row = _write_title(ws, 1, "Сценарии") row = _write_kv(ws, row, "Резюме", scenarios.get("summary")) row += 1 # ── Таблица сценариев (имя + сводный дефицит 12 мес + advisory) ── row = _write_title(ws, row, "Сводка по сценариям") headers = ["Сценарий", "Индекс дефицита (12 мес)", "Advisory"] table_rows: list[list[Any]] = [] for name, payload in by_scenario.items(): data = _as_dict(payload) table_rows.append([name, _scenario_deficit_index(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