feat(export): Telegram summary format (tg) for §22 forecast — slice #959 #1093

Merged
bot-backend merged 1 commit from feat/forecast-export-telegram-959 into main 2026-06-06 18:55:48 +00:00
4 changed files with 398 additions and 11 deletions

View file

@ -44,7 +44,10 @@ from app.services.analysis_runs.repository import (
persist_analysis_run,
)
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.report_md import (
render_report_markdown,
render_report_telegram_summary,
)
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.cadastre_fetch import (
@ -1319,34 +1322,44 @@ def export_parcel_forecast(
cad_num: str,
db: Annotated[Session, Depends(get_db)],
format: Annotated[
Literal["md", "json"],
Query(description="Формат выгрузки: md (Markdown) | json (сырой отчёт)"),
Literal["md", "json", "tg"],
Query(description="Формат выгрузки: md (Markdown) | json (сырой отчёт) | tg (сводка)"),
] = "md",
) -> Response:
"""Скачать §22-форсайт участка файлом — Markdown (default) или JSON (#959, EPIC export).
"""Экспорт §22-форсайта участка — Markdown / JSON (файл) или TG-сводка (#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-чтения.
"1.0" (тот же блоб, что отдаёт GET /{cad}/forecast inline) и отдаёт его в нужной форме.
format=md `render_report_markdown(run.result)` attachment (.md, чистый Markdown).
format=json сырой `run.result` attachment (.json, download-вариант inline-чтения).
format=tg `render_report_telegram_summary(run.result)` КРАТКАЯ plain-text сводка
для копипаста в Telegram, INLINE (без Content-Disposition: это сниппет читать/копировать
и отправить, а не файл-download).
В отличие от read-only поллинга GET /{cad}/forecast (нет рана 202 pending), это
download-эндпоинт: нет рана 404 (graceful HTTPException «прогноз ещё не посчитан»,
export-эндпоинт: нет рана 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".
format: "md" (default) | "json" | "tg".
Returns:
Response-attachment с Content-Disposition (имя
`gendesign_forecast_<cad>_<YYYY-MM-DD>.<ext>`).
Response для md/json attachment с Content-Disposition (имя
`gendesign_forecast_<cad>_<YYYY-MM-DD>.<ext>`); для tg inline text/plain сводка.
"""
run = latest_run_for(db, cad_num, schema_version=_FORECAST_SCHEMA_VERSION)
if run is None:
raise HTTPException(status_code=404, detail="прогноз ещё не посчитан")
# tg — INLINE сниппет (не файл): краткая сводка для копипаста в Telegram, без attachment.
if format == "tg":
return Response(
content=render_report_telegram_summary(run.result),
media_type="text/plain; charset=utf-8",
)
cad_safe = cad_num.replace(":", "_")
today = _dt.date.today().strftime("%Y-%m-%d")
base_name = f"gendesign_forecast_{cad_safe}_{today}"

View file

@ -58,6 +58,19 @@ _TITLE_CONFIDENCE: str = "Уверенность"
# Краткий RU-дисклеймер (формулировка из ТЗ #959) — после H1, виден сразу.
_ADVISORY_DISCLAIMER: str = "Оценка advisory — не основание для инвест-решения."
# Telegram-сводка (#959): сверхкраткий caveat в финальной строке (≤ одной строки чата).
_TG_ADVISORY_CAVEAT: str = "Оценка advisory, не основание для инвест-решения."
# Заголовок-бренд Telegram-сводки (1-я строка). Plain-text, без Markdown.
_TG_TITLE: str = "Site Finder v2 — сводка §22"
# Канонический порядок сценариев в однострочной сводке дефицита (зеркало report_assembler).
_TG_SCENARIO_ORDER: tuple[str, ...] = ("base", "aggressive", "conservative")
# RU-метки сценариев для компактной строки «дефицит по сценариям».
_TG_SCENARIO_RU: dict[str, str] = {
"base": "база",
"aggressive": "агр",
"conservative": "конс",
}
# Сколько конкурентов выводить в таблицу (top-N, чтобы не раздувать MD). Зеркало
# competitors_limit=5 у snapshot-экспорта.
_COMPETITORS_TOP_N: int = 5
@ -401,3 +414,114 @@ def render_report_markdown(report: Any) -> str:
data.get("advisory"),
)
return markdown
# ──────────────────────────────────────────────────────────────────────────────
# §22 Telegram-сводка — СВЕРХКРАТКИЙ копипаст-сниппет для чата (#959, EPIC export).
# Не файл-download, а текст-в-чат: длинный MD-отчёт не годится для Telegram, нужна
# выжимка ~8-15 строк. PURE, без новых зависимостей (только str), graceful, тот же
# вход/нормализация, что у render_report_markdown. DRY: переиспользует импортированные
# из report_pdf хелперы (`_normalize`/`_fmt`/`_level_ru`/`_as_dict`/`_as_list`/
# `_scenario_deficit_index`) — НЕ дублирует их. Telegram-safe: plain text + «•»-буллеты,
# без тяжёлых Markdown-таблиц (MarkdownV2 капризен) — копипаст-готово как есть.
# ──────────────────────────────────────────────────────────────────────────────
def _tg_scenario_deficit_line(report: dict[str, Any]) -> str | None:
"""Однострочная сводка дефицита по сценариям: «база -1.0 · агр -1.0 · конс -1.0».
PURE, graceful. Берёт `scenarios.by_scenario`, для каждого сценария сводный
deficit_index основного горизонта через импортированный `_scenario_deficit_index`
(DRY не считаем заново). Порядок канонический (`_TG_SCENARIO_ORDER`), но если
встречаются иные имена сценариев добавляем их в конец. Нет данных None
(вызывающий просто пропустит строку graceful, без «нет данных»-шума в чате).
"""
by_scenario = _as_dict(_as_dict(report.get("scenarios")).get("by_scenario"))
if not by_scenario:
return None
ordered = [name for name in _TG_SCENARIO_ORDER if name in by_scenario]
ordered += [name for name in by_scenario if name not in _TG_SCENARIO_ORDER]
parts: list[str] = []
for name in ordered:
label = _TG_SCENARIO_RU.get(name, str(name))
deficit = _scenario_deficit_index(_as_dict(by_scenario.get(name)))
parts.append(f"{label} {_fmt(deficit)}")
return " · ".join(parts) if parts else None
def render_report_telegram_summary(report: Any) -> str:
"""§22 Свернуть `SiteFinderReport` (#987) в КРАТКУЮ Telegram-сводку и вернуть str.
Копипаст-готовый сниппет ~8-15 строк для отправки в чат (длинный MD-отчёт для
Telegram непригоден здесь только выжимка):
1-я строка заголовок-бренд + кадастр + район;
вердикт «что строить» (headline) + ключевые числа компактно (индекс дефицита,
мес. предложения, итоговый скор, уверенность RU);
целевой горизонт + дефицит по сценариям (база/агр/конс) одной строкой, если есть;
рекомендованный класс продукта (`product_tz.obj_class`), если есть;
финальная строка короткий advisory-caveat.
Telegram-safe: PLAIN TEXT с «»-буллетами, БЕЗ Markdown-таблиц (MarkdownV2 капризен).
ДЕТЕРМИНИРОВАННО, БЕЗ LLM/БД/сети, БЕЗ новых зависимостей (чистый str рендер текста;
САМА отправка в Telegram отдельная инфраструктура, вне scope). Принимает КАК
`SiteFinderReport`-инстанс, ТАК и его `as_dict()`-словарь (нормализуется `_normalize`).
GRACEFUL: пустой/тонкий/мусорный отчёт короткая валидная сводка (минимум заголовок
+ caveat), НИКОГДА не падает (нет KeyError на тонком отчёте).
Args:
report: `SiteFinderReport`-инстанс или его `as_dict()`-словарь (или мусор минимум).
Returns:
Короткую (8-15 строк) plain-text строку, готовую для
`Response(media_type="text/plain; charset=utf-8")` и копипаста в Telegram.
"""
data = _normalize(report)
meta = _as_dict(data.get("meta"))
exec_summary = _as_dict(data.get("exec_summary"))
key_numbers = _as_dict(exec_summary.get("key_numbers"))
product = _as_dict(data.get("product_tz"))
lines: list[str] = []
# 1-я строка: бренд + кадастр + район (компактно, plain text).
cad = _fmt(meta.get("cad_num"))
district = _fmt(meta.get("district"))
lines.append(f"{_TG_TITLE} · {cad} · {district}")
# Вердикт «что строить» (headline) — отдельной строкой, если есть.
headline = exec_summary.get("headline")
if headline:
lines.append(str(headline))
# Ключевые числа компактными буллетами (уверенность — RU; уровень читаем из
# key_numbers, иначе из exec_summary.overall_confidence — оба varianta встречаются).
confidence_level = key_numbers.get("confidence") or exec_summary.get("overall_confidence")
lines.append(f"• Индекс дефицита: {_fmt(key_numbers.get('deficit_index'))}")
lines.append(f"Мес. предложения: {_fmt(key_numbers.get('months_of_inventory'))}")
lines.append(f"• Итоговый скор: {_fmt(key_numbers.get('overall_score'))}")
lines.append(f"• Уверенность: {_level_ru(confidence_level)}")
# Целевой горизонт + дефицит по сценариям одной строкой (если есть сценарии).
scenario_line = _tg_scenario_deficit_line(data)
if scenario_line is not None:
horizons = _as_list(meta.get("horizons"))
horizon_note = f" (гор. {_PRIMARY_HORIZON_MONTHS} мес)" if horizons else ""
lines.append(f"• Дефицит по сценариям{horizon_note}: {scenario_line}")
# Рекомендованный класс продукта (если задан).
obj_class = product.get("obj_class")
if obj_class:
lines.append(f"• Рекомендованный класс: {_fmt(obj_class)}")
# Финальная строка — короткий advisory-caveat (всегда).
lines.append(_TG_ADVISORY_CAVEAT)
summary = "\n".join(lines)
logger.info(
"render_report_telegram_summary: cad_num=%s lines=%d len=%d advisory=%s",
meta.get("cad_num"),
len(lines),
len(summary),
data.get("advisory"),
)
return summary

View file

@ -427,6 +427,50 @@ def test_export_forecast_json_returns_dict() -> None:
app.dependency_overrides.clear()
def test_export_forecast_tg_returns_inline_plain_text() -> None:
"""format=tg → 200 text/plain; charset=utf-8, INLINE (НЕТ attachment Content-Disposition)."""
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=tg")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"] == "text/plain; charset=utf-8"
# Сниппет для копипаста — НЕ файл-download: attachment-заголовка быть не должно.
assert "content-disposition" not in {k.lower() for k in resp.headers}
body = resp.text
# Краткая сводка несёт вердикт + advisory-caveat.
assert "комфорт" in body
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_tg_no_run_returns_404() -> None:
"""format=tg + рана нет → 404 (как md/json export-вариант, НЕ 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?format=tg")
assert resp.status_code == 404, resp.text
assert "посчитан" in resp.json()["detail"]
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

View file

@ -0,0 +1,206 @@
"""Unit-тесты §22 Telegram-сводки (#959, EPIC export) — `render_report_telegram_summary`.
Чистые тесты БЕЗ БД/сети (рендер только ПОТРЕБЛЯЕТ уже-собранный отчёт):
реальный fixture-отчёт (frontend mock `parcel-forecast.json` `report`) сворачивается
в КРАТКУЮ plain-text сводку с вердиктом «что строить», ключевыми числами
(дефицит/уверенность) и advisory-caveat;
рендер принимает КАК `SiteFinderReport`-инстанс, ТАК и его `as_dict()`-словарь;
пустой / частичный / мусорный отчёт короткая валидная сводка БЕЗ падения (graceful);
сводка КОРОТКАЯ (Telegram-friendly): ~1500 символов / ~20 строк, без MD-таблиц.
Детерминированно, без LLM. DATABASE_URL выставляем до импорта app-модулей (зеркало
test_report_md.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 (
_TG_ADVISORY_CAVEAT,
render_report_telegram_summary,
)
from app.services.forecasting.report import (
ReportExecSummary,
ReportMeta,
ReportProductTz,
SiteFinderReport,
)
# Реальный fixture: frontend mock §22-форсайта — берём вложенный `report` (форма #987
# `as_dict()`). Так тест не разойдётся с реальной формой отчёта.
_FIXTURE_PATH = (
Path(__file__).resolve().parents[4]
/ "frontend"
/ "src"
/ "lib"
/ "mocks"
/ "parcel-forecast.json"
)
# Telegram-границы «короткости» (мягкие пороги из ТЗ #959).
_MAX_CHARS = 1500
_MAX_LINES = 20
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
# ── Реальный fixture-отчёт: содержание + краткость ──────────────────────────────
class TestFixtureReportSummary:
def test_returns_non_empty_str(self) -> None:
tg = render_report_telegram_summary(_fixture_report())
assert isinstance(tg, str)
assert tg.strip()
def test_contains_verdict_headline(self) -> None:
report = _fixture_report()
headline = report["exec_summary"]["headline"]
tg = render_report_telegram_summary(report)
# Вердикт «что строить» присутствует в сводке.
assert headline.split(".")[0] in tg
def test_contains_deficit_value(self) -> None:
report = _fixture_report()
deficit = report["exec_summary"]["key_numbers"]["deficit_index"]
tg = render_report_telegram_summary(report)
# Сводный индекс дефицита (1.0) попадает в сводку (через _fmt → "-1.0").
assert str(round(float(deficit), 3)) in tg
def test_contains_confidence_label_ru(self) -> None:
# Уверенность отрендерена RU-меткой ("low" → "низкая").
tg = render_report_telegram_summary(_fixture_report())
assert "низкая" in tg
def test_contains_months_of_inventory(self) -> None:
report = _fixture_report()
moi = report["exec_summary"]["key_numbers"]["months_of_inventory"]
tg = render_report_telegram_summary(report)
assert str(round(float(moi), 3)) in tg
def test_contains_recommended_class(self) -> None:
report = _fixture_report()
obj_class = report["product_tz"]["obj_class"]
tg = render_report_telegram_summary(report)
assert obj_class in tg
def test_contains_cad_and_district(self) -> None:
tg = render_report_telegram_summary(_fixture_report())
assert "66:41:0702048:27" in tg
assert "Кировский" in tg
def test_contains_scenario_deficit_line(self) -> None:
# Дефицит по сценариям одной строкой: RU-метки сценариев присутствуют.
tg = render_report_telegram_summary(_fixture_report())
assert "база" in tg
assert "агр" in tg
assert "конс" in tg
def test_contains_advisory_caveat(self) -> None:
tg = render_report_telegram_summary(_fixture_report())
assert _TG_ADVISORY_CAVEAT in tg
# Caveat — последняя строка (финальный дисклеймер).
assert tg.rstrip().endswith(_TG_ADVISORY_CAVEAT)
def test_is_short_telegram_friendly(self) -> None:
# Сводка должна быть КОРОТКОЙ (Telegram): мягкие пороги по символам и строкам.
tg = render_report_telegram_summary(_fixture_report())
assert len(tg) <= _MAX_CHARS, f"сводка слишком длинная: {len(tg)} символов"
assert len(tg.splitlines()) <= _MAX_LINES, (
f"сводка слишком много строк: {len(tg.splitlines())}"
)
def test_no_markdown_tables(self) -> None:
# Telegram-safe: без GFM-таблиц (разделитель шапки не должен встречаться).
tg = render_report_telegram_summary(_fixture_report())
assert "| --- |" not in tg
assert "---" not in tg
# ── Вход: 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]),
product_tz=ReportProductTz(obj_class="комфорт"),
)
tg = render_report_telegram_summary(report)
assert "Строить комфорт" in tg
assert "66:41:0000000:1" in tg
assert "комфорт" in tg
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_telegram_summary(report) == render_report_telegram_summary(
report.as_dict()
)
# ── Graceful: пустой / частичный / мусорный вход → короткая сводка без падения ───
class TestGraceful:
def test_empty_dict_does_not_crash(self) -> None:
# Пустой отчёт {} → короткая валидная сводка (no KeyError) с caveat.
tg = render_report_telegram_summary({})
assert isinstance(tg, str)
assert _TG_ADVISORY_CAVEAT in tg
assert len(tg) <= _MAX_CHARS
assert len(tg.splitlines()) <= _MAX_LINES
def test_none_does_not_crash(self) -> None:
tg = render_report_telegram_summary(None)
assert isinstance(tg, str)
assert _TG_ADVISORY_CAVEAT in tg
def test_default_report_does_not_crash(self) -> None:
# Полностью дефолтный (пустой) SiteFinderReport — короткая валидная сводка.
tg = render_report_telegram_summary(SiteFinderReport())
assert _TG_ADVISORY_CAVEAT in tg
assert len(tg.splitlines()) <= _MAX_LINES
def test_garbage_input_returns_valid_summary(self) -> None:
# Мусор (int / строка / список) → нормализуется в {} → валидная сводка, не падает.
for junk in (123, "not a report", []):
tg = render_report_telegram_summary(junk)
assert isinstance(tg, str)
assert _TG_ADVISORY_CAVEAT in tg
def test_partial_report_no_scenarios_skips_line(self) -> None:
# Тонкий отчёт без сценариев → строки «дефицит по сценариям» нет, но не падает.
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Тонкий анализ."),
meta=ReportMeta(cad_num="66:41:0000000:2", horizons=[12]),
)
tg = render_report_telegram_summary(report)
assert "Тонкий анализ" in tg
assert "66:41:0000000:2" in tg
assert "Дефицит по сценариям" not in tg
assert _TG_ADVISORY_CAVEAT in tg
def test_partial_report_no_obj_class_skips_line(self) -> None:
# Нет obj_class → строки «Рекомендованный класс» нет (graceful).
report = SiteFinderReport(
exec_summary=ReportExecSummary(headline="Без класса."),
meta=ReportMeta(cad_num="66:41:0000000:3"),
)
tg = render_report_telegram_summary(report)
assert "Рекомендованный класс" not in tg