""" 标准库匹配规则 - 实际调用测试 展示如何实际调用 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())