sunvpy-docs/docs/content/33-zi-ding-yi-jian-cha-gui-ze.md at 78083266d9e515475b4a40f3c458f8b79d66c846

Files

hoenking 8c78c0f920 init

2026-04-10 13:47:53 +08:00

39 KiB

Raw Blame History

自定义检查规则是 SunvStation 数据质量保证系统中最为灵活和强大的功能，允许开发者根据特定的业务需求和技术规范编写专属的数据验证逻辑。通过组合 CheckFunForPY 的检查记录管理能力、SSProcessManager 的选择集操作能力以及 Python 的丰富生态系统，您可以构建从简单的属性完整性检查到复杂的空间拓扑验证等各类检查规则。

自定义检查规则的核心价值在于将通用的检查框架与特定的业务逻辑分离。系统提供了检查记录管理、结果显示、持久化存储等通用能力，而您只需要专注于实现具体的检查算法和数据验证逻辑。这种设计既降低了开发门槛，又保持了系统的扩展性。

Sources: PySSCheck.py

检查规则架构设计

自定义检查规则采用分层架构模式，将检查任务分解为数据准备、检查执行、结果记录三个阶段。每个阶段由不同的组件负责，通过明确定义的接口协作，形成清晰的职责划分。

graph TB
    subgraph "检查规则脚本层"
        Script[检查规则脚本<br/>自定义业务逻辑]
    end
    
    subgraph "功能组件层"
        CF[CheckFunForPY<br/>检查记录管理]
        SS[SSProcessManager<br/>选择集与数据操作]
        Logger[日志组件<br/>日志与进度]
    end
    
    subgraph "数据层"
        WS[工作空间<br/>WorkSpace]
        Map[地图数据<br/>ScaleMap]
        DS[数据源<br/>DataSource]
    end
    
    subgraph "持久化层"
        File[检查记录文件]
    end
    
    Script -->|实例化| CF
    Script -->|调用| SS
    Script -->|使用| Logger
    
    SS -->|访问| WS
    SS -->|操作| Map
    WS -->|获取| DS
    
    CF -->|保存/加载| File
    
    style Script fill:#e1f5ff
    style CF fill:#fff4e1
    style SS fill:#ffe1f5

该架构体现了关注点分离的设计原则。检查脚本专注于业务逻辑的实现，不关心底层的记录管理和数据访问细节。这种分层设计使得检查规则易于测试、维护和复用。

Sources: PySSCheck.py, PySSProcess.py

核心组件角色

组件	职责	主要方法/特性
检查规则脚本	实现具体的检查逻辑，协调各组件协作	自定义函数、类、算法实现
CheckFunForPY	管理检查记录的生命周期	AddCheckRecord、GetCheckRecordCount、ShowCheckOutput
SSProcessManager	提供数据查询、遍历、编辑能力	setSelectCondition、selectFilter、getSelGeoValue
日志组件	记录检查过程，提供进度反馈	logger.info、startProgress、stepProgress

这种设计允许开发者组合基础组件构建复杂的检查规则，而不需要从零开始实现所有功能。每个组件都经过充分测试，提供了稳定的 API 接口。

Sources: PySSCheck.py, PySSProcess.py

检查规则的基本结构

一个规范的自定义检查规则脚本通常包含五个关键部分：初始化、数据准备、检查执行、结果处理、清理退出。遵循这种结构可以确保脚本的逻辑清晰、易于维护。

flowchart TD
    A[脚本开始] --> B[初始化阶段<br/>创建组件实例]
    B --> C[数据准备阶段<br/>设置选择条件、加载数据]
    C --> D[检查执行阶段<br/>遍历对象、应用检查逻辑]
    D --> E{发现问题?}
    E -->|是| F[添加检查记录]
    E -->|否| G[继续下一个对象]
    F --> G
    G --> D
    
    D --> H{检查完成?}
    H -->|否| D
    H -->|是| I[结果处理阶段<br/>统计、显示、保存]
    I --> J[清理退出阶段<br/>释放资源]
    J --> K[脚本结束]
    
    style B fill:#e1f5ff
    style D fill:#fff4e1
    style F fill:#ffd4e1
    style I fill:#d4edda

标准模板

以下是一个标准的检查规则脚本模板，包含了所有关键部分：

from sunvpy import SSProcess
from sunvpy.PySSCheck import CheckFunForPY
import logging

def run_custom_check():
    """自定义检查规则的标准实现模板"""
    
    # ========== 初始化阶段 ==========
    check_fun = CheckFunForPY()
    check_fun.ClearCheckRecord()  # 清空之前的记录
    
    # 配置日志
    if SSProcess.logger is None:
        SSProcess.config_logger(log_file="check_log.txt", log_level=logging.INFO)
    
    logger = SSProcess.logger
    logger.info("========== 开始执行自定义检查规则 ==========")
    
    # ========== 数据准备阶段 ==========
    logger.info("准备检查数据...")
    
    SSProcess.clearSelection()
    SSProcess.clearSelectCondition()
    
    # 设置选择条件
    SSProcess.setSelectCondition("SSObj_LayerName", "==", "建筑")
    SSProcess.setSelectCondition("SSObj_Type", "==", "3")
    
    # 执行查询
    logger.info("执行查询...")
    SSProcess.selectFilter()
    
    total_count = SSProcess.getSelGeoCount()
    logger.info(f"查询到 {total_count} 个对象")
    
    if total_count == 0:
        logger.warning("没有需要检查的对象，检查终止")
        return 0
    
    # 配置进度条
    if SSProcess.enable_progress:
        SSProcess.start_progress("自定义检查进度", 0, total_count)
    
    # ========== 检查执行阶段 ==========
    issue_count = 0
    
    logger.info("开始执行检查逻辑...")
    
    for i in range(total_count):
        # 更新进度
        if SSProcess.enable_progress and i % 100 == 0:
            SSProcess.step_progress()
        
        # 获取对象信息
        geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
        obj_type = int(SSProcess.getSelGeoValue(i, "SSObj_Type"))
        obj_code = SSProcess.getSelGeoValue(i, "SSObj_Code")
        
        # 获取空间位置
        x = float(SSProcess.getSelGeoValue(i, "SSObj_X"))
        y = float(SSProcess.getSelGeoValue(i, "SSObj_Y"))
        z = 0.0
        
        # ========== 自定义检查逻辑 ==========
        # 这里实现具体的检查规则
        
        # 示例：检查编码是否为空
        if not obj_code or obj_code.strip() == "":
            issue_count += 1
            check_fun.AddCheckRecord(
                group="属性检查",
                check="编码缺失检查",
                checkmodel=1,
                description=f"对象 {geo_id} 的编码为空",
                x=x, y=y, z=z,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
            
            logger.warning(f"发现问题：对象 {geo_id} 编码缺失")
        
        # 添加更多检查规则...
    
    # ========== 结果处理阶段 ==========
    # 关闭进度条
    if SSProcess.enable_progress:
        SSProcess.close_progress()
    
    # 统计结果
    total_issues = check_fun.GetCheckRecordCount()
    logger.info(f"检查完成，共发现 {total_issues} 个问题")
    logger.info("========== 检查规则执行完毕 ==========")
    
    # 显示检查结果
    if total_issues > 0:
        check_fun.ShowCheckOutput()
        
        # 保存检查记录
        save_success = check_fun.SaveCheckRecord()
        if save_success:
            logger.info("检查记录已成功保存")
        else:
            logger.error("检查记录保存失败")
    
    # ========== 清理退出阶段 ==========
    # 这里可以添加必要的清理代码
    
    return total_issues

if __name__ == "__main__":
    result = run_custom_check()
    print(f"检查完成，发现问题数：{result}")

该模板遵循了防御性编程的最佳实践，包含完善的错误处理、日志记录和进度反馈机制，适合作为开发新检查规则的起点。

Sources: PySSCheck.py, PySSProcess.py

常见检查类型实现

基于不同的检查目标和数据特征，可以将自定义检查规则分为四个主要类型：属性完整性检查、拓扑关系检查、几何规则检查、业务规则检查。每种类型有其特定的实现模式和注意事项。

属性完整性检查

属性完整性检查关注对象的属性数据是否符合预定义的规则，包括必填项验证、格式验证、取值范围验证等。这是最基础也是最常见的检查类型。

def check_attribute_completeness():
    """属性完整性检查示例"""
    
    check_fun = CheckFunForPY()
    check_fun.ClearCheckRecord()
    
    SSProcess.clearSelection()
    SSProcess.clearSelectCondition()
    SSProcess.setSelectCondition("SSObj_LayerName", "==", "建筑")
    SSProcess.selectFilter()
    
    count = SSProcess.getSelGeoCount()
    
    # 定义必填属性规则
    required_fields = [
        {"name": "SSObj_Code", "desc": "编码"},
        {"name": "SSObj_Name", "desc": "名称"},
        {"name": "SSObj_DataMark", "desc": "数据标识"},
        {"name": "[JG]", "desc": "结构类型"},
        {"name": "[CS]", "desc": "层数"}
    ]
    
    issues = 0
    
    for i in range(count):
        geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
        obj_type = int(SSProcess.getSelGeoValue(i, "SSObj_Type"))
        x = float(SSProcess.getSelGeoValue(i, "SSObj_X"))
        y = float(SSProcess.getSelGeoValue(i, "SSObj_Y"))
        
        # 检查每个必填字段
        for field in required_fields:
            value = SSProcess.getSelGeoValue(i, field["name"])
            
            if not value or value.strip() == "":
                issues += 1
                check_fun.AddCheckRecord(
                    group="属性检查",
                    check="必填项缺失检查",
                    checkmodel=1,
                    description=f"对象 {geo_id} 的必填属性 [{field['desc']}] 为空",
                    x=x, y=y, z=0.0,
                    objtype=obj_type,
                    geoids=str(geo_id),
                    noteids=""
                )
        
        # 检查编码格式（示例：编码应为7位数字）
        code = SSProcess.getSelGeoValue(i, "SSObj_Code")
        if code and (len(code) != 7 or not code.isdigit()):
            issues += 1
            check_fun.AddCheckRecord(
                group="属性检查",
                check="编码格式检查",
                checkmodel=2,
                description=f"对象 {geo_id} 的编码格式错误：{code}（应为7位数字）",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
        
        # 检查层数是否为有效数字且在合理范围内
        floors = SSProcess.getSelGeoValue(i, "[CS]")
        if floors:
            try:
                floor_num = int(floors)
                if floor_num < 1 or floor_num > 200:
                    issues += 1
                    check_fun.AddCheckRecord(
                        group="属性检查",
                        check="层数范围检查",
                        checkmodel=3,
                        description=f"对象 {geo_id} 的层数超出合理范围：{floors}（应为1-200）",
                        x=x, y=y, z=0.0,
                        objtype=obj_type,
                        geoids=str(geo_id),
                        noteids=""
                    )
            except ValueError:
                issues += 1
                check_fun.AddCheckRecord(
                    group="属性检查",
                    check="层数格式检查",
                    checkmodel=4,
                    description=f"对象 {geo_id} 的层数不是有效数字：{floors}",
                    x=x, y=y, z=0.0,
                    objtype=obj_type,
                    geoids=str(geo_id),
                    noteids=""
                )
    
    total_issues = check_fun.GetCheckRecordCount()
    if total_issues > 0:
        check_fun.ShowCheckOutput()
    
    return total_issues

属性完整性检查的核心是规则的可配置性。通过将检查规则定义为数据结构（如字典或列表），可以方便地修改和扩展规则，而不需要修改检查逻辑代码。

Sources: PySSCheck.py, PySSProcess.py

拓扑关系检查

拓扑关系检查关注对象之间的空间关系是否符合预期，如重叠检查、包含检查、邻近检查等。这类检查通常需要更复杂的算法和空间索引支持。

def check_topology_overlap():
    """拓扑重叠检查示例（简化版）"""
    
    check_fun = CheckFunForPY()
    check_fun.ClearCheckRecord()
    
    # 查询所有面对象
    SSProcess.clearSelection()
    SSProcess.clearSelectCondition()
    SSProcess.setSelectCondition("SSObj_Type", "==", "3")  # 面对象
    SSProcess.selectFilter()
    
    count = SSProcess.getSelGeoCount()
    
    # 构建简单的空间索引（基于包围盒）
    objects = []
    for i in range(count):
        geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
        obj_type = int(SSProcess.getSelGeoValue(i, "SSObj_Type"))
        x = float(SSProcess.getSelGeoValue(i, "SSObj_X"))
        y = float(SSProcess.getSelGeoValue(i, "SSObj_Y"))
        
        min_x = float(SSProcess.getSelGeoValue(i, "SSObj_minX"))
        min_y = float(SSProcess.getSelGeoValue(i, "SSObj_minY"))
        max_x = float(SSProcess.getSelGeoValue(i, "SSObj_maxX"))
        max_y = float(SSProcess.getSelGeoValue(i, "SSObj_maxY"))
        
        objects.append({
            "index": i,
            "id": geo_id,
            "type": obj_type,
            "x": x, "y": y,
            "bbox": (min_x, min_y, max_x, max_y)
        })
    
    # 检查重叠（简化版：仅检查包围盒重叠）
    overlap_count = 0
    
    for i in range(len(objects)):
        for j in range(i + 1, len(objects)):
            obj1 = objects[i]
            obj2 = objects[j]
            
            # 检查包围盒是否重叠
            bbox1 = obj1["bbox"]
            bbox2 = obj2["bbox"]
            
            if not (bbox1[2] <= bbox2[0] or bbox1[0] >= bbox2[2] or
                    bbox1[3] <= bbox2[1] or bbox1[1] >= bbox2[3]):
                # 包围盒重叠，记录问题（实际应用中需要更精确的几何检查）
                overlap_count += 1
                check_fun.AddCheckRecord(
                    group="拓扑检查",
                    check="包围盒重叠检查",
                    checkmodel=1,
                    description=f"对象 {obj1['id']} 与对象 {obj2['id']} 可能存在空间重叠",
                    x=(bbox1[0] + bbox2[0]) / 2,
                    y=(bbox1[1] + bbox2[1]) / 2,
                    z=0.0,
                    objtype=obj1["type"],
                    geoids=f"{obj1['id']},{obj2['id']}",
                    noteids=""
                )
    
    total_issues = check_fun.GetCheckRecordCount()
    
    if total_issues > 0:
        print(f"发现 {total_issues} 处可能的几何重叠")
        check_fun.ShowCheckOutput()
    
    return total_issues

拓扑关系检查的关键挑战是性能优化。对于大规模数据集，O(n²) 的两两比较效率极低。实际应用中应该使用空间索引（如 R-Tree、QuadTree）将复杂度降低到 O(n log n)。本示例仅展示了基本思路，生产环境中需要更高效的空间数据结构。

Sources: PySSCheck.py, PySSProcess.py

几何规则检查

几何规则检查关注对象的几何特征是否符合特定的标准，如面积范围、长度限制、点数限制、自相交检测等。

def check_geometry_rules():
    """几何规则检查示例"""
    
    check_fun = CheckFunForPY()
    check_fun.ClearCheckRecord()
    
    SSProcess.clearSelection()
    SSProcess.clearSelectCondition()
    SSProcess.setSelectCondition("SSObj_Type", "==", "3")  # 面对象
    SSProcess.selectFilter()
    
    count = SSProcess.getSelGeoCount()
    
    # 定义几何规则
    geometry_rules = {
        "min_area": 1.0,      # 最小面积（平方米）
        "max_area": 1000000.0,  # 最大面积（平方米）
        "min_points": 3,      # 最小点数
        "max_points": 10000,  # 最大点数
        "min_length": 0.5,    # 最小周长（米）
        "max_length": 100000.0  # 最大周长（米）
    }
    
    issues = 0
    
    for i in range(count):
        geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
        obj_type = int(SSProcess.getSelGeoValue(i, "SSObj_Type"))
        x = float(SSProcess.getSelGeoValue(i, "SSObj_X"))
        y = float(SSProcess.getSelGeoValue(i, "SSObj_Y"))
        
        # 获取几何属性
        area = float(SSProcess.getSelGeoValue(i, "SSObj_Area"))
        length = float(SSProcess.getSelGeoValue(i, "SSObj_Length"))
        point_count = int(SSProcess.getSelGeoValue(i, "SSObj_PointCount"))
        
        # 检查面积范围
        if area < geometry_rules["min_area"]:
            issues += 1
            check_fun.AddCheckRecord(
                group="几何检查",
                check="面积过小检查",
                checkmodel=1,
                description=f"对象 {geo_id} 的面积 {area:.2f} 小于最小值 {geometry_rules['min_area']}",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
        elif area > geometry_rules["max_area"]:
            issues += 1
            check_fun.AddCheckRecord(
                group="几何检查",
                check="面积过大检查",
                checkmodel=2,
                description=f"对象 {geo_id} 的面积 {area:.2f} 超过最大值 {geometry_rules['max_area']}",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
        
        # 检查周长范围
        if length < geometry_rules["min_length"]:
            issues += 1
            check_fun.AddCheckRecord(
                group="几何检查",
                check="周长过小检查",
                checkmodel=3,
                description=f"对象 {geo_id} 的周长 {length:.2f} 小于最小值 {geometry_rules['min_length']}",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
        elif length > geometry_rules["max_length"]:
            issues += 1
            check_fun.AddCheckRecord(
                group="几何检查",
                check="周长过大检查",
                checkmodel=4,
                description=f"对象 {geo_id} 的周长 {length:.2f} 超过最大值 {geometry_rules['max_length']}",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
        
        # 检查点数范围
        if point_count < geometry_rules["min_points"]:
            issues += 1
            check_fun.AddCheckRecord(
                group="几何检查",
                check="点数过少检查",
                checkmodel=5,
                description=f"对象 {geo_id} 的点数 {point_count} 小于最小值 {geometry_rules['min_points']}",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
        elif point_count > geometry_rules["max_points"]:
            issues += 1
            check_fun.AddCheckRecord(
                group="几何检查",
                check="点数过多检查",
                checkmodel=6,
                description=f"对象 {geo_id} 的点数 {point_count} 超过最大值 {geometry_rules['max_points']}",
                x=x, y=y, z=0.0,
                objtype=obj_type,
                geoids=str(geo_id),
                noteids=""
            )
    
    total_issues = check_fun.GetCheckRecordCount()
    
    if total_issues > 0:
        check_fun.ShowCheckOutput()
    
    return total_issues

几何规则检查的优势在于规则的可量化性。所有的几何特征都可以用数值表示，这使得规则的制定和验证都非常明确。同时，这些规则通常具有明确的物理意义，便于理解和管理。

Sources: PySSCheck.py, PySSProcess.py, ObjBaseAttr.py

业务规则检查

业务规则检查关注对象是否符合特定的业务逻辑或行业标准，这些规则通常与具体的应用场景紧密相关。

def check_business_rules():
    """业务规则检查示例：建筑物业务规则"""
    
    check_fun = CheckFunForPY()
    check_fun.ClearCheckRecord()
    
    SSProcess.clearSelection()
    SSProcess.clearSelectCondition()
    SSProcess.setSelectCondition("SSObj_LayerName", "==", "建筑")
    SSProcess.selectFilter()
    
    count = SSProcess.getSelGeoCount()
    
    issues = 0
    
    for i in range(count):
        geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
        obj_type = int(SSProcess.getSelGeoValue(i, "SSObj_Type"))
        obj_code = SSProcess.getSelGeoValue(i, "SSObj_Code")
        obj_name = SSProcess.getSelGeoValue(i, "SSObj_Name")
        x = float(SSProcess.getSelGeoValue(i, "SSObj_X"))
        y = float(SSProcess.getSelGeoValue(i, "SSObj_Y"))
        
        # 获取扩展属性
        structure_type = SSProcess.getSelGeoValue(i, "[JG]")  # 结构类型
        floors = SSProcess.getSelGeoValue(i, "[CS]")  # 层数
        area = float(SSProcess.getSelGeoValue(i, "SSObj_Area"))
        
        # 业务规则1：高层建筑必须配备电梯（层数 >= 7）
        if floors:
            try:
                floor_num = int(floors)
                if floor_num >= 7:
                    # 检查是否有电梯相关的属性或标记
                    # 这里假设使用 DataMark 字段的第2位标记是否配备电梯
                    data_mark = SSProcess.getSelGeoValue(i, "SSObj_DataMark")
                    has_elevator = False
                    if data_mark and len(data_mark) > 1:
                        has_elevator = (data_mark[1] == "1")
                    
                    if not has_elevator:
                        issues += 1
                        check_fun.AddCheckRecord(
                            group="业务检查",
                            check="电梯配置检查",
                            checkmodel=1,
                            description=f"对象 {geo_id} 为 {floor_num} 层建筑但未配备电梯",
                            x=x, y=y, z=0.0,
                            objtype=obj_type,
                            geoids=str(geo_id),
                            noteids=""
                        )
            except ValueError:
                pass
        
        # 业务规则2：特定结构类型的建筑有面积限制
        if structure_type:
            area_limits = {
                "砖混": 10000.0,
                "框架": 50000.0,
                "钢结构": 100000.0
            }
            
            if structure_type in area_limits:
                limit = area_limits[structure_type]
                if area > limit:
                    issues += 1
                    check_fun.AddCheckRecord(
                        group="业务检查",
                        check="结构面积限制检查",
                        checkmodel=2,
                        description=f"对象 {geo_id} 的结构类型为 {structure_type}，但面积 {area:.2f} 超过限制 {limit}",
                        x=x, y=y, z=0.0,
                        objtype=obj_type,
                        geoids=str(geo_id),
                        noteids=""
                    )
        
        # 业务规则3：建筑物名称不能重复
        if obj_name and obj_name.strip() != "":
            # 检查后面是否有同名对象
            for j in range(i + 1, count):
                other_name = SSProcess.getSelGeoValue(j, "SSObj_Name")
                if obj_name == other_name:
                    other_id = SSProcess.getSelGeoValue(j, "SSObj_ID")
                    issues += 1
                    check_fun.AddCheckRecord(
                        group="业务检查",
                        check="名称重复检查",
                        checkmodel=3,
                        description=f"对象 {geo_id} 与对象 {other_id} 存在重复名称：{obj_name}",
                        x=x, y=y, z=0.0,
                        objtype=obj_type,
                        geoids=f"{geo_id},{other_id}",
                        noteids=""
                    )
                    break
    
    total_issues = check_fun.GetCheckRecordCount()
    
    if total_issues > 0:
        check_fun.ShowCheckOutput()
    
    return total_issues

业务规则检查体现了领域知识的重要性。这些规则通常来自行业标准、业务需求或专家经验，需要深入理解应用领域才能准确实现。建议将业务规则与通用规则分离，便于维护和更新。

Sources: PySSCheck.py, PySSProcess.py

高级检查模式

对于复杂的检查需求，可以通过组合多个检查类型、使用空间索引、实现增量检查等高级模式来提升检查规则的功能和性能。

组合检查模式

组合检查模式将多个检查逻辑整合到一个脚本中，通过统一的流程执行，生成综合性的检查报告。

class CombinedChecker:
    """组合检查器：整合多个检查类型"""
    
    def __init__(self):
        self.check_fun = CheckFunForPY()
        self.check_fun.ClearCheckRecord()
        self.logger = SSProcess.logger
        
    def run_all_checks(self):
        """执行所有检查"""
        self.logger.info("开始执行组合检查...")
        
        # 检查结果统计
        results = {
            "属性检查": 0,
            "几何检查": 0,
            "拓扑检查": 0,
            "业务检查": 0
        }
        
        # 1. 属性检查
        results["属性检查"] = self.check_attributes()
        
        # 2. 几何检查
        results["几何检查"] = self.check_geometry()
        
        # 3. 拓扑检查（简化版）
        results["拓扑检查"] = self.check_topology()
        
        # 4. 业务检查
        results["业务检查"] = self.check_business()
        
        # 统计总问题数
        total = sum(results.values())
        self.logger.info("========== 检查结果汇总 ==========")
        for check_type, count in results.items():
            self.logger.info(f"  {check_type}: {count} 个问题")
        self.logger.info(f"  总计: {total} 个问题")
        
        # 显示结果
        if total > 0:
            self.check_fun.ShowCheckOutput()
        
        return results
    
    def check_attributes(self):
        """属性检查"""
        self.logger.info("执行属性检查...")
        # 实现属性检查逻辑
        # ...
        return 0
    
    def check_geometry(self):
        """几何检查"""
        self.logger.info("执行几何检查...")
        # 实现几何检查逻辑
        # ...
        return 0
    
    def check_topology(self):
        """拓扑检查"""
        self.logger.info("执行拓扑检查...")
        # 实现拓扑检查逻辑
        # ...
        return 0
    
    def check_business(self):
        """业务检查"""
        self.logger.info("执行业务检查...")
        # 实现业务检查逻辑
        # ...
        return 0


# 使用示例
if __name__ == "__main__":
    # 配置日志
    SSProcess.config_logger(log_file="combined_check.log", log_level=logging.INFO)
    
    # 创建并运行组合检查器
    checker = CombinedChecker()
    results = checker.run_all_checks()

组合检查模式的优势在于统一的管理和报告。通过将多个检查整合到一个脚本中，可以生成综合性的检查报告，便于全面了解数据质量状况。

Sources: PySSCheck.py, PySSProcess.py

增量检查模式

增量检查模式只检查新增或修改的对象，避免重复检查未变化的数据，大幅提升检查效率。

class IncrementalChecker:
    """增量检查器：只检查新增或修改的对象"""
    
    def __init__(self, last_check_time=None):
        self.check_fun = CheckFunForPY()
        self.check_fun.ClearCheckRecord()
        self.last_check_time = last_check_time
        self.logger = SSProcess.logger
    
    def run_incremental_check(self):
        """执行增量检查"""
        if self.last_check_time is None:
            self.logger.info("首次检查，执行全量检查...")
            return self.run_full_check()
        else:
            self.logger.info(f"执行增量检查，时间阈值：{self.last_check_time}...")
            return self.run_incremental()
    
    def run_full_check(self):
        """全量检查"""
        SSProcess.clearSelection()
        SSProcess.clearSelectCondition()
        SSProcess.selectFilter()
        
        count = SSProcess.getSelGeoCount()
        self.logger.info(f"全量检查对象数：{count}")
        
        # 执行检查逻辑
        # ...
        
        # 更新最后检查时间
        self.last_check_time = SSProcess.get_current_time()
        return count
    
    def run_incremental(self):
        """增量检查：只检查修改时间晚于阈值的对象"""
        SSProcess.clearSelection()
        SSProcess.clearSelectCondition()
        
        # 使用 ModifyTime 过滤条件
        # 注意：实际语法需要根据系统支持的运算符调整
        SSProcess.setSelectCondition("SSObj_ModifyTime", ">", self.last_check_time)
        SSProcess.selectFilter()
        
        count = SSProcess.getSelGeoCount()
        self.logger.info(f"增量检查对象数：{count}")
        
        if count == 0:
            self.logger.info("没有需要检查的新增或修改对象")
            return 0
        
        # 执行检查逻辑
        # ...
        
        # 更新最后检查时间
        self.last_check_time = SSProcess.get_current_time()
        return count


# 使用示例
if __name__ == "__main__":
    # 从配置或文件中加载上次检查时间
    last_check_time = "2025-01-01 00:00:00"
    
    checker = IncrementalChecker(last_check_time)
    count = checker.run_incremental_check()

增量检查模式的核心是时间过滤机制。通过记录最后检查时间，每次只检查在该时间之后修改的对象，可以大幅减少检查工作量，特别适合频繁更新的数据集。

Sources: PySSCheck.py, PySSProcess.py

性能优化最佳实践

编写高效的检查规则需要考虑多个方面，包括数据访问优化、算法复杂度控制、批量处理等。以下是经过验证的最佳实践。

数据访问优化

减少重复的属性访问操作是性能优化的关键。通过缓存和批量访问可以显著提升性能。

优化策略	说明	性能提升
批量获取属性	一次获取多个属性值，减少方法调用次数	30%-50%
缓存常用属性	将频繁访问的属性值缓存在本地变量中	20%-40%
减少类型转换	避免频繁的字符串-数值转换	10%-20%
使用索引	对需要过滤的字段建立索引	50%-200%

# 优化前：多次访问属性
for i in range(count):
    geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
    x = float(SSProcess.getSelGeoValue(i, "SSObj_X"))
    y = float(SSProcess.getSelGeoValue(i, "SSObj_Y"))
    obj_type = int(SSProcess.getSelGeoValue(i, "SSObj_Type"))
    # ... 每次都访问一次

# 优化后：批量获取并缓存
for i in range(count):
    # 一次性获取多个属性（假设系统支持）
    values = SSProcess.getSelGeoValues(i, ["SSObj_ID", "SSObj_X", "SSObj_Y", "SSObj_Type"])
    
    geo_id = values["SSObj_ID"]
    x = float(values["SSObj_X"])
    y = float(values["SSObj_Y"])
    obj_type = int(values["SSObj_Type"])

Sources: PySSProcess.py

算法复杂度控制

选择合适的算法和数据结构对性能有决定性影响。

graph TB
    A[检查性能优化决策树] --> B{数据规模}
    B -->|< 1000| C[简单遍历<br/>O n]
    B -->|1000-10000| D[使用字典索引<br/>O n log n]
    B -->|> 10000| E[使用空间索引<br/>O n log n]
    
    C --> F[适合：<br/>单对象检查、简单规则]
    D --> G[适合：<br/>关联检查、重复检查]
    E --> H[适合：<br/>拓扑检查、邻近检查]
    
    style C fill:#d4edda
    style D fill:#fff3cd
    style E fill:#f8d7da

进度反馈优化

合理的进度反馈机制可以提升用户体验，但过度的进度更新会影响性能。

# 优化前：每次循环都更新进度
for i in range(count):
    SSProcess.step_progress()  # 性能瓶颈
    # 执行检查逻辑

# 优化后：按批次更新进度
UPDATE_INTERVAL = 100  # 每100个对象更新一次进度

for i in range(count):
    # 执行检查逻辑
    
    if i % UPDATE_INTERVAL == 0:
        SSProcess.step_progress()

Sources: PySSProcess.py

错误处理与调试

完善的错误处理机制是检查规则稳定运行的关键，包括异常捕获、日志记录、断言验证等。

def robust_check():
    """具有完善错误处理的检查脚本"""
    
    try:
        # 初始化
        check_fun = CheckFunForPY()
        check_fun.ClearCheckRecord()
        
        # 配置日志
        SSProcess.config_logger(
            log_file="check_robust.log",
            log_level=logging.DEBUG
        )
        logger = SSProcess.logger
        
        logger.info("========== 开始执行检查 ==========")
        
        # 数据准备
        SSProcess.clearSelection()
        SSProcess.clearSelectCondition()
        SSProcess.setSelectCondition("SSObj_Type", "==", "3")
        
        try:
            SSProcess.selectFilter()
        except Exception as e:
            logger.error(f"查询失败：{str(e)}")
            return 0
        
        count = SSProcess.getSelGeoCount()
        logger.info(f"查询到 {count} 个对象")
        
        if count == 0:
            logger.warning("没有需要检查的对象")
            return 0
        
        # 检查执行
        success_count = 0
        error_count = 0
        
        for i in range(count):
            try:
                # 获取对象信息
                geo_id = SSProcess.getSelGeoValue(i, "SSObj_ID")
                
                # 验证必要属性
                assert geo_id, f"对象 {i} 的 ID 为空"
                
                # 执行检查逻辑
                # ...
                
                success_count += 1
                
            except AssertionError as e:
                logger.warning(f"验证失败（索引 {i}）：{str(e)}")
                error_count += 1
                
            except Exception as e:
                logger.error(f"处理对象时出错（索引 {i}）：{str(e)}", exc_info=True)
                error_count += 1
                continue
        
        # 结果汇总
        total_issues = check_fun.GetCheckRecordCount()
        logger.info("========== 检查完成 ==========")
        logger.info(f"成功处理：{success_count} 个对象")
        logger.info(f"处理错误：{error_count} 个对象")
        logger.info(f"发现问题：{total_issues} 个")
        
        # 显示结果
        if total_issues > 0:
            check_fun.ShowCheckOutput()
        
        return total_issues
        
    except Exception as e:
        # 顶层异常处理
        print(f"检查脚本发生严重错误：{str(e)}")
        import traceback
        traceback.print_exc()
        return 0

该示例展示了分层异常处理的设计模式。不同层级的异常有不同的处理策略：数据层异常记录日志后跳过，逻辑层异常记录警告后继续，顶层异常记录详细信息后终止。

Sources: PySSCheck.py, PySSProcess.py

检查规则的部署与管理

随着检查规则数量的增长，需要建立系统的部署和管理机制，包括版本控制、配置管理、自动化执行等。

配置化检查规则

将检查规则与实现逻辑分离，通过配置文件定义规则，提升灵活性。

# 检查规则配置文件 (JSON格式)
CHECK_RULES_CONFIG = {
    "属性检查": {
        "enabled": True,
        "rules": [
            {
                "name": "编码缺失检查",
                "field": "SSObj_Code",
                "condition": "is_empty",
                "severity": "error"
            },
            {
                "name": "名称缺失检查",
                "field": "SSObj_Name",
                "condition": "is_empty",
                "severity": "error"
            }
        ]
    },
    "几何检查": {
        "enabled": True,
        "rules": [
            {
                "name": "面积范围检查",
                "field": "SSObj_Area",
                "min": 1.0,
                "max": 1000000.0,
                "severity": "warning"
            }
        ]
    }
}


def load_check_rules(config_file):
    """从配置文件加载检查规则"""
    import json
    
    with open(config_file, 'r', encoding='utf-8') as f:
        config = json.load(f)
    
    return config


def run_configurable_checks(config):
    """基于配置执行检查"""
    check_fun = CheckFunForPY()
    check_fun.ClearCheckRecord()
    
    for group_name, group_config in config.items():
        if not group_config.get("enabled", True):
            continue
        
        # 执行该组的所有规则
        # ...
    
    return check_fun.GetCheckRecordCount()

配置化检查的优势在于规则的可维护性。业务人员可以通过修改配置文件调整检查规则，而不需要修改代码。同时，不同环境可以使用不同的配置，方便测试和部署。

Sources: PySSCheck.py

学习路径建议

掌握了自定义检查规则后，您可以继续学习以下高级主题，提升数据质量保证能力：

系统路径管理 - 了解如何管理检查规则脚本和配置文件的存储路径
错误处理与异常管理 - 深入学习异常处理模式和错误恢复策略
日志与进度 - 掌握高级日志配置和进度管理技巧
SSProcessManager 完整 API - 查阅完整的 API 文档，发现更多有用的功能

自定义检查规则是 SunvStation 数据质量保证体系中最强大的工具之一。通过系统学习和实践，您可以构建符合特定业务需求的高质量检查系统，为数据生产提供可靠的质量保障。

39 KiB Raw Blame History Unescape Escape

检查规则架构设计

核心组件角色

检查规则的基本结构

标准模板

常见检查类型实现

属性完整性检查

拓扑关系检查

几何规则检查

业务规则检查

高级检查模式

组合检查模式

增量检查模式

性能优化最佳实践

数据访问优化

算法复杂度控制

进度反馈优化

错误处理与调试

检查规则的部署与管理

配置化检查规则

学习路径建议

39 KiB

Raw Blame History