bot-backend
14f3ef2019
Co-authored-by: bot-backend <bot-backend@gendsgn.local> Co-committed-by: bot-backend <bot-backend@gendsgn.local>
406 lines
19 KiB
Python
406 lines
19 KiB
Python
"""Роутинг RU-вопроса в intent + сборка шаблонного RU-ответа (#957, Step 1, БЕЗ LLM).
|
||
|
||
ДЕТЕРМИНИРОВАННО:
|
||
• route_intent — явный intent (если задан) ИЛИ keyword/regex-матч RU-вопроса; нет
|
||
совпадения → ChatIntent.unknown.
|
||
• render_answer — тянет нужную(ые) секцию(и) `SiteFinderReport.as_dict()` и собирает
|
||
краткий RU-ответ с числами движка ВЕРБАТИМ (числа берутся ИЗ отчёта, НИКОГДА не
|
||
фабрикуются). Возвращает (answer_text, sections_used).
|
||
|
||
ROBUST к частичному/пустому отчёту (отчёт может быть тонким — каждая секция Optional):
|
||
никаких KeyError — отсутствует секция → честно говорим «данных по X в отчёте нет», НЕ
|
||
выдумываем. Микрокопирайт: RU, нейтральный бизнес-тон (ui-microcopy.md), `₽/м²` с
|
||
пробелами, тысячи через пробел, без emoji/«Готово!». Каждый ответ завершается advisory-
|
||
оговоркой (отчёт советующий, не основание для инвест-решения).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
import re
|
||
from typing import Any
|
||
|
||
from app.schemas.chat import ChatIntent
|
||
|
||
# Неразрывный пробел — разделитель тысяч (зеркало toLocaleString("ru"), ui-microcopy.md:
|
||
# '6 832 540'). Явная константа, чтобы separator не зависел от способа ввода в исходнике.
|
||
_NBSP = " "
|
||
|
||
# advisory-оговорка в конце КАЖДОГО ответа (зеркало report.advisory). Полное
|
||
# предложение, не аббревиатура (ui-microcopy.md).
|
||
_ADVISORY_NOTE = (
|
||
"Отчёт носит советующий характер и не является основанием для инвестиционного решения."
|
||
)
|
||
|
||
# Сообщение, когда конкретной секции в отчёте нет (честный graceful, не фабрикуем).
|
||
_NO_SECTION_TMPL = "Данных по разделу «{name}» в отчёте нет."
|
||
|
||
# RU-уровни уверенности (movement из low/medium/high в человекочитаемое).
|
||
_CONFIDENCE_RU = {
|
||
"high": "высокая",
|
||
"medium": "средняя",
|
||
"low": "низкая",
|
||
}
|
||
|
||
# RU-метки сценариев.
|
||
_SCENARIO_RU = {
|
||
"conservative": "консервативный",
|
||
"base": "базовый",
|
||
"aggressive": "агрессивный",
|
||
}
|
||
|
||
# RU-метки ключей §25 спец-индексов (для intent=risks).
|
||
_INDEX_RU = {
|
||
"launch_window": "окно запуска",
|
||
"product_void": "продуктовое «белое пятно»",
|
||
"cannibalization": "каннибализация портфеля",
|
||
"competitor_strength": "сила конкурентов",
|
||
"artificial_demand": "искусственный (ипотечный) спрос",
|
||
"cost_of_error": "цена ошибки выбора продукта",
|
||
}
|
||
|
||
# ── Keyword-роутинг intent'ов ───────────────────────────────────────────────────
|
||
# Каждый intent → список regex'ов по строчному RU-вопросу. Порядок проверки задан
|
||
# _INTENT_ORDER (specific → general): «почему такой прогноз» матчим РАНЬШЕ голого
|
||
# «прогноз», иначе why_forecast перехватится более общим правилом.
|
||
_INTENT_PATTERNS: dict[ChatIntent, list[re.Pattern[str]]] = {
|
||
ChatIntent.what_to_build: [
|
||
re.compile(r"что\s+(тут|здесь|строить|возвод)"),
|
||
re.compile(r"какой\s+продукт"),
|
||
re.compile(r"квартирограф|нарезк[аеуи]\b"),
|
||
re.compile(r"какой\s+класс"),
|
||
re.compile(r"\bтз\b|техзадан"),
|
||
],
|
||
ChatIntent.why_forecast: [
|
||
re.compile(r"почему\s+(такой|так)\b"),
|
||
re.compile(r"на\s+чём\s+основан"),
|
||
re.compile(r"откуда\s+(такой|эти|цифр|прогноз)"),
|
||
re.compile(r"чем\s+обоснован"),
|
||
re.compile(r"(почему|обоснуй).{0,20}прогноз"),
|
||
],
|
||
ChatIntent.risks: [
|
||
re.compile(r"риск"),
|
||
re.compile(r"опасн|угроз"),
|
||
re.compile(r"каннибал"),
|
||
re.compile(r"подводн|что\s+может\s+пойти"),
|
||
re.compile(r"спец.?индекс|индекс[ыаов]\b"),
|
||
],
|
||
ChatIntent.scenarios: [
|
||
re.compile(r"сценари"),
|
||
re.compile(r"консерватив|агрессивн|базов"),
|
||
re.compile(r"разброс|варианты\s+развит"),
|
||
],
|
||
ChatIntent.summary: [
|
||
re.compile(r"резюме|итог|кратк|сводк|вывод"),
|
||
re.compile(r"вердикт|общая\s+картин"),
|
||
re.compile(r"расскажи\s+про\s+участ|стоит\s+ли"),
|
||
],
|
||
}
|
||
|
||
# Порядок проверки: более СПЕЦИФИЧНЫЕ intent'ы раньше общих.
|
||
_INTENT_ORDER: tuple[ChatIntent, ...] = (
|
||
ChatIntent.why_forecast,
|
||
ChatIntent.what_to_build,
|
||
ChatIntent.risks,
|
||
ChatIntent.scenarios,
|
||
ChatIntent.summary,
|
||
)
|
||
|
||
|
||
def route_intent(message: str, explicit: ChatIntent | None) -> ChatIntent:
|
||
"""Определить intent: явный (приоритет) ИЛИ keyword-матч; нет совпадения → unknown.
|
||
|
||
Явный intent уважаем как есть (фронт/клиент знает лучше). Иначе нормализуем текст
|
||
в нижний регистр и идём по _INTENT_ORDER (specific → general): первый сматчившийся
|
||
pattern выигрывает. Ничего не сматчилось → ChatIntent.unknown (отдадим меню тем).
|
||
"""
|
||
if explicit is not None:
|
||
return explicit
|
||
|
||
text = message.lower().strip()
|
||
if not text:
|
||
return ChatIntent.unknown
|
||
|
||
for intent in _INTENT_ORDER:
|
||
for pattern in _INTENT_PATTERNS[intent]:
|
||
if pattern.search(text):
|
||
return intent
|
||
return ChatIntent.unknown
|
||
|
||
|
||
# ── Хелперы форматирования (RU-микрокопирайт) ────────────────────────────────────
|
||
|
||
|
||
def _fmt_thousands(value: float | int) -> str:
|
||
"""Целое с разделителем-NBSP тысяч ('6 832 540', зеркало ru-locale). PURE."""
|
||
return f"{round(value):,}".replace(",", _NBSP)
|
||
|
||
|
||
def _fmt_number(value: Any) -> str | None:
|
||
"""Привести число к RU-строке (тысячи NBSP), не-числа → str() или None.
|
||
|
||
None → None (вызывающий пропустит поле). int → тысячи NBSP. float: целозначный →
|
||
тысячи NBSP; дробный → запятая-десятичный БЕЗ потери точности (число из отчёта уже
|
||
округлено движком — НЕ пере-округляем, лишь меняем '.'→','). Прочее (готовая строка-
|
||
метка) → как есть.
|
||
"""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, bool): # bool — подтип int, не форматируем как число
|
||
return str(value)
|
||
if isinstance(value, int):
|
||
return _fmt_thousands(value)
|
||
if isinstance(value, float):
|
||
if not math.isfinite(value): # NaN/Inf — не число: честно пропускаем (как None),
|
||
return None # иначе в RU-прозу утёк бы англ. литерал 'nan'/'inf' (#1585)
|
||
if value == int(value):
|
||
return _fmt_thousands(value)
|
||
# Точность отчёта сохраняем: repr float'а → '.'→','. (0.31 → '0,31').
|
||
return repr(value).replace(".", ",")
|
||
return str(value)
|
||
|
||
|
||
def _confidence_ru(level: Any) -> str | None:
|
||
"""RU-метка уровня уверенности или None (если нет/неизвестный). PURE."""
|
||
if not isinstance(level, str):
|
||
return None
|
||
return _CONFIDENCE_RU.get(level.lower())
|
||
|
||
|
||
def _assemble(lines: list[str]) -> str:
|
||
"""Склеить непустые строки-абзацы в ответ + advisory-оговорка в конце. PURE.
|
||
|
||
Пустые/None-строки выкидываем (graceful: пропущенные поля не оставляют дыр).
|
||
advisory-оговорка добавляется ВСЕГДА последней строкой.
|
||
"""
|
||
body = [ln for ln in lines if ln]
|
||
body.append(_ADVISORY_NOTE)
|
||
return "\n".join(body)
|
||
|
||
|
||
# ── Рендереры по секциям ─────────────────────────────────────────────────────────
|
||
|
||
|
||
def _render_summary(report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""summary → exec_summary (headline / verdict / key_numbers / overall_confidence)."""
|
||
section = report.get("exec_summary")
|
||
if not isinstance(section, dict):
|
||
return _assemble([_NO_SECTION_TMPL.format(name="резюме")]), []
|
||
|
||
lines: list[str] = []
|
||
headline = section.get("headline")
|
||
if headline:
|
||
lines.append(f"Вердикт: {headline}")
|
||
verdict = section.get("verdict")
|
||
if verdict:
|
||
lines.append(str(verdict))
|
||
|
||
key_numbers = section.get("key_numbers")
|
||
if isinstance(key_numbers, dict) and key_numbers:
|
||
parts = []
|
||
for label, raw in key_numbers.items():
|
||
shown = _fmt_number(raw)
|
||
if shown is not None:
|
||
parts.append(f"{label}: {shown}")
|
||
if parts:
|
||
lines.append("Ключевые числа — " + "; ".join(parts) + ".")
|
||
|
||
conf = _confidence_ru(section.get("overall_confidence"))
|
||
if conf:
|
||
lines.append(f"Общая уверенность: {conf}.")
|
||
|
||
if not lines:
|
||
lines.append("В отчёте пока нет заполненного резюме по участку.")
|
||
return _assemble(lines), ["exec_summary"]
|
||
|
||
|
||
def _render_what_to_build(report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""what_to_build → product_tz (класс / mix / коммерция / USP) + exec_summary.headline."""
|
||
section = report.get("product_tz")
|
||
sections_used: list[str] = []
|
||
lines: list[str] = []
|
||
|
||
exec_summary = report.get("exec_summary")
|
||
if isinstance(exec_summary, dict) and exec_summary.get("headline"):
|
||
lines.append(f"Вердикт: {exec_summary['headline']}")
|
||
sections_used.append("exec_summary")
|
||
|
||
if not isinstance(section, dict):
|
||
lines.append(_NO_SECTION_TMPL.format(name="рекомендация продукта"))
|
||
return _assemble(lines), sections_used
|
||
|
||
sections_used.append("product_tz")
|
||
|
||
obj_class = section.get("obj_class")
|
||
if obj_class:
|
||
lines.append(f"Рекомендованный класс: {obj_class}.")
|
||
|
||
mix = section.get("mix")
|
||
if isinstance(mix, list) and mix:
|
||
lines.append(f"Квартирография проработана ({len(mix)} форматов).")
|
||
|
||
commercial = section.get("commercial")
|
||
if isinstance(commercial, dict) and commercial:
|
||
lines.append("Заложен коммерческий блок (см. раздел продукта).")
|
||
|
||
usp = section.get("usp")
|
||
if isinstance(usp, list) and usp:
|
||
lines.append(f"Выделено USP-преимуществ: {len(usp)}.")
|
||
|
||
summary = section.get("summary")
|
||
if summary:
|
||
lines.append(str(summary))
|
||
|
||
if not any(
|
||
section.get(k) for k in ("obj_class", "mix", "commercial", "usp", "summary")
|
||
):
|
||
lines.append("Раздел рекомендации продукта в отчёте пуст.")
|
||
return _assemble(lines), sections_used
|
||
|
||
|
||
def _render_why_forecast(report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""why_forecast → future_market (по горизонтам + future-supply) + product_tz.reasons +
|
||
confidence.rationale."""
|
||
sections_used: list[str] = []
|
||
lines: list[str] = []
|
||
|
||
future = report.get("future_market")
|
||
if isinstance(future, dict):
|
||
future_emitted = False # секцию в provenance кладём только если выведена строка (#1630)
|
||
horizons = future.get("forecasts_by_horizon")
|
||
if isinstance(horizons, list) and horizons:
|
||
lines.append(f"Прогноз построен по {len(horizons)} горизонтам спроса/предложения.")
|
||
future_emitted = True
|
||
future_supply = future.get("future_supply")
|
||
if isinstance(future_supply, dict) and future_supply:
|
||
lines.append("Учтено давление будущего предложения (выходящие проекты).")
|
||
future_emitted = True
|
||
summary = future.get("summary")
|
||
if summary:
|
||
lines.append(str(summary))
|
||
future_emitted = True
|
||
if future_emitted:
|
||
sections_used.append("future_market")
|
||
else:
|
||
lines.append(_NO_SECTION_TMPL.format(name="будущий рынок"))
|
||
|
||
product = report.get("product_tz")
|
||
if isinstance(product, dict):
|
||
reasons = product.get("reasons")
|
||
if isinstance(reasons, list) and reasons:
|
||
lines.append(f"Рекомендация опирается на {len(reasons)} обоснований (§16).")
|
||
sections_used.append("product_tz")
|
||
|
||
confidence = report.get("confidence")
|
||
if isinstance(confidence, dict):
|
||
rationale = confidence.get("rationale")
|
||
if rationale:
|
||
lines.append(f"Обоснование уверенности: {rationale}")
|
||
sections_used.append("confidence")
|
||
|
||
if not lines or all(ln.startswith("Данных по") for ln in lines):
|
||
lines.append("Детали обоснования прогноза в отчёте пока не заполнены.")
|
||
return _assemble(lines), sections_used
|
||
|
||
|
||
def _render_risks(report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""risks → scoring.special_indices (§25, вкл. каннибализацию) + confidence (уровень +
|
||
тянущие факторы)."""
|
||
sections_used: list[str] = []
|
||
lines: list[str] = []
|
||
|
||
scoring = report.get("scoring")
|
||
special = scoring.get("special_indices") if isinstance(scoring, dict) else None
|
||
if isinstance(special, dict):
|
||
indices = special.get("indices")
|
||
if isinstance(indices, dict) and indices:
|
||
sections_used.append("scoring")
|
||
parts = []
|
||
for key, idx in indices.items():
|
||
if not isinstance(idx, dict):
|
||
continue
|
||
name = _INDEX_RU.get(key, key)
|
||
value = idx.get("value")
|
||
label = idx.get("label")
|
||
if value is None and not label:
|
||
continue # индекс не посчитан (тонкие данные) — пропускаем честно
|
||
shown = label if label else _fmt_number(value)
|
||
if shown is not None:
|
||
parts.append(f"{name}: {shown}")
|
||
if parts:
|
||
lines.append("Специальные индексы §25 — " + "; ".join(parts) + ".")
|
||
if not lines:
|
||
lines.append("Специальные индексы §25 в отчёте не посчитаны (тонкие данные).")
|
||
else:
|
||
lines.append(_NO_SECTION_TMPL.format(name="специальные индексы §25"))
|
||
|
||
confidence = report.get("confidence")
|
||
if isinstance(confidence, dict):
|
||
level = _confidence_ru(confidence.get("level"))
|
||
if level:
|
||
lines.append(f"Уверенность отчёта: {level}.")
|
||
sections_used.append("confidence")
|
||
factors = confidence.get("factors")
|
||
if isinstance(factors, dict) and factors:
|
||
lines.append(f"Факторы, влияющие на уверенность: {len(factors)}.")
|
||
if "confidence" not in sections_used:
|
||
sections_used.append("confidence")
|
||
|
||
return _assemble(lines), sections_used
|
||
|
||
|
||
def _render_scenarios(report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""scenarios → scenarios (by_scenario + summary)."""
|
||
section = report.get("scenarios")
|
||
if not isinstance(section, dict):
|
||
return _assemble([_NO_SECTION_TMPL.format(name="сценарии")]), []
|
||
|
||
lines: list[str] = []
|
||
by_scenario = section.get("by_scenario")
|
||
if isinstance(by_scenario, dict) and by_scenario:
|
||
names: list[str] = [_SCENARIO_RU.get(str(k), str(k)) for k in by_scenario]
|
||
lines.append("Проработаны сценарии: " + ", ".join(names) + ".")
|
||
else:
|
||
lines.append("Разбивка по сценариям в отчёте не заполнена.")
|
||
|
||
summary = section.get("summary")
|
||
if summary:
|
||
lines.append(str(summary))
|
||
return _assemble(lines), ["scenarios"]
|
||
|
||
|
||
def _render_unknown(report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""unknown → graceful RU-меню поддерживаемых типов вопросов (секции не трогаем)."""
|
||
menu = _assemble(
|
||
[
|
||
"Я отвечаю по форсайт-отчёту участка. Могу помочь с такими вопросами:",
|
||
"• что здесь строить и какой продукт рекомендуется;",
|
||
"• почему такой прогноз и на чём он основан;",
|
||
"• какие риски и специальные индексы (включая каннибализацию);",
|
||
"• как различаются сценарии (консервативный / базовый / агрессивный);",
|
||
"• краткое резюме и общий вердикт по участку.",
|
||
]
|
||
)
|
||
return menu, []
|
||
|
||
|
||
# Диспетч intent → рендерер. unknown не зависит от отчёта.
|
||
_RENDERERS = {
|
||
ChatIntent.summary: _render_summary,
|
||
ChatIntent.what_to_build: _render_what_to_build,
|
||
ChatIntent.why_forecast: _render_why_forecast,
|
||
ChatIntent.risks: _render_risks,
|
||
ChatIntent.scenarios: _render_scenarios,
|
||
ChatIntent.unknown: _render_unknown,
|
||
}
|
||
|
||
|
||
def render_answer(intent: ChatIntent, report: dict[str, Any]) -> tuple[str, list[str]]:
|
||
"""Собрать RU-ответ под intent из секции(й) отчёта → (answer_text, sections_used).
|
||
|
||
Числа берутся ВЕРБАТИМ из `report` (никогда не фабрикуются). ROBUST к частичному
|
||
отчёту: отсутствует секция → честное «данных по X нет», без KeyError. Ответ всегда
|
||
заканчивается advisory-оговоркой (см. _assemble). Неизвестный intent → меню тем.
|
||
"""
|
||
renderer = _RENDERERS.get(intent, _render_unknown)
|
||
return renderer(report)
|