LQAgentPlatform/test_prep_basis_review.py

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

"""
测试prep_basis_review方法的简单脚本
"""

import asyncio
import sys
import os

# 添加项目根目录到sys.path
current_file = os.path.abspath(__file__)
root_dir = os.path.dirname(os.path.dirname(os.path.dirname(current_file)))
sys.path.append(root_dir)

from core.base.task_models import TaskFileInfo
from core.construction_review.component.ai_review_engine import AIReviewEngine

async def test_prep_basis_review():
    """测试编制依据审查方法"""
    print("🚀 开始测试 prep_basis_review 方法")

    # 模拟编制依据文本内容
    test_content = """
    (1)《架桥机通用技术条件》（GB/T 26470-2011）；
    (2)《起重机设计规范》（GB/T 3811-2008）；
    (3)《起重机械安全规程 第 5 部分：桥式和门式起重机》（GB 6067.5-2014）；
    (4)《电气装置安装工程 起重机电气装置施工及验收规范》（GB 50256-2014）；
    (5)《起重设备安装工程施工及验收规范》（GB50278-2010）；
    (6)《施工现场机械设备检查技术规范》（JGJ 160-2016）；
    (7)《公路桥涵施工技术规范》（JTG/T 3650-2020）；
    (8)《建设工程安全生产管理条例》《四川省安全生产条例》（2023）；
    (9)《重要用途钢丝绳》（GB 8918-2006）；
    (10)《起重机用钢丝绳》（GB T 34198-2017）；
    """

    # 创建模拟的TaskFileInfo
    file_info_dict = {
        'file_id': "test_file_001",
        'callback_task_id': "test_callback_001",
        'user_id': "test_user_001",
        'review_config': [],
        'project_plan_type': "construction_plan",
        'tendency_review_role': "safety_reviewer",
        'file_name': "test_document.doc",
        'file_type': "doc",
        'launched_at': 1234567890
    }
    task_info = TaskFileInfo(file_info_dict)

    # 创建AIReviewEngine实例
    ai_engine = AIReviewEngine(task_info, max_concurrent_reviews=4)

    # 准备审查数据
    review_data = {
        'content': test_content.strip(),
        'max_concurrent': 4
    }

    # 执行编制依据审查
    try:
        result = await ai_engine.prep_basis_review(
            review_data=review_data,
            trace_id="test_prep_basis_001",
            state={"test": True},
            stage_name="编制依据审查测试"
        )

        print("✅ 编制依据审查成功完成!")
        print(f"📊 审查结果统计:")
        print(f"   - 成功状态: {result.get('success', False)}")
        print(f"   - 执行时间: {result.get('execution_time', 0):.2f}秒")
        print(f"   - 总编制依据数: {result.get('total_basis_items', 0)}")
        print(f"   - 有效项目数: {result.get('valid_items', 0)}")
        print(f"   - 标准项目数: {result.get('standard_items', 0)}")

        # 显示部分审查结果详情
        review_results = result.get('review_results', [])
        if review_results:
            print(f"📋 审查结果详情 (前2个批次):")
            for i, batch in enumerate(review_results[:2]):
                print(f"   批次 {i+1}: {len(batch)} 条编制依据")
                for j, item in enumerate(batch[:3]):  # 每批次显示前3条
                    name = item.get('name', 'Unknown')[:30] + "..." if len(item.get('name', '')) > 30 else item.get('name', 'Unknown')
                    status = item.get('status', 'Unknown')
                    is_standard = item.get('is_standard', False)
                    print(f"     {j+1}. {name}: {status} ({'标准' if is_standard else '非标准'})")
        else:
            print("⚠️ 没有返回审查结果")

    except Exception as e:
        print(f"❌ 测试失败: {str(e)}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    asyncio.run(test_prep_basis_review())