LQAgentPlatform/utils_test/Sensitive_Test/test_sensitive_check_standalone.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
敏感词检查功能单元测试（完整版本）
测试 AIReviewEngine.check_sensitive 方法
包含大模型二审功能，支持自定义检查文本
"""

import asyncio
import sys
import os
import time
from pathlib import Path
from typing import Dict, Any, Optional

# 添加项目根目录到Python路径
current_dir = Path(__file__).parent.absolute()
project_root = current_dir.parent.parent
sys.path.insert(0, str(project_root))
os.chdir(str(project_root))


class SimpleLogger:
    """简单日志类"""
    
    @staticmethod
    def info(msg):
        print(f"[INFO] {msg}")
    
    @staticmethod
    def warning(msg):
        print(f"[WARNING] {msg}")
    
    @staticmethod
    def error(msg):
        print(f"[ERROR] {msg}")
    
    @staticmethod
    def exception(msg):
        import traceback
        print(f"[EXCEPTION] {msg}")
        traceback.print_exc()


logger = SimpleLogger()


class TestSensitiveCheck:
    """敏感词检查功能测试类"""

    def __init__(self):
        self.test_results = []
        self.ai_review_engine = None
        logger.info("=" * 80)
        logger.info("初始化敏感词检查测试类")
        logger.info("=" * 80)

    def initialize_engine(self):
        """初始化AI审查引擎"""
        try:
            from core.construction_review.component.ai_review_engine import AIReviewEngine
            from core.base.task_models import TaskFileInfo
            
            # 创建 mock 的 TaskFileInfo 对象
            logger.info("创建 mock TaskFileInfo 对象...")
            mock_task_info = TaskFileInfo(
                file_id="test_file_001",
                callback_task_id="test_task_001",
                user_id="test_user",
                review_config=["sensitive_check"],  # 只启用敏感词检查
                project_plan_type="test_project",
                tendency_review_role="test_role"
            )
            
            logger.info("初始化 AI 审查引擎...")
            self.ai_review_engine = AIReviewEngine(task_file_info=mock_task_info)
            logger.info("AI审查引擎初始化成功")
            return True
        except Exception as e:
            logger.error(f"AI审查引擎初始化失败: {e}")
            logger.exception("详细错误:")
            return False

    async def check_sensitive_with_llm_review(
        self, 
        trace_id_idx: str, 
        review_content: str, 
        review_references: str = "",
        review_location_label: str = "",
        state: Dict = None,
        stage_name: str = "测试阶段"
    ) -> Any:
        """
        完整版的敏感信息检查函数（包含大模型二审）
        
        功能说明：
        1. 使用关键词匹配进行敏感词检测
        2. 如果检测到敏感词，调用大模型进行二审（与原函数一致）
        3. 如果未检测到敏感词，返回成功结果
        4. 移除了消息推送功能（简化测试）
        
        Args:
            trace_id_idx: 追踪ID索引
            review_content: 审查内容
            review_references: 审查参考信息
            review_location_label: 审查位置标签
            state: 状态字典
            stage_name: 阶段名称
            
        Returns:
            审查结果对象
        """
        try:
            from core.construction_review.component.reviewers.utils import check_sensitive_words_async
            from core.construction_review.component.reviewers.base_reviewer import ReviewResult
        except ImportError as e:
            logger.error(f"导入模块失败: {e}")
            return None
        
        logger.info("=" * 80)
        logger.info(f"开始执行敏感词检查 - trace_id: sensitive_check{trace_id_idx}")
        logger.info(f"阶段名称: {stage_name}")
        logger.info(f"审查位置: {review_location_label}")
        logger.info(f"审查内容长度: {len(review_content)} 字符")
        if len(review_content) > 100:
            logger.info(f"审查内容预览: {review_content[:100]}...")
        else:
            logger.info(f"审查内容: {review_content}")
        logger.info("=" * 80)
        
        start_time = time.time()
        trace_id = "sensitive_check" + trace_id_idx
        
        # 第一步：使用关键词匹配式审查
        logger.info("步骤1: 开始关键词匹配检测...")
        try:
            first_results = await check_sensitive_words_async(review_content)
        except Exception as e:
            logger.error(f"关键词检测失败: {e}")
            logger.exception("详细错误:")
            return None
        
        detection_time = time.time() - start_time
        logger.info(f"步骤1: 关键词检测完成，耗时: {detection_time:.4f}s")
        
        # 判断是否检测到敏感词
        if first_results:
            logger.warning(f"步骤2: 检测到 {len(first_results)} 个敏感词，准备送入大模型二审")
            logger.warning("-" * 80)
            
            # 格式化敏感词列表
            sensitive_words_info = []
            for idx, item in enumerate(first_results, 1):
                word_info = f"敏感词: {item['word']}, 位置: {item['position']}-{item['end_position']}, 来源: {item['source']}"
                sensitive_words_info.append(word_info)
                logger.warning(f"  敏感词 #{idx}: {word_info}")
            
            formatted_sensitive_words = "\n".join(sensitive_words_info)
            logger.warning("-" * 80)
            
            # 调用大模型进行二审（与原函数保持一致）
            logger.info("步骤3: 调用大模型进行二审...")
            try:
                if not self.ai_review_engine:
                    logger.error("AI审查引擎未初始化")
                    return None
                
                # 调用 review 方法进行大模型二审
                result = await self.ai_review_engine.review(
                    "sensitive_check", 
                    trace_id, 
                    "basic", 
                    "sensitive_word_check",
                    review_content, 
                    formatted_sensitive_words,
                    None, 
                    review_location_label, 
                    state, 
                    stage_name
                )
                
                llm_review_time = time.time() - start_time
                logger.info(f"步骤3: 大模型二审完成，总耗时: {llm_review_time:.4f}s")
                logger.info("=" * 80)
                
                return result
                
            except Exception as e:
                logger.error(f"大模型二审失败: {e}")
                logger.exception("详细错误:")
                return None
            
        else:
            # 没有检测到敏感词，构造返回体
            logger.info("步骤2: 未检测到敏感词")
            
            execution_time = time.time() - start_time
            result = ReviewResult(
                success=True,
                details={"name": "sensitive_check", "response": "无明显问题"},
                error_message=None,
                execution_time=execution_time
            )
            
            logger.info(f"步骤3: 敏感词检查完成（未检测到敏感词），总耗时: {execution_time:.4f}s")
            logger.info("=" * 80)
            
            return result

    def get_custom_text_from_user(self) -> Optional[str]:
        """从用户获取自定义检查文本"""
        logger.info("\n" + "=" * 80)
        logger.info("请输入要检查的文本内容")
        logger.info("=" * 80)
        logger.info("提示：")
        logger.info("  1. 可以输入多行文本，输入完成后单独一行输入 'END' 结束")
        logger.info("  2. 直接按回车（输入空行）将跳过自定义输入")
        logger.info("=" * 80)
        
        lines = []
        first_line = True
        
        while True:
            try:
                if first_line:
                    line = input(">>> ")
                    first_line = False
                    
                    # 如果第一行就是空行，跳过自定义输入
                    if not line.strip():
                        logger.info("跳过自定义文本输入")
                        return None
                else:
                    line = input("... ")
                
                # 检查是否结束输入
                if line.strip().upper() == 'END':
                    break
                    
                lines.append(line)
                
            except EOFError:
                break
            except KeyboardInterrupt:
                logger.info("\n用户取消输入")
                return None
        
        custom_text = "\n".join(lines)
        
        if not custom_text.strip():
            logger.info("未输入有效文本")
            return None
        
        logger.info(f"\n已接收文本，长度: {len(custom_text)} 字符")
        return custom_text

    async def run_custom_text_check(self, custom_text: str):
        """运行自定义文本检查"""
        logger.info("\n" + "█" * 80)
        logger.info("执行自定义文本敏感词检查")
        logger.info("█" * 80)
        
        try:
            # 执行检查
            result = await self.check_sensitive_with_llm_review(
                trace_id_idx="_custom_001",
                review_content=custom_text,
                review_location_label="自定义文本检查",
                stage_name="自定义测试"
            )
            
            # 显示结果
            logger.info("\n" + "-" * 80)
            logger.info("检查结果:")
            logger.info("-" * 80)
            
            if result:
                logger.info(f"  返回类型: {type(result).__name__}")
                logger.info(f"  是否成功: {result.success}")
                logger.info(f"  执行时间: {result.execution_time:.4f}s")
                logger.info(f"  详细信息: {result.details}")
                
                if result.error_message:
                    logger.error(f"  错误信息: {result.error_message}")
                
                if not result.success:
                    logger.warning("\n检测到敏感内容，请查看上方详细信息")
                else:
                    logger.info("\n✓ 未检测到敏感内容")
            else:
                logger.error("检查失败，未返回结果")
            
            logger.info("-" * 80)
            
        except Exception as e:
            logger.error(f"✗ 检查异常: {str(e)}")
            logger.exception("详细异常信息:")

    async def run_interactive_mode(self):
        """运行交互模式"""
        logger.info("\n" + "█" * 80)
        logger.info("敏感词检查 - 交互模式")
        logger.info("█" * 80)
        
        # 初始化AI审查引擎
        if not self.initialize_engine():
            logger.error("AI审查引擎初始化失败，无法继续")
            return False
        
        # 初始化敏感词检测器
        try:
            from core.construction_review.component.reviewers.utils import SensitiveWordChecker
            logger.info("\n正在初始化敏感词检测器...")
            stats = SensitiveWordChecker.initialize()
            logger.info(f"敏感词检测器初始化成功: {stats}")
        except Exception as e:
            logger.error(f"敏感词检测器初始化失败: {str(e)}")
            logger.exception("详细错误:")
            return False
        
        # 获取自定义文本
        custom_text = self.get_custom_text_from_user()
        
        if custom_text:
            # 执行自定义文本检查
            await self.run_custom_text_check(custom_text)
        else:
            logger.info("\n未提供自定义文本，程序结束")
        
        return True

    def run_interactive_sync(self):
        """同步方式运行交互模式"""
        return asyncio.run(self.run_interactive_mode())


def main():
    """主函数"""
    print("=" * 80)
    print("敏感词检查功能单元测试（完整版本）")
    print("测试文件: test_sensitive_check_standalone.py")
    print("=" * 80)
    
    tester = TestSensitiveCheck()
    success = tester.run_interactive_sync()
    
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()