# 标准库匹配规则系统 - 内存处理版本 基于 `standard_timeliness_new.md` 文档实现的标准库时效性审查系统,采用**内存数据处理**架构。 ## 架构设计 ``` ┌─────────────────────────────────────────────────────────────┐ │ StandardMatchingService │ │ (对外服务接口) │ └───────────────────────┬─────────────────────────────────────┘ │ ┌───────────────┴───────────────┐ │ │ ┌───────▼────────┐ ┌────────▼────────┐ │StandardRepository│ │ StandardMatcher │ │ (数据仓库) │ │ (匹配逻辑) │ └───────┬────────┘ └─────────────────┘ │ ┌───────▼────────┐ │ MySQL DAO │ │ (一次性加载) │ └────────────────┘ ``` ### 核心组件 | 组件 | 职责 | 说明 | |------|------|------| | `StandardRepository` | 内存数据存储和索引 | 加载数据后建立多索引加速查询 | | `StandardMatcher` | 匹配规则逻辑 | 纯内存匹配,无数据库访问 | | `StandardMatchingService` | 对外服务接口 | 统一入口,管理数据加载 | | `StandardDAO` | 数据库访问 | 只负责一次性加载所有数据 | ## 文件结构 ``` utils_test/standard_new_Test/ ├── standard_dao.py # 数据访问层(一次性加载) ├── standard_service.py # 核心业务逻辑(内存处理) │ ├── StandardRepository # 数据仓库类 │ ├── StandardMatcher # 匹配器类 │ └── StandardMatchingService # 服务类 ├── test_app.py # 实际调用测试 ├── test_standard_matching.py # 单元测试 └── README.md # 说明文档 ``` ## 标准库数据结构 | 字段 | 说明 | |------|------| | id | 序号 | | standard_number | 标准号 | | standard_name (chinese_name) | 标准名称 | | validity | 时效性 (XH-现行, SX-试行, FZ-废止) | ## 五种匹配结果状态 | 情况 | 条件 | process_result | status_code | final_result | |------|------|----------------|-------------|--------------| | 1 | 名称+标准号都匹配,现行/试行 | 状态正常 | OK | 无问题 | | 2 | 名称+标准号都匹配,废止,有现行替代 | 状态被替代 | SUBSTITUTED | 《标准》(号)已废止,替代《标准》(号) | | 3 | 名称+标准号都匹配,废止,无替代 | 状态废止无现行 | ABOLISHED | 《标准》(号)已废止,无现行状态 | | 4 | 名称或标准号只有一个匹配 | 不匹配 | MISMATCH | 《提取》(号)与实际《标准》(号)不匹配 | | 5 | 都不匹配 | 标准库不存在 | NOT_FOUND | 《标准》(号)标准库不存在,请确认 | ## 使用方法 ### 1. 初始化服务(一次性加载数据到内存) ```python from utils_test.standard_new_Test.standard_service import StandardMatchingService # Mock 模式(无需数据库) service = StandardMatchingService(db_pool=None) await service.initialize() # 真实数据库模式 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() # 从MySQL加载所有数据到内存 ``` ### 2. 单个标准检查 ```python result = service.check_single( seq_no=1, standard_name="铁路桥涵设计规范", standard_number="TB 10002-2017" ) print(f"状态: {result.status_code}") # OK print(f"结果: {result.final_result}") # 无问题 ``` ### 3. 批量检查 ```python standards = [ {"standard_name": "铁路桥涵设计规范", "standard_number": "TB 10002-2017"}, {"standard_name": "缆索起重机", "standard_number": "GB/T 28756-2012"}, ] results = service.check_standards(standards) for r in results: print(f"{r.seq_no}. {r.original_name} ({r.original_number})") print(f" 状态: {r.status_code}") print(f" 结果: {r.final_result}") if r.substitute_number: print(f" 替代: {r.substitute_name} ({r.substitute_number})") ``` ## 运行测试 ```bash # 运行实际调用测试 cd utils_test/standard_new_Test python test_app.py # 运行单元测试 python -m pytest test_standard_matching.py -v ``` ## 代码设计特点 ### 1. 内存数据处理 - 服务初始化时一次性从MySQL加载所有数据到内存 - 后续匹配操作完全在内存中进行,无数据库访问 - 适合标准库数据量不大但需要高频查询的场景 ### 2. 可维护性优化 - **分层架构**:Repository(数据)+ Matcher(逻辑)+ Service(接口) - **单一职责**:每个方法只做一件事,如 `_set_ok_result()`、`_handle_abolished()` - **索引优化**:标准号、名称多索引加速内存查询 ### 3. 匹配规则实现 匹配流程拆分为多个独立方法: ``` match() ├── _handle_number_matched() # 标准号匹配 │ ├── _handle_full_match() # 名称也匹配 │ │ ├── _set_ok_result() # 现行/试行 │ │ └── _handle_abolished() # 废止 │ └── _handle_name_mismatch() # 名称不匹配 └── _handle_number_not_matched() # 标准号不匹配 ├── _check_name_in_records() └── _search_by_name_only() ``` ## 数据表结构 ```sql CREATE TABLE t_samp_standard_base_info ( id INT PRIMARY KEY AUTO_INCREMENT, chinese_name VARCHAR(500) COMMENT '标准名称', standard_number VARCHAR(100) COMMENT '标准编号', validity VARCHAR(10) COMMENT '时效性 (XH-现行, SX-试行, FZ-废止)', INDEX idx_number (standard_number), INDEX idx_name (chinese_name) ) COMMENT='标准库基础信息表'; ```