LQAgentPlatform/utils_test/standard_new_Test

标准库匹配规则系统 - 内存处理版本

基于 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. 初始化服务（一次性加载数据到内存）

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. 单个标准检查

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. 批量检查

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})")

运行测试

# 运行实际调用测试
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()

数据表结构

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='标准库基础信息表';