feat(export): Markdown + JSON download of §22 forecast report (#959)
All checks were successful
Deploy / changes (push) Successful in 7s
Deploy / build-backend (push) Successful in 1m41s
Deploy / build-frontend (push) Has been skipped
Deploy / build-worker (push) Successful in 2m34s
Deploy / deploy (push) Successful in 1m12s
CI / changes (push) Successful in 8s
CI / backend-tests (push) Has been skipped

render_report_markdown (pure, no new runtime deps) reuses report_pdf's str
helpers (DRY), + GET /{cad_num}/forecast/export?format=md|json. No forecast run
→ 404; graceful on thin/empty reports; GFM-safe table escaping. PDF/XLSX already
existed; this adds the cheapest no-dep formats. Part of EPIC #959.
This commit is contained in:
Light1YT 2026-06-06 23:39:17 +05:00 committed by bot-backend
parent 4e41f9be73
commit 25f7052084
4 changed files with 783 additions and 1 deletions

View file

@ -3,7 +3,7 @@ import json
import logging import logging
import math import math
import time import time
from typing import Annotated, Any from typing import Annotated, Any, Literal
import httpx import httpx
from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Response from fastapi import APIRouter, Body, Depends, Header, HTTPException, Query, Response
@ -44,6 +44,7 @@ from app.services.analysis_runs.repository import (
persist_analysis_run, persist_analysis_run,
) )
from app.services.exporters.layout_tz_pdf import render_layout_tz_pdf from app.services.exporters.layout_tz_pdf import render_layout_tz_pdf
from app.services.exporters.report_md import render_report_markdown
from app.services.exporters.snapshot_pdf import generate_snapshot_pdf from app.services.exporters.snapshot_pdf import generate_snapshot_pdf
from app.services.site_finder.best_layouts import get_best_layouts from app.services.site_finder.best_layouts import get_best_layouts
from app.services.site_finder.cadastre_fetch import ( from app.services.site_finder.cadastre_fetch import (
@ -1313,6 +1314,59 @@ def get_parcel_forecast(
return {"status": "pending"} return {"status": "pending"}
@router.get("/{cad_num}/forecast/export")
def export_parcel_forecast(
cad_num: str,
db: Annotated[Session, Depends(get_db)],
format: Annotated[
Literal["md", "json"],
Query(description="Формат выгрузки: md (Markdown) | json (сырой отчёт)"),
] = "md",
) -> Response:
"""Скачать §22-форсайт участка файлом — Markdown (default) или JSON (#959, EPIC export).
Самый дешёвый, БЕЗ-runtime-dep срез экспорта: читает ПОСЛЕДНИЙ §22-ран schema_version
"1.0" (тот же блоб, что отдаёт GET /{cad}/forecast inline) и отдаёт его как attachment.
format=md `render_report_markdown(run.result)` (text/markdown, чистый Markdown).
format=json сырой `run.result` (application/json) download-вариант inline-чтения.
В отличие от read-only поллинга GET /{cad}/forecast (нет рана 202 pending), это
download-эндпоинт: нет рана 404 (graceful HTTPException «прогноз ещё не посчитан»,
НЕ 500). 3-сегментный путь (`/{cad}/forecast/export`) не конфликтует с 2-сегментными
`/{cad}/forecast`, `/{cad}/runs`, `/runs/{run_id}` и 1-сегментным `/{parcel_id}`.
Args:
cad_num: кадастровый номер участка (в имени файла `:` `_`).
format: "md" (default) | "json".
Returns:
Response-attachment с Content-Disposition (имя
`gendesign_forecast_<cad>_<YYYY-MM-DD>.<ext>`).
"""
run = latest_run_for(db, cad_num, schema_version=_FORECAST_SCHEMA_VERSION)
if run is None:
raise HTTPException(status_code=404, detail="прогноз ещё не посчитан")
cad_safe = cad_num.replace(":", "_")
today = _dt.date.today().strftime("%Y-%m-%d")
base_name = f"gendesign_forecast_{cad_safe}_{today}"
if format == "json":
payload = json.dumps(run.result, ensure_ascii=False, default=str)
media_type = "application/json"
filename = f"{base_name}.json"
else:
payload = render_report_markdown(run.result)
media_type = "text/markdown; charset=utf-8"
filename = f"{base_name}.md"
return Response(
content=payload,
media_type=media_type,
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.post("/{cad_num}/analyze", response_model=AnalyzeResponse) @router.post("/{cad_num}/analyze", response_model=AnalyzeResponse)
def analyze_parcel( def analyze_parcel(
cad_num: str, cad_num: str,

View file

@ -0,0 +1,403 @@
"""§13/§22 Markdown-экспортёр итогового советующего отчёта Site Finder v2.
#959 (EPIC export). Самый дешёвый, БЕЗ-runtime-dep двойник PDF/Excel-экспортёров
(`report_pdf` / `excel`): рендерит `SiteFinderReport` (#987) в ЧИТАЕМЫЙ Markdown-текст
и возвращает СТРОКУ готовую для `Response(media_type="text/markdown")` / записи.
Чистый str-билдинг (НИКАКИХ новых зависимостей: ни python-docx, ни pptx, ни шаблонизатора).
DRY: PURE-хелперы нормализации/форматирования НЕ дублируем импортируем из
`report_pdf` (там они уже есть: `_normalize` приём инстанса ИЛИ `as_dict()`-словаря,
`_fmt` строковое форматирование ячейки/None"", `_level_ru`, `_as_dict`/`_as_list`,
`_future_supply_pairs`, `_scenario_deficit_index`) + переиспользуем его named-константы
(`_DASH`/`_NO_DATA`/`_ADVISORY_MARKER`). `report_pdf._fmt` возвращает `str` (в отличие
от `excel._fmt`, отдающего `Any` под ячейку openpyxl) для Markdown нужен именно str,
поэтому источник хелперов `report_pdf`. Здесь добавлены ТОЛЬКО Markdown-специфичные
микро-билдеры (kv-строки / таблицы / экранирование pipe).
ДЕТЕРМИНИРОВАННЫЙ, БЕЗ LLM, БЕЗ БД/сети: только ПОТРЕБЛЯЕТ уже-собранный отчёт (его
наполняет сборщик #988). Принимает КАК `SiteFinderReport`-инстанс, ТАК и его
`as_dict()`-словарь (нормализуется через `_normalize`).
GRACEFUL (дух всего форсайт-стека): частичный/пустой/мусорный отчёт ВАЛИДЕН пустая
секция рисует «нет данных», экспортёр НИКОГДА не падает (нет KeyError на тонком отчёте).
Отчёт СОВЕТУЮЩИЙ: после H1 заметный advisory-дисклеймер (оценка не основание для
инвест-решения). Содержание и ПОРЯДОК секций ЗЕРКАЛЯТ `report_pdf`.
"""
from __future__ import annotations
import logging
from typing import 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_index,
)
logger = logging.getLogger(__name__)
# ── Named-константы: заголовки (зеркало report_pdf — те же восемь секций §13) ──
_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) — после H1, виден сразу.
_ADVISORY_DISCLAIMER: str = "Оценка advisory — не основание для инвест-решения."
# Сколько конкурентов выводить в таблицу (top-N, чтобы не раздувать MD). Зеркало
# competitors_limit=5 у snapshot-экспорта.
_COMPETITORS_TOP_N: int = 5
# Основной продуктовый горизонт (мес) для подписи сводного дефицита сценария
# (зеркало report_pdf._PRIMARY_HORIZON_MONTHS — там private, держим локальный дубль
# ТОЛЬКО как подпись столбца, само значение тянет импортируемый _scenario_deficit_index).
_PRIMARY_HORIZON_MONTHS: int = 12
# ──────────────────────────────────────────────────────────────────────────────
# Markdown-специфичные микро-билдеры. Все PURE: строят строку из уже-форматированных
# через `_fmt` кусочков. (Хелперы нормализации/форматирования — импортированы из
# report_pdf, не дублируются — DRY.)
# ──────────────────────────────────────────────────────────────────────────────
def _md_cell(value: Any) -> str:
"""Значение → текст ячейки Markdown-таблицы: `_fmt` + экранирование `|`. PURE.
Pipe внутри значения сломал бы разметку таблицы заменяем на escaped `\\|`.
Переводы строк сворачиваем в пробел (ячейка таблицы однострочна).
"""
text = _fmt(value)
return text.replace("|", "\\|").replace("\n", " ").replace("\r", " ")
def _md_table(headers: list[str], rows: list[list[Any]]) -> str:
"""GFM-таблица: шапка + разделитель + строки (через `_md_cell`). PURE.
Пустой `rows` строка-заглушка «нет данных» (graceful таблица валидна).
Возвращает блок без хвостовых пустых строк (склейку делает вызывающий).
"""
head = "| " + " | ".join(headers) + " |"
sep = "| " + " | ".join("---" for _ in headers) + " |"
if not rows:
empty = "| " + " | ".join([_NO_DATA] + [""] * (len(headers) - 1)) + " |"
return "\n".join([head, sep, empty])
body = [
"| " + " | ".join(_md_cell(cell) for cell in row) + " |"
for row in rows
]
return "\n".join([head, sep, *body])
def _md_kv_table(data: dict[str, Any]) -> str:
"""Таблица «Показатель → Значение» из плоского dict. Пустой → «нет данных». PURE."""
rows = [[str(key), value] for key, value in data.items()]
return _md_table(["Показатель", "Значение"], rows)
def _md_kv_lines(pairs: list[tuple[str, Any]]) -> str:
"""Список «**метка:** значение» построчно (для коротких карточек meta). PURE."""
if not pairs:
return _NO_DATA
return "\n".join(f"- **{label}:** {_fmt(value)}" for label, value in pairs)
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
# ──────────────────────────────────────────────────────────────────────────────
# Построители секций — по одной на содержательную секцию §13, тот же порядок/набор,
# что и в report_pdf. Каждый graceful (пустая секция → «нет данных», не падает) и
# возвращает Markdown-блок (строку) без ведущего/хвостового перевода строки.
# ──────────────────────────────────────────────────────────────────────────────
def _build_header(report: dict[str, Any]) -> str:
"""H1 + advisory-дисклеймер + карточка meta (cad/район/горизонты/дата/схема). PURE."""
meta = _as_dict(report.get("meta"))
schema_version = report.get("schema_version") or meta.get("schema_version")
meta_pairs: list[tuple[str, Any]] = [
("Кадастровый номер", meta.get("cad_num")),
("Район", meta.get("district")),
("Горизонты (мес)", _join_horizons(_as_list(meta.get("horizons")))),
("Сформировано", meta.get("generated_at")),
("Версия схемы", schema_version),
]
return (
f"# {_TITLE_DOC}\n\n"
f"> **{_ADVISORY_DISCLAIMER}** {_ADVISORY_MARKER}\n\n"
f"{_md_kv_lines(meta_pairs)}"
)
def _build_summary(report: dict[str, Any]) -> str:
"""§13.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"))
return (
f"## {_TITLE_SUMMARY}\n\n"
f"**{_fmt(headline)}**\n\n"
f"{_fmt(verdict)}\n\n"
f"_Общая уверенность: {overall_conf}_\n\n"
f"### Ключевые числа\n\n"
f"{_md_kv_table(key_numbers)}"
)
def _build_market_now(report: dict[str, Any]) -> str:
"""§13.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)
]
return (
f"## {_TITLE_MARKET_NOW}\n\n"
f"{_fmt(market_now.get('summary'))}\n\n"
f"### Метрики рынка\n\n"
f"{_md_kv_table(metrics)}\n\n"
f"### Слои предложения\n\n"
f"{_md_kv_table(supply)}\n\n"
f"### Конкуренты\n\n"
f"{_md_table(['ЖК', 'ID', 'Релевантность'], comp_rows)}"
)
def _build_future_market(report: dict[str, Any]) -> str:
"""§13.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)
]
return (
f"## {_TITLE_FUTURE_MARKET}\n\n"
f"{_fmt(future.get('summary'))}\n\n"
f"### Прогноз по горизонтам\n\n"
f"{_md_table(forecast_headers, forecast_rows)}\n\n"
f"### Давление будущего предложения\n\n"
f"{_md_kv_table(_future_supply_pairs(future.get('future_supply')))}\n\n"
f"### Разброс сценариев (индекс дефицита)\n\n"
f"{_md_kv_table(scenarios_summary)}"
)
def _build_scenarios(report: dict[str, Any]) -> str:
"""§13.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_index(data), rate_str, data.get("advisory")]
)
headers = [
"Сценарий",
f"Индекс дефицита ({_PRIMARY_HORIZON_MONTHS} мес)",
"Траектория ставки",
"Advisory",
]
return (
f"## {_TITLE_SCENARIOS}\n\n"
f"{_fmt(scenarios.get('summary'))}\n\n"
f"{_md_table(headers, rows)}"
)
def _build_product_tz(report: dict[str, Any]) -> str:
"""§13.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)
]
return (
f"## {_TITLE_PRODUCT_TZ}\n\n"
f"{_md_kv_lines(head_pairs)}\n\n"
f"### Квартирография (mix)\n\n"
f"{_md_table(['Формат', 'Доля, %', 'Класс', 'Индекс дефицита'], mix_rows)}\n\n"
f"### Коммерция\n\n"
f"{_md_kv_table(commercial)}\n\n"
f"### USP-ниши\n\n"
f"{_md_table(['Сегмент', 'USP'], usp_rows)}\n\n"
f"### Причины (§16)\n\n"
f"{_md_table(['Почему', 'Advisory'], reason_rows)}"
)
def _build_scoring(report: dict[str, Any]) -> str:
"""§13.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()
]
return (
f"## {_TITLE_SCORING}\n\n"
f"_Итоговый скор (overall): {_fmt(scoring.get('overall'))}_\n\n"
f"### Продуктовые скоры\n\n"
f"{_md_table(['Скор', 'Значение'], score_rows)}\n\n"
f"### Специальные индексы\n\n"
f"{_md_table(['Индекс', 'Значение', 'Метка'], index_rows)}"
)
def _build_confidence(report: dict[str, Any]) -> str:
"""§13.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])
return (
f"## {_TITLE_CONFIDENCE}\n\n"
f"_Уровень: {level}_\n\n"
f"{_fmt(rationale)}\n\n"
f"### Факторы уверенности\n\n"
f"{_md_table(['Фактор', 'Значение', 'Уровень', 'Комментарий'], factor_rows)}"
)
# Реестр построителей. Порядок = порядок блоков в документе (зеркало report_pdf:
# Сводка → Рынок сейчас → Будущий рынок → Сценарии → Продукт ТЗ → Скоринг →
# Уверенность; header идёт отдельно первым).
_SECTION_BUILDERS: tuple[Any, ...] = (
_build_summary,
_build_market_now,
_build_future_market,
_build_scenarios,
_build_product_tz,
_build_scoring,
_build_confidence,
)
# ──────────────────────────────────────────────────────────────────────────────
# Публичный API — рендер отчёта в Markdown-строку (без файлового I/O, без зависимостей).
# ──────────────────────────────────────────────────────────────────────────────
def render_report_markdown(report: Any) -> str:
"""§13/§22 Отрендерить `SiteFinderReport` (#987) в читаемый Markdown и вернуть str.
H1 + advisory-дисклеймер + карточка meta, далее восемь содержательных секций §13
(Сводка / Рынок сейчас / Будущий рынок / Сценарии / Продукт ТЗ / Скоринг /
Уверенность) содержание и порядок ЗЕРКАЛЯТ `report_pdf`. Числа форматируются
через импортированный `_fmt` (None ""), уровни уверенности RU-метками.
Чистый Markdown (GFM-таблицы), без HTML.
ДЕТЕРМИНИРОВАННО, БЕЗ LLM/БД/сети, БЕЗ новых зависимостей. Принимает КАК
`SiteFinderReport`-инстанс, ТАК и его `as_dict()`-словарь (нормализуется через
импортированный `_normalize`). GRACEFUL: частичный/пустой/мусорный отчёт секции
с «нет данных», НИКОГДА не падает (нет KeyError на тонком отчёте).
Args:
report: `SiteFinderReport`-инстанс или его `as_dict()`-словарь (или мусор "").
Returns:
Непустую Markdown-строку, готовую для
`Response(media_type="text/markdown; charset=utf-8")`.
"""
data = _normalize(report)
blocks: list[str] = [_build_header(data)]
blocks.extend(builder(data) for builder in _SECTION_BUILDERS)
markdown = "\n\n".join(blocks) + "\n"
meta = _as_dict(data.get("meta"))
logger.info(
"render_report_markdown: cad_num=%s sections=%d len=%d advisory=%s",
meta.get("cad_num"),
len(_SECTION_BUILDERS),
len(markdown),
data.get("advisory"),
)
return markdown

View file

@ -327,3 +327,133 @@ def test_get_forecast_db_error_returns_pending_not_500() -> None:
assert resp.json() == {"status": "pending"} assert resp.json() == {"status": "pending"}
finally: finally:
app.dependency_overrides.clear() app.dependency_overrides.clear()
# ── GET /{cad}/forecast/export (#959, EPIC export — Markdown + JSON download) ────
_REPORT_FIXTURE = {
"schema_version": "1.0",
"advisory": True,
"exec_summary": {
"headline": "Строить комфорт-класс.",
"verdict": "Дефицит малых форматов на горизонте 12 мес.",
"key_numbers": {"deficit_index": -1.0, "overall_score": 0.42},
"overall_confidence": "low",
},
"future_market": {
"forecasts_by_horizon": [
{
"horizon_months": 12,
"deficit_index": -1.0,
"projected_demand_units": 2330.8,
"projected_supply_units": 42953.2,
"confidence": "low",
}
],
},
"meta": {"cad_num": _CAD, "district": "Кировский", "horizons": [6, 12, 18]},
}
def test_export_forecast_md_returns_markdown() -> None:
"""format=md → 200 text/markdown + Content-Disposition .md, тело — Markdown отчёт."""
from app.core.db import get_db
fake_run = MagicMock()
fake_run.result = _REPORT_FIXTURE
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=fake_run) as lrf:
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast/export?format=md")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"].startswith("text/markdown")
cd = resp.headers["content-disposition"]
assert "attachment;" in cd
assert cd.endswith('.md"')
# cad-двоеточия санитизированы в имени файла.
assert f"gendesign_forecast_{_CAD.replace(':', '_')}_" in cd
body = resp.text
assert body.lstrip().startswith("# ")
assert "Оценка advisory" in body
# Читаем именно §22-схему "1.0".
_, kwargs = lrf.call_args
assert kwargs.get("schema_version") == "1.0"
finally:
app.dependency_overrides.clear()
def test_export_forecast_md_is_default_format() -> None:
"""Без format → default md (text/markdown)."""
from app.core.db import get_db
fake_run = MagicMock()
fake_run.result = _REPORT_FIXTURE
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=fake_run):
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast/export")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"].startswith("text/markdown")
finally:
app.dependency_overrides.clear()
def test_export_forecast_json_returns_dict() -> None:
"""format=json → 200 application/json + attachment .json, тело = run.result dict."""
from app.core.db import get_db
fake_run = MagicMock()
fake_run.result = _REPORT_FIXTURE
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=fake_run):
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast/export?format=json")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"] == "application/json"
cd = resp.headers["content-disposition"]
assert "attachment;" in cd
assert cd.endswith('.json"')
assert resp.json() == _REPORT_FIXTURE
finally:
app.dependency_overrides.clear()
def test_export_forecast_no_run_returns_404() -> None:
"""Рана нет (latest_run_for → None) → 404 (download-вариант, НЕ 202/500)."""
from app.core.db import get_db
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=None):
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast/export")
assert resp.status_code == 404, resp.text
assert "посчитан" in resp.json()["detail"]
finally:
app.dependency_overrides.clear()
def test_export_forecast_invalid_format_returns_422() -> None:
"""format вне {md,json} → 422 (Literal-валидация ДО работы с БД)."""
from app.core.db import get_db
db = MagicMock()
app.dependency_overrides[get_db] = _override_db(db)
try:
with patch("app.api.v1.parcels.latest_run_for", return_value=None) as lrf:
client = TestClient(app)
resp = client.get(f"/api/v1/parcels/{_CAD}/forecast/export?format=pdf")
assert resp.status_code == 422, resp.text
lrf.assert_not_called()
finally:
app.dependency_overrides.clear()

View file

@ -0,0 +1,195 @@
"""Unit-тесты §13/§22 Markdown-экспортёра (#959, EPIC export) — `render_report_markdown`.
Чистые тесты БЕЗ БД/сети (экспортёр только ПОТРЕБЛЯЕТ уже-собранный отчёт):
реальный fixture-отчёт (frontend mock `parcel-forecast.json` `report`) рендерится
в Markdown с заголовками всех секций, advisory-дисклеймером и значением дефицита;
экспортёр принимает КАК `SiteFinderReport`-инстанс, ТАК и его `as_dict()`-словарь;
пустой / частичный / мусорный отчёт валидный Markdown БЕЗ падения (graceful);
pipe в данных экранируется (не ломает GFM-таблицу).
Детерминированно, без LLM. DATABASE_URL выставляем до импорта app-модулей (зеркало
test_report_pdf.py) на случай side-effect'ов импорта пакета.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services.exporters.report_md import (
_ADVISORY_DISCLAIMER,
_TITLE_CONFIDENCE,
_TITLE_FUTURE_MARKET,
_TITLE_MARKET_NOW,
_TITLE_PRODUCT_TZ,
_TITLE_SCENARIOS,
_TITLE_SCORING,
_TITLE_SUMMARY,
render_report_markdown,
)
from app.services.forecasting.report import (
ReportExecSummary,
ReportMeta,
SiteFinderReport,
)
# Реальный fixture: frontend mock §22-форсайта — берём вложенный `report` (форма #987
# `as_dict()`). Так тест не разойдётся с реальной формой отчёта.
_FIXTURE_PATH = (
Path(__file__).resolve().parents[4]
/ "frontend"
/ "src"
/ "lib"
/ "mocks"
/ "parcel-forecast.json"
)
def _fixture_report() -> dict[str, Any]:
"""`report`-словарь из frontend-мока §22 (форма SiteFinderReport.as_dict())."""
data = json.loads(_FIXTURE_PATH.read_text(encoding="utf-8"))
report = data["report"]
assert isinstance(report, dict)
return report
_EXPECTED_TITLES: tuple[str, ...] = (
_TITLE_SUMMARY,
_TITLE_MARKET_NOW,
_TITLE_FUTURE_MARKET,
_TITLE_SCENARIOS,
_TITLE_PRODUCT_TZ,
_TITLE_SCORING,
_TITLE_CONFIDENCE,
)
# ── Реальный fixture-отчёт: структура + ключевые значения ───────────────────────
class TestFixtureReportRender:
def test_returns_non_empty_str(self) -> None:
md = render_report_markdown(_fixture_report())
assert isinstance(md, str)
assert len(md) > 200
def test_h1_and_advisory_disclaimer_present(self) -> None:
md = render_report_markdown(_fixture_report())
# H1 заголовок документа.
assert md.lstrip().startswith("# ")
# Advisory-дисклеймер обязателен (отчёт советующий).
assert _ADVISORY_DISCLAIMER in md
def test_all_section_headers_present(self) -> None:
md = render_report_markdown(_fixture_report())
for title in _EXPECTED_TITLES:
assert f"## {title}" in md, f"отсутствует заголовок секции «{title}»"
def test_meta_cad_num_present(self) -> None:
md = render_report_markdown(_fixture_report())
# Кадастр из meta фикстуры попадает в карточку.
assert "66:41:0702048:27" in md
def test_deficit_value_present(self) -> None:
# Сводный индекс дефицита (1.0) из exec_summary/forecasts должен быть в MD.
report = _fixture_report()
deficit = report["exec_summary"]["key_numbers"]["deficit_index"]
assert deficit is not None
# _fmt округляет float до 3 знаков → "-1.0".
assert str(round(float(deficit), 3)) in md_value(report)
def test_headline_present(self) -> None:
report = _fixture_report()
headline = report["exec_summary"]["headline"]
md = render_report_markdown(report)
# Заголовок-вердикт присутствует (bold).
assert headline.split(".")[0] in md
def test_gfm_table_separator_present(self) -> None:
# Хотя бы одна GFM-таблица (разделитель шапки) есть в выводе.
md = render_report_markdown(_fixture_report())
assert "| --- |" in md
def md_value(report: dict[str, Any]) -> str:
"""Helper: отрендерить отчёт (сахар для inline-ассертов)."""
return render_report_markdown(report)
# ── Вход: dataclass-инстанс ИЛИ as_dict()-словарь ───────────────────────────────
class TestInputForms:
def test_accepts_dataclass_instance(self) -> None:
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Строить комфорт-класс."),
meta=ReportMeta(cad_num="66:41:0000000:1", horizons=[12]),
)
md = render_report_markdown(report)
assert "Строить комфорт" in md
assert "66:41:0000000:1" in md
def test_dataclass_and_dict_equivalent(self) -> None:
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Один заголовок."),
meta=ReportMeta(cad_num="66:41:0000000:9"),
)
assert render_report_markdown(report) == render_report_markdown(report.as_dict())
# ── Graceful: пустой / частичный / мусорный вход → валидный MD без падения ───────
class TestGraceful:
def test_empty_dict_does_not_crash(self) -> None:
# Пустой отчёт {} → валидный MD (no KeyError), с дисклеймером и «нет данных».
md = render_report_markdown({})
assert isinstance(md, str)
assert _ADVISORY_DISCLAIMER in md
assert "нет данных" in md
# Все секции всё равно отрисованы (стабильный контракт).
for title in _EXPECTED_TITLES:
assert f"## {title}" in md
def test_default_report_does_not_crash(self) -> None:
# Полностью дефолтный (пустой) SiteFinderReport — валидный MD.
md = render_report_markdown(SiteFinderReport())
assert _ADVISORY_DISCLAIMER in md
assert "нет данных" in md
def test_garbage_input_returns_valid_md(self) -> None:
# Мусор (None / int / строка) → нормализуется в {} → валидный MD, не падает.
for junk in (None, 123, "not a report", []):
md = render_report_markdown(junk)
assert isinstance(md, str)
assert _ADVISORY_DISCLAIMER in md
def test_partial_report_some_sections(self) -> None:
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Тонкий анализ — только заголовок."),
meta=ReportMeta(cad_num="66:41:0000000:2", horizons=[12]),
)
md = render_report_markdown(report)
assert "Тонкий анализ" in md
assert "66:41:0000000:2" in md
# Незаполненные секции рисуют «нет данных», но MD валиден.
assert "нет данных" in md
# ── Безопасность таблиц: pipe в данных не ломает GFM-разметку ───────────────────
class TestPipeEscaping:
def test_pipe_in_value_escaped(self) -> None:
report = SiteFinderReport(
exec_summary=ReportExecSummary(
key_numbers={"weird|key": "value|with|pipes"},
),
)
md = render_report_markdown(report)
# Сырой неэкранированный pipe из значения не должен утечь — он экранирован.
assert "value\\|with\\|pipes" in md