LQAgentPlatform/utils_test/standard_new_Test/test_app.py

"""
标准库匹配规则 - 实际调用测试
展示如何实际调用标准匹配逻辑进行标准检查
直接使用生产代码 core.construction_review.component.standard_matching 模块

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

import sys
import os

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

# ==================== 路径设置 ====================
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 conftest import (
    StandardRepository,
    StandardMatcher,
    MatchResultCode,
)

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


class MockMatchingService:
    """模拟 StandardMatchingService 接口，使用生产代码的 Repository + Matcher"""

    def __init__(self, mock_data):
        self.repository = StandardRepository()
        self.repository.load_data(mock_data)
        self.matcher = StandardMatcher(self.repository)

    def check_single(self, seq_no, standard_name, standard_number):
        return self.matcher.match(seq_no, standard_name, standard_number)

    def check_standards(self, standards):
        results = []
        for idx, std in enumerate(standards, start=1):
            result = self.matcher.match(
                seq_no=idx,
                input_name=std.get("standard_name", ""),
                input_number=std.get("standard_number", "")
            )
            if result is not None:
                results.append(result)
        return results


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

async def create_service():
    """创建并初始化服务实例"""
    if MOCK_MODE:
        print("[Mock 模式] 使用模拟数据，无需数据库连接")
        return MockMatchingService(MOCK_DATA)
    else:
        print("[真实模式] 连接数据库并加载标准数据...")
        from foundation.database.base.sql.async_mysql_conn_pool import AsyncMySQLPool
        db_pool = AsyncMySQLPool()
        await db_pool.initialize()

        from core.construction_review.component.standard_matching import StandardMatchingService
        service = StandardMatchingService(db_pool=db_pool)
        await service.initialize()
        return service


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

async def test_single_standard(service):
    """测试单个标准检查"""
    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.raw_name}》({result.raw_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):
    """测试批量标准检查 - 文档中的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"},
    ]

    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.raw_name:<40} {r.raw_number:<20} {r.status_code:<12}")

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

    for r in results:
        print(f"\n【{r.seq_no}】{r.raw_name}")
        print(f"    标准号: {r.raw_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):
    """测试各种边界情况"""
    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.raw_name}》({result.raw_number})")
        print(f"输出: [{result.status_code}] {result.final_result}")


async def export_results_to_json(service, filename="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,
            "raw_name": r.raw_name,
            "raw_number": r.raw_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())