wecom_it_smart_desk/backend/app/services/exclusion_service.py

# =============================================================================
# 企微IT智能服务台 — 代答排除匹配引擎
# =============================================================================
# 说明：责任链调度入口，按优先级排序规则，依次调用对应 Matcher，
#   命中即停止并记录日志、更新 hit_count。
#
#   核心方法：
#     1. check_exclusions — 检查消息是否命中排除规则
#     2. test_match — 测试匹配（管理后台用，不记录日志）
#     3. execute_action — 执行命中后动作（4种）
# =============================================================================

import logging
import uuid
from datetime import datetime
from typing import Any, Dict, List, Optional

from sqlalchemy import func, select, and_
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.exclusion_log import ExclusionLog
from app.models.exclusion_rule import ExclusionRule
from app.services.matchers import MATCHER_REGISTRY, MatchResult

logger = logging.getLogger(__name__)

# 优先级排序权重（P0 最高）
_PRIORITY_ORDER = {"P0": 0, "P1": 1, "P2": 2, "P3": 3}


class ExclusionCheckResult:
    """排除检查结果。

    Attributes:
        matched: 是否命中
        rule_id: 命中的规则ID
        rule_name: 命中的规则名称
        match_type: 匹配方式
        matched_detail: 命中详情
        action_type: 命中后动作类型
        transfer_message: 转人工提示语
    """

    def __init__(
        self,
        matched: bool = False,
        rule_id: str = "",
        rule_name: str = "",
        match_type: str = "",
        matched_detail: str = "",
        action_type: str = "",
        transfer_message: str = "",
    ):
        self.matched = matched
        self.rule_id = rule_id
        self.rule_name = rule_name
        self.match_type = match_type
        self.matched_detail = matched_detail
        self.action_type = action_type
        self.transfer_message = transfer_message


class ExclusionService:
    """代答排除匹配引擎。

    责任链调度：
    1. 查询所有启用的排除规则，按优先级排序（P0 > P1 > P2 > P3）
    2. 依次调用对应 Matcher 进行匹配
    3. 命中即停止，记录 exclusion_logs，更新 hit_count
    4. 返回命中结果（含 action_type）
    """

    async def check_exclusions(
        self,
        db: AsyncSession,
        message: str,
        conversation_id: str,
        user_id: str,
    ) -> ExclusionCheckResult:
        """检查消息是否命中排除规则。

        Args:
            db: 数据库会话
            message: 用户消息文本
            conversation_id: 会话ID
            user_id: 用户ID

        Returns:
            ExclusionCheckResult: 检查结果
        """
        # 查询所有启用的规则
        result = await db.execute(
            select(ExclusionRule)
            .where(ExclusionRule.status == "enabled")
            .order_by(ExclusionRule.priority, ExclusionRule.created_at)
        )
        rules = result.scalars().all()

        if not rules:
            return ExclusionCheckResult(matched=False)

        # 按优先级排序（P0 > P1 > P2 > P3）
        rules_sorted = sorted(
            rules,
            key=lambda r: _PRIORITY_ORDER.get(r.priority, 99),
        )

        # 构建上下文
        context: Dict[str, Any] = {
            "conversation_id": conversation_id,
            "user_id": user_id,
            "db": db,
        }

        # 责任链：依次调用对应 Matcher
        for rule in rules_sorted:
            matcher = MATCHER_REGISTRY.get(rule.match_type)
            if matcher is None:
                logger.warning("未知匹配类型: %s, rule_id=%s", rule.match_type, rule.id)
                continue

            try:
                match_result: MatchResult = await matcher.match(
                    message=message,
                    condition=rule.match_condition,
                    context=context,
                )
            except Exception as e:
                logger.error(
                    "匹配器异常: rule=%s, type=%s, error=%s",
                    rule.rule_name, rule.match_type, e,
                )
                continue

            if match_result.matched:
                # 命中！记录日志、更新 hit_count
                await self._log_hit(
                    db=db,
                    rule=rule,
                    message=message,
                    conversation_id=conversation_id,
                    user_id=user_id,
                    match_result=match_result,
                )

                logger.info(
                    "排除规则命中: rule=%s, type=%s, detail=%s, action=%s",
                    rule.rule_name, rule.match_type,
                    match_result.matched_detail, rule.action_type,
                )

                return ExclusionCheckResult(
                    matched=True,
                    rule_id=rule.id,
                    rule_name=rule.rule_name,
                    match_type=rule.match_type,
                    matched_detail=match_result.matched_detail,
                    action_type=rule.action_type,
                    transfer_message=rule.transfer_message or "",
                )

        return ExclusionCheckResult(matched=False)

    async def test_match(
        self,
        db: AsyncSession,
        message: str,
        rule_id: Optional[str] = None,
    ) -> ExclusionCheckResult:
        """测试匹配（管理后台用，不记录日志、不更新 hit_count）。

        Args:
            db: 数据库会话
            message: 测试消息文本
            rule_id: 指定规则ID（可选，不指定则测试所有启用规则）

        Returns:
            ExclusionCheckResult: 测试结果
        """
        if rule_id:
            # 测试指定规则
            result = await db.execute(
                select(ExclusionRule).where(ExclusionRule.id == rule_id)
            )
            rule = result.scalar_one_or_none()
            if not rule:
                return ExclusionCheckResult(matched=False)
            rules_to_test = [rule]
        else:
            # 测试所有启用规则
            result = await db.execute(
                select(ExclusionRule)
                .where(ExclusionRule.status == "enabled")
                .order_by(ExclusionRule.priority)
            )
            rules_to_test = result.scalars().all()

        rules_sorted = sorted(
            rules_to_test,
            key=lambda r: _PRIORITY_ORDER.get(r.priority, 99),
        )

        context: Dict[str, Any] = {
            "conversation_id": "",
            "user_id": "",
            "db": db,
        }

        for rule in rules_sorted:
            matcher = MATCHER_REGISTRY.get(rule.match_type)
            if matcher is None:
                continue

            try:
                match_result = await matcher.match(
                    message=message,
                    condition=rule.match_condition,
                    context=context,
                )
            except Exception as e:
                logger.error("测试匹配异常: rule=%s, error=%s", rule.rule_name, e)
                continue

            if match_result.matched:
                return ExclusionCheckResult(
                    matched=True,
                    rule_id=rule.id,
                    rule_name=rule.rule_name,
                    match_type=rule.match_type,
                    matched_detail=match_result.matched_detail,
                    action_type=rule.action_type,
                    transfer_message=rule.transfer_message or "",
                )

        return ExclusionCheckResult(matched=False)

    async def get_stats(self, db: AsyncSession) -> Dict[str, Any]:
        """获取排除规则统计概要。

        Args:
            db: 数据库会话

        Returns:
            Dict[str, Any]: {enabled_count, disabled_count, monthly_hits, monthly_transfers}
        """
        # 启用规则数
        enabled_result = await db.execute(
            select(func.count()).select_from(ExclusionRule).where(
                ExclusionRule.status == "enabled"
            )
        )
        enabled_count = enabled_result.scalar() or 0

        # 停用规则数
        disabled_result = await db.execute(
            select(func.count()).select_from(ExclusionRule).where(
                ExclusionRule.status == "disabled"
            )
        )
        disabled_count = disabled_result.scalar() or 0

        # 本月命中次数
        now = datetime.now()
        month_start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
        hits_result = await db.execute(
            select(func.count()).select_from(ExclusionLog).where(
                ExclusionLog.created_at >= month_start
            )
        )
        monthly_hits = hits_result.scalar() or 0

        # 本月转人工次数（action_type 含 transfer 的日志）
        transfer_result = await db.execute(
            select(func.count()).select_from(ExclusionLog).where(
                and_(
                    ExclusionLog.created_at >= month_start,
                    ExclusionLog.action_type.like("transfer%"),
                )
            )
        )
        monthly_transfers = transfer_result.scalar() or 0

        return {
            "enabled_count": enabled_count,
            "disabled_count": disabled_count,
            "monthly_hits": monthly_hits,
            "monthly_transfers": monthly_transfers,
        }

    async def _log_hit(
        self,
        db: AsyncSession,
        rule: ExclusionRule,
        message: str,
        conversation_id: str,
        user_id: str,
        match_result: MatchResult,
    ) -> None:
        """记录命中日志并更新 hit_count。

        Args:
            db: 数据库会话
            rule: 命中的规则对象
            message: 用户消息文本
            conversation_id: 会话ID
            user_id: 用户ID
            match_result: 匹配结果
        """
        # 创建命中日志
        log_entry = ExclusionLog(
            id=str(uuid.uuid4()),
            rule_id=rule.id,
            rule_name=rule.rule_name,
            conversation_id=conversation_id,
            user_id=user_id,
            message_content=message[:2000] if message else "",
            match_type=rule.match_type,
            matched_detail=match_result.matched_detail,
            action_type=rule.action_type,
            action_result="success",
        )
        db.add(log_entry)

        # 更新 hit_count
        rule.hit_count = (rule.hit_count or 0) + 1
        rule.updated_at = datetime.now()

        await db.commit()


# 单例
_exclusion_service: Optional[ExclusionService] = None


def get_exclusion_service() -> ExclusionService:
    """获取 ExclusionService 单例。

    Returns:
        ExclusionService: 单例实例
    """
    global _exclusion_service
    if _exclusion_service is None:
        _exclusion_service = ExclusionService()
    return _exclusion_service