RAG_helper/services/intent_service.py

"""Ветки графовой архитектуры: каталог intents + сид при первом запуске."""
import logging
from pathlib import Path

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from db.models import Intent

logger = logging.getLogger(__name__)

PROMPTS_INTENTS_DIR = Path(__file__).resolve().parent.parent / "prompts" / "intents"

ROUTER_INTENT_CODE = "_router"  # техническая ветка — промпт классификатора (код с `_` = системная).

# Стартовый набор веток (Спринт 4). Порядок — в каком показывать в UI.
# Ветки с кодом, начинающимся на `_`, — системные: их промпты используются служебными
# частями системы (например, классификатор), а не отвечают пациенту напрямую.
SEED_INTENTS: list[dict] = [
    {"code": "new_booking",       "name": "Новая запись",           "description": "Пациент хочет записаться на приём."},
    {"code": "reschedule",        "name": "Перенос / отмена",       "description": "Пациент хочет перенести или отменить существующую запись."},
    {"code": "price_question",    "name": "Цены и ДМС",             "description": "Вопросы про стоимость услуг, оплату, ДМС."},
    {"code": "medical_question",  "name": "Медицинский вопрос",     "description": "Симптомы, лекарства, диагноз — требует врача."},
    {"code": "general_info",      "name": "Общая справка",          "description": "Адрес, часы работы, как доехать, общие вопросы."},
    {"code": "escalate_human",    "name": "Перевод на оператора",   "description": "Перевод диалога на живого оператора."},
    {"code": ROUTER_INTENT_CODE,  "name": "Маршрутизатор",          "description": "Системная ветка: промпт классификатора намерений. Пациенту напрямую не отвечает."},
]


def is_system_code(code: str) -> bool:
    """Код, начинающийся с подчёркивания — системная ветка (не responder)."""
    return code.startswith("_")


def load_seed_prompt(code: str) -> str:
    """Стартовый промпт ветки из prompts/intents/{code}.md. Если файла нет — пустая строка."""
    # Код с точкой или пробелом — явно ошибка, не трогаем файловую систему.
    path = PROMPTS_INTENTS_DIR / f"{code}.md"
    try:
        return path.read_text(encoding="utf-8").strip()
    except FileNotFoundError:
        logger.warning("Seed prompt for intent %r not found at %s", code, path)
        return ""


async def list_intents(session: AsyncSession) -> list[Intent]:
    stmt = select(Intent).order_by(Intent.order_index, Intent.id)
    return list((await session.execute(stmt)).scalars().all())


async def get_intent_by_code(session: AsyncSession, code: str) -> Intent | None:
    stmt = select(Intent).where(Intent.code == code)
    return (await session.execute(stmt)).scalar_one_or_none()


async def set_intent_enabled(session: AsyncSession, code: str, is_enabled: bool) -> Intent | None:
    intent = await get_intent_by_code(session, code)
    if intent is None:
        return None
    intent.is_enabled = is_enabled
    await session.commit()
    await session.refresh(intent)
    return intent


async def ensure_seed_intents(session: AsyncSession) -> None:
    """Досиживает недостающие ветки из SEED_INTENTS. Существующие не трогаются."""
    existing = set((await session.execute(select(Intent.code))).scalars().all())
    added = 0
    for order, data in enumerate(SEED_INTENTS):
        if data["code"] in existing:
            continue
        session.add(Intent(
            code=data["code"],
            name=data["name"],
            description=data["description"],
            is_enabled=True,
            order_index=order,
        ))
        added += 1
    if added:
        await session.commit()
        logger.info("Seeded %d missing intents", added)


# Точечные переименования: пользователь правит UI-копию, но в БД уже залит
# старый сид. Применяем мягко — только если поле в точности совпадает со старым
# значением (значит оператор не правил его сам).
_INTENT_NAME_MIGRATIONS: list[dict] = [
    {
        "code": "escalate_human",
        "old_name": "Эскалация на оператора",
        "new_name": "Перевод на оператора",
        "old_description": "Передача диалога живому оператору.",
        "new_description": "Перевод диалога на живого оператора.",
    },
]


async def migrate_intent_copy(session: AsyncSession) -> None:
    """Обновляет name/description у системных веток, если в БД лежит старый текст."""
    updated = 0
    for spec in _INTENT_NAME_MIGRATIONS:
        intent = await get_intent_by_code(session, spec["code"])
        if intent is None:
            continue
        changed = False
        if intent.name == spec["old_name"]:
            intent.name = spec["new_name"]
            changed = True
        if intent.description == spec["old_description"]:
            intent.description = spec["new_description"]
            changed = True
        if changed:
            updated += 1
    if updated:
        await session.commit()
        logger.info("Migrated copy for %d intent(s)", updated)