LQAgentPlatform/utils_test/standard_new_Test/test_app.py

"""
标准库匹配规则 - 实际调用测试
展示如何实际调用 StandardMatchingService 进行标准检查

使用说明:
1. MOCK_MODE = True:  使用模拟数据，无需数据库
2. MOCK_MODE = False: 使用真实数据库，需要确保 config/config.ini 配置正确
"""

import sys
import os

# ==================== 配置区域 ====================
# 设置为 True 使用 Mock 模式（无需数据库），False 使用真实数据库
MOCK_MODE = False

# ==================== 路径设置 ====================
CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_ROOT = os.path.dirname(os.path.dirname(CURRENT_DIR))
sys.path.insert(0, PROJECT_ROOT)

# 真实模式下需要修改工作目录以读取配置文件
if not MOCK_MODE:
    os.chdir(PROJECT_ROOT)

import asyncio
import json
from utils_test.standard_new_Test.standard_service import (
    StandardMatchingService,
    MatchResultCode
)


# ==================== 服务创建 ====================

async def create_service():
    """创建并初始化服务实例"""
    if MOCK_MODE:
        print("[Mock 模式] 使用模拟数据，无需数据库连接")
        service = StandardMatchingService(db_pool=None)
        await service.initialize()
        return service
    else:
        print("[真实模式] 连接数据库并加载标准数据...")
        # 延迟导入，避免 Mock 模式下也需要安装依赖
        from foundation.database.base.sql.async_mysql_conn_pool import AsyncMySQLPool
        db_pool = AsyncMySQLPool()
        await db_pool.initialize()

        service = StandardMatchingService(db_pool=db_pool)
        await service.initialize()
        return service


# ==================== 测试函数 ====================

async def test_single_standard(service: StandardMatchingService):
    """测试单个标准检查"""
    print("\n" + "=" * 60)
    print("【测试1】单个标准检查")
    print("=" * 60)

    result = service.check_single(
        seq_no=1,
        standard_name="铁路桥涵设计规范",
        standard_number="TB 10002-2017"
    )

    print(f"\n序号: {result.seq_no}")
    print(f"原始标准: 《{result.original_name}》({result.original_number})")
    print(f"处理结果: {result.process_result}")
    print(f"状态码: {result.status_code}")
    print(f"结果消息: {result.final_result}")
    if result.substitute_number:
        print(f"替代标准: 《{result.substitute_name}》({result.substitute_number})")


async def test_batch_check(service: StandardMatchingService):
    """测试批量标准检查 - 文档中的7个示例"""
    print("\n" + "=" * 60)
    print("【测试2】批量标准检查 - 文档中的7个示例")
    print("=" * 60)

    standards = [
        {"standard_name": "铁路桥涵设计规范", "standard_number": "TB 10002-2017"},
        {"standard_name": "铁路工程抗震设计规范", "standard_number": "GB 50111-2006"},
        {"standard_name": "铁路混凝土工程施工质量验收标准", "standard_number": "TB 10424-2018"},
        {"standard_name": "公路水运危险性较大工程专项施工方案编制审查规程", "standard_number": "JT/T 1495-2023"},
        {"standard_name": "起重机 钢丝绳 保养、维护、检验和报废", "standard_number": "GB/T 5972-2016"},
        {"standard_name": "缆索起重机", "standard_number": "GB/T 28756-2012"},
        {"standard_name": "电力高处作业防坠器", "standard_number": "DL/T 1147-2009"},


    #    {"standard_name": "铁路桥涵设计", "standard_number": "TB 10002-2017"},
    #    {"standard_name": "铁路桥涵设计规范", "standard_number": "TB 10002"},
    #    {"standard_name": "铁路桥涵设计", "standard_number": "TB 10002"},
    #    {"standard_name": "《铁路桥涵设计规范》", "standard_number": " （TB 10002-2017）"},
    #     {"standard_name": "《铁路桥涵设计规范》", "standard_number": "(TB 10002-2017)"},
    #    {"standard_name": " 铁路桥涵设计规范 ", "standard_number": "  TB 10002-2017 "},
    #    {"standard_name": "铁路桥涵 设计规范", "standard_number": "TB 10002  -2017"},
    #     {"standard_name": "铁路桥涵设计规范", "standard_number": "TB 10002  -2017"},
    #      {"standard_name": "铁路桥涵 设计规范", "standard_number": "TB 10002-2017"},
    ]

    results = service.check_standards(standards)

    print("\n检查结果汇总:")
    print("-" * 80)
    print(f"{'序号':<4} {'标准名称':<40} {'标准号':<20} {'状态':<12}")
    print("-" * 80)

    for r in results:
        print(f"{r.seq_no:<4} {r.original_name:<40} {r.original_number:<20} {r.status_code:<12}")

    print("\n" + "=" * 80)
    print("详细结果:")
    print("=" * 80)

    for r in results:
        print(f"\n【{r.seq_no}】{r.original_name}")
        print(f"    标准号: {r.original_number}")
        print(f"    状态: {r.process_result} ({r.status_code})")
        print(f"    结果: {r.final_result}")
        if r.substitute_number:
            print(f"    ↳ 替代: 《{r.substitute_name}》({r.substitute_number})")


async def test_various_cases(service: StandardMatchingService):
    """测试各种边界情况"""
    print("\n" + "=" * 60)
    print("【测试3】各种边界情况测试")
    print("=" * 60)

    test_cases = [
        {"name": "测试现行标准", "standard_name": "铁路桥涵设计规范", "standard_number": "TB 10002-2017"},
        {"name": "测试被替代", "standard_name": "起重机 钢丝绳 保养、维护、检验和报废", "standard_number": "GB/T 5972-2016"},
        {"name": "测试废止无替代", "standard_name": "缆索起重机", "standard_number": "GB/T 28756-2012"},
        {"name": "测试不匹配", "standard_name": "公路水运危险性较大工程专项施工方案编制审查规程", "standard_number": "JT/T 1495-2023"},
        {"name": "测试不存在", "standard_name": "不存在的标准", "standard_number": "GB/T 99999-9999"},
    ]

    for case in test_cases:
        result = service.check_single(
            seq_no=1,
            standard_name=case["standard_name"],
            standard_number=case["standard_number"]
        )
        print(f"\n--- {case['name']} ---")
        print(f"输入: 《{result.original_name}》({result.original_number})")
        print(f"输出: [{result.status_code}] {result.final_result}")


async def export_results_to_json(service: StandardMatchingService, filename: str = "standard_check_results.json"):
    """导出检查结果为JSON文件"""
    print("\n" + "=" * 60)
    print("【测试4】导出结果为JSON")
    print("=" * 60)

    standards = [
        {"standard_name": "铁路桥涵设计规范", "standard_number": "TB 10002-2017"},
        {"standard_name": "铁路混凝土工程施工质量验收标准", "standard_number": "TB 10424-2018"},
        {"standard_name": "公路水运危险性较大工程专项施工方案编制审查规程", "standard_number": "JT/T 1495-2023"},
        {"standard_name": "起重机 钢丝绳 保养、维护、检验和报废", "standard_number": "GB/T 5972-2016"},
    ]

    results = service.check_standards(standards)

    output = []
    for r in results:
        output.append({
            "seq_no": r.seq_no,
            "original_name": r.original_name,
            "original_number": r.original_number,
            "substitute_name": r.substitute_name,
            "substitute_number": r.substitute_number,
            "process_result": r.process_result,
            "status_code": r.status_code,
            "final_result": r.final_result
        })

    with open(filename, "w", encoding="utf-8") as f:
        json.dump(output, f, ensure_ascii=False, indent=2)

    print(f"\n结果已导出到: {filename}")
    print(json.dumps(output, ensure_ascii=False, indent=2))


# ==================== 主函数 ====================

async def main():
    """主函数"""
    print("=" * 80)
    print("标准库匹配规则 - 实际调用测试（内存处理版本）")
    print("=" * 80)
    print(f"\n当前模式: {'Mock 测试模式（无需数据库）' if MOCK_MODE else '真实数据库模式'}")
    print("-" * 80)

    service = None
    try:
        # 创建并初始化服务
        service = await create_service()
        print(f"服务初始化成功! 数据已加载到内存\n")

        # 执行测试
        await test_single_standard(service)
        await test_batch_check(service)
        await test_various_cases(service)
        # await export_results_to_json(service)  # 取消注释以测试JSON导出

        print("\n" + "=" * 80)
        print("所有测试执行完毕!")
        print("=" * 80)

    except Exception as e:
        print(f"\n错误: {e}")
        import traceback
        traceback.print_exc()


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