wecom_it_smart_desk/backend/app/services/funny_phrase_service.py

# =============================================================================
# 企微IT智能服务台 — 趣味话术服务
# =============================================================================
# 说明：管理各场景的趣味话术，包括：
#   1. 根据触发场景返回对应话术
#   2. 从 funny_phrases 表读取话术配置
#   3. 支持按员工 VIP 等级自动切换话术（VIP → 正式版话术）
#   4. 预置 6 种场景的默认话术
# =============================================================================

import logging
import random
from typing import Optional

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

from app.models.funny_phrase import FunnyPhrase

logger = logging.getLogger(__name__)


class FunnyPhraseService:
    """趣味话术服务。

    根据触发场景返回对应的趣味话术。
    支持后台动态修改话术内容（通过 funny_phrases 表）。
    """

    # 默认话术（当数据库未配置时使用，和 PRD 一致）
    DEFAULT_PHRASES = {
        "shake": "已为您呼叫人工坐席，请稍等!",
        "keyword": "已为您呼叫人工坐席，请稍等!",
        "waiting": "人还在路上，别急别急～",
        "connected": "坐席正在查看您的信息，请等待处理回复！",
        "timeout": "坐席都在忙，不过AI还在呢，要不先聊聊？我再继续摇",
        "vip": "这就帮您安排专家，请稍候",
    }

    def __init__(self, db: AsyncSession):
        """初始化趣味话术服务。

        Args:
            db: 异步数据库会话
        """
        self.db = db

    # --------------------------------------------------------------------------
    # 获取话术
    # --------------------------------------------------------------------------
    async def get_phrase(
        self, scene: str, is_vip: bool = False
    ) -> str:
        """根据触发场景获取趣味话术。

        优先从 funny_phrases 表读取，如果未配置则使用默认话术。
        VIP 员工自动使用 "vip" 场景的话术。

        场景说明：
        - click_shake / shake: 点击摇人按钮
        - keyword: 关键词触发转人工
        - waiting: 排队等待（30秒无人接单）
        - connected: 坐席接入
        - timeout: 等待超时（2分钟）
        - vip: VIP员工专用

        Args:
            scene: 触发场景（shake/keyword/waiting/connected/timeout/vip）
            is_vip: 是否 VIP 员工（VIP 优先使用 vip 场景话术）

        Returns:
            str: 话术内容
        """
        # VIP 员工优先使用 vip 场景话术
        actual_scene = scene
        if is_vip and scene != "vip":
            # 尝试获取 VIP 话术，如果不存在则回退到原场景
            vip_phrase = await self._get_phrase_from_db("vip")
            if vip_phrase:
                logger.debug(f"VIP员工使用专属话术: scene=vip")
                return vip_phrase

        # 从数据库获取对应场景的话术
        phrase = await self._get_phrase_from_db(actual_scene)

        if phrase:
            return phrase

        # 数据库未配置，使用默认话术
        default = self.DEFAULT_PHRASES.get(actual_scene, "请稍候...")
        logger.debug(f"使用默认话术: scene={actual_scene}")
        return default

    # --------------------------------------------------------------------------
    # 从数据库获取话术
    # --------------------------------------------------------------------------
    async def _get_phrase_from_db(self, scene: str) -> Optional[str]:
        """从 funny_phrases 表获取指定场景的话术。

        同一场景可能有多条话术，随机返回一条（增加趣味性）。
        只返回 is_active=True 的话术。

        Args:
            scene: 触发场景

        Returns:
            Optional[str]: 话术内容，未找到返回 None
        """
        stmt = (
            select(FunnyPhrase)
            .where(
                FunnyPhrase.scene == scene,
                FunnyPhrase.is_active == True,
            )
            .order_by(FunnyPhrase.sort_order)
        )
        result = await self.db.execute(stmt)
        phrases = list(result.scalars().all())

        if not phrases:
            return None

        # 随机选一条（如果有多个话术，增加随机趣味性）
        chosen = random.choice(phrases)
        return chosen.content

    # --------------------------------------------------------------------------
    # 获取所有场景的话术
    # --------------------------------------------------------------------------
    async def get_all_phrases(self) -> dict:
        """获取所有场景的话术。

        用于后台管理页面展示当前话术配置。

        Returns:
            dict: 按场景分组的话术字典
        """
        stmt = select(FunnyPhrase).order_by(
            FunnyPhrase.scene, FunnyPhrase.sort_order
        )
        result = await self.db.execute(stmt)
        phrases = list(result.scalars().all())

        # 按场景分组
        grouped: dict = {}
        for phrase in phrases:
            if phrase.scene not in grouped:
                grouped[phrase.scene] = []
            grouped[phrase.scene].append({
                "id": str(phrase.id),
                "content": phrase.content,
                "tone": phrase.tone,
                "sort_order": phrase.sort_order,
                "is_active": phrase.is_active,
            })

        return grouped