ai-robot-core/ai-service/app/services/intent/tester.py

"""
Intent rule tester for AI Service.
[AC-AISVC-96] Intent rule testing service with conflict detection.
"""

import logging
import re
from dataclasses import dataclass, field
from typing import Any

from app.models.entities import IntentRule
from app.services.intent.router import IntentRouter

logger = logging.getLogger(__name__)


@dataclass
class IntentRuleTestCase:
    """Result of testing a single message against a rule."""

    message: str
    matched: bool
    matched_keywords: list[str] = field(default_factory=list)
    matched_patterns: list[str] = field(default_factory=list)
    match_type: str | None = None
    priority: int = 0
    priority_rank: int = 0
    conflict_rules: list[dict[str, Any]] = field(default_factory=list)
    reason: str | None = None

    def to_dict(self) -> dict[str, Any]:
        return {
            "message": self.message,
            "matched": self.matched,
            "matchedKeywords": self.matched_keywords,
            "matchedPatterns": self.matched_patterns,
            "matchType": self.match_type,
            "priority": self.priority,
            "priorityRank": self.priority_rank,
            "conflictRules": self.conflict_rules,
            "reason": self.reason,
        }


@dataclass
class IntentRuleTestResult:
    """Result of testing multiple messages against a rule."""

    rule_id: str
    rule_name: str
    results: list[IntentRuleTestCase]
    summary: dict[str, Any]

    def to_dict(self) -> dict[str, Any]:
        return {
            "ruleId": self.rule_id,
            "ruleName": self.rule_name,
            "results": [r.to_dict() for r in self.results],
            "summary": self.summary,
        }


class IntentRuleTester:
    """
    [AC-AISVC-96] Intent rule testing service.

    Features:
    - Test single rule against multiple messages
    - Detect priority conflicts (other rules that also match)
    - Provide detailed match information
    """

    def __init__(self):
        self._router = IntentRouter()

    async def test_rule(
        self,
        rule: IntentRule,
        test_messages: list[str],
        all_rules: list[IntentRule],
    ) -> IntentRuleTestResult:
        """
        Test a rule against multiple messages and detect conflicts.

        Args:
            rule: The rule to test
            test_messages: List of test messages
            all_rules: All rules for conflict detection (ordered by priority DESC)

        Returns:
            IntentRuleTestResult with detailed test results
        """
        results = []
        priority_rank = self._calculate_priority_rank(rule, all_rules)

        for message in test_messages:
            test_case = self._test_single_message(rule, message, all_rules, priority_rank)
            results.append(test_case)

        matched_count = sum(1 for r in results if r.matched)
        summary = {
            "totalTests": len(test_messages),
            "matchedCount": matched_count,
            "matchRate": matched_count / len(test_messages) if test_messages else 0,
        }

        logger.info(
            f"[AC-AISVC-96] Tested rule {rule.name}: "
            f"{matched_count}/{len(test_messages)} matched"
        )

        return IntentRuleTestResult(
            rule_id=str(rule.id),
            rule_name=rule.name,
            results=results,
            summary=summary,
        )

    def _test_single_message(
        self,
        rule: IntentRule,
        message: str,
        all_rules: list[IntentRule],
        priority_rank: int,
    ) -> IntentRuleTestCase:
        """Test a single message against a rule."""
        matched_keywords = self._match_keywords(message, rule)
        matched_patterns = self._match_patterns(message, rule)

        matched = len(matched_keywords) > 0 or len(matched_patterns) > 0
        match_type = None
        reason = None

        if matched:
            if matched_keywords:
                match_type = "keyword"
            else:
                match_type = "regex"
        else:
            reason = self._determine_unmatch_reason(message, rule)

        conflict_rules = self._detect_conflicts(rule, message, all_rules)

        return IntentRuleTestCase(
            message=message,
            matched=matched,
            matched_keywords=matched_keywords,
            matched_patterns=matched_patterns,
            match_type=match_type,
            priority=rule.priority,
            priority_rank=priority_rank,
            conflict_rules=conflict_rules,
            reason=reason,
        )

    def _match_keywords(self, message: str, rule: IntentRule) -> list[str]:
        """Match message against rule keywords."""
        matched = []
        keywords = rule.keywords or []
        message_lower = message.lower()

        for keyword in keywords:
            if keyword and keyword.lower() in message_lower:
                matched.append(keyword)

        return matched

    def _match_patterns(self, message: str, rule: IntentRule) -> list[str]:
        """Match message against rule regex patterns."""
        matched = []
        patterns = rule.patterns or []

        for pattern in patterns:
            if not pattern:
                continue
            try:
                if re.search(pattern, message, re.IGNORECASE):
                    matched.append(pattern)
            except re.error as e:
                logger.warning(f"Invalid regex pattern: {pattern}, error: {e}")

        return matched

    def _detect_conflicts(
        self,
        current_rule: IntentRule,
        message: str,
        all_rules: list[IntentRule],
    ) -> list[dict[str, Any]]:
        """Detect other rules that also match the message (priority conflicts)."""
        conflicts = []

        for other_rule in all_rules:
            if str(other_rule.id) == str(current_rule.id):
                continue

            if not other_rule.is_enabled:
                continue

            matched_keywords = self._match_keywords(message, other_rule)
            matched_patterns = self._match_patterns(message, other_rule)

            if matched_keywords or matched_patterns:
                conflicts.append({
                    "ruleId": str(other_rule.id),
                    "ruleName": other_rule.name,
                    "priority": other_rule.priority,
                    "reason": f"同时匹配（优先级：{other_rule.priority}）",
                })

        return conflicts

    def _calculate_priority_rank(
        self,
        rule: IntentRule,
        all_rules: list[IntentRule],
    ) -> int:
        """Calculate the priority rank of a rule among all rules."""
        enabled_rules = [r for r in all_rules if r.is_enabled]
        sorted_rules = sorted(enabled_rules, key=lambda r: r.priority, reverse=True)

        for rank, r in enumerate(sorted_rules, start=1):
            if str(r.id) == str(rule.id):
                return rank

        return 0

    def _determine_unmatch_reason(self, message: str, rule: IntentRule) -> str:
        """Determine why a message did not match a rule."""
        keywords = rule.keywords or []
        patterns = rule.patterns or []

        if not keywords and not patterns:
            return "规则未配置关键词或正则表达式"

        if keywords:
            keyword_str = "、".join(keywords[:3])
            if len(keywords) > 3:
                keyword_str += f"等{len(keywords)}个"
            return f"关键词不匹配（规则关键词：{keyword_str}）"

        if patterns:
            return f"正则表达式不匹配（规则模式：{len(patterns)}个）"

        return "未匹配"