3. 数据质量维度

在数字工地移动巡检中，数据清洗的核心目标是提升数据质量。国际通用的数据质量五大维度——完整性、准确性、一致性、时效性、唯一性——在工地巡检场景下具有特定的业务含义。理解这些维度的定义，是制定清洗规则、优化安全报告性能的前提。

3.1 五大维度定义与巡检场景映射

下表将通用数据质量维度与工地巡检的典型数据项（如巡检点、隐患记录、人员签到、设备状态）进行映射，帮助快速建立场景化认知。

3.2 维度间关联与冲突场景

五大维度并非孤立存在，在工地巡检数据清洗中常出现维度间的权衡与冲突。理解这些关系有助于设计合理的清洗优先级。

• 完整性 vs 准确性：强制补全缺失字段可能引入错误值。例如，自动填充默认GPS坐标会破坏准确性。
• 一致性 vs 时效性：跨系统数据同步需要时间，强一致性要求可能导致数据上传延迟，影响时效性。
• 唯一性 vs 完整性：去重操作可能误删具有相同ID但不同时间戳的有效记录，导致数据丢失。
• 准确性 vs 时效性：现场快速录入时，为满足时效性可能牺牲部分准确性（如手写识别错误）。

3.3 维度评估指标与阈值

为量化数据质量，需为每个维度定义可测量的指标和可接受的阈值。下表给出工地巡检场景的推荐标准。

3.4 数据质量维度关系图

下图展示了五大维度在工地巡检数据清洗中的相互作用关系，以及它们对安全报告性能的最终影响路径。

3.5 维度评估代码示例

以下Python代码片段展示了如何对巡检数据表进行五大维度的批量评估，输出各维度得分与不合格记录清单。

import pandas as pd
import numpy as np

def evaluate_data_quality(df):
    """
    评估工地巡检数据五大维度质量
    :param df: pandas DataFrame，包含巡检数据
    :return: dict，各维度得分与问题记录
    """
    results = {}
    total = len(df)
    
    # 1. 完整性：检查关键字段非空率
    required_cols = ['point_id', 'check_time', 'result', 'inspector']
    non_null = df[required_cols].notna().all(axis=1).sum()
    completeness = round(non_null / total * 100, 2)
    results['completeness'] = {
        'score': completeness,
        'threshold': 98.0,
        'passed': completeness >= 98.0,
        'missing_records': df[df[required_cols].isna().any(axis=1)].index.tolist()
    }
    
    # 2. 准确性：检查GPS坐标是否在工地范围内
    # 假设工地边界：经度[113.5, 114.0]，纬度[22.0, 22.5]
    valid_gps = (
        (df['longitude'] >= 113.5) & (df['longitude'] <= 114.0) &
        (df['latitude'] >= 22.0) & (df['latitude'] <= 22.5)
    )
    accuracy = round(valid_gps.sum() / total * 100, 2)
    results['accuracy'] = {
        'score': accuracy,
        'threshold': 98.0,
        'passed': accuracy >= 98.0,
        'invalid_gps_records': df[~valid_gps].index.tolist()
    }
    
    # 3. 一致性：检查隐患状态在APP与后台是否一致
    consistent = (df['status_app'] == df['status_backend']).sum()
    consistency = round(consistent / total * 100, 2)
    results['consistency'] = {
        'score': consistency,
        'threshold': 99.0,
        'passed': consistency >= 99.0,
        'inconsistent_records': df[df['status_app'] != df['status_backend']].index.tolist()
    }
    
    # 4. 时效性：检查上传延迟是否≤30分钟
    df['delay_min'] = (pd.to_datetime(df['upload_time']) - 
                       pd.to_datetime(df['check_time'])).dt.total_seconds() / 60
    timely = (df['delay_min'] <= 30).sum()
    timeliness = round(timely / total * 100, 2)
    results['timeliness'] = {
        'score': timeliness,
        'threshold': 95.0,
        'passed': timeliness >= 95.0,
        'delayed_records': df[df['delay_min'] > 30].index.tolist()
    }
    
    # 5. 唯一性：检查point_id是否有重复
    duplicate_count = df['point_id'].duplicated().sum()
    uniqueness = round((1 - duplicate_count / total) * 100, 2)
    results['uniqueness'] = {
        'score': uniqueness,
        'threshold': 99.5,
        'passed': uniqueness >= 99.5,
        'duplicate_records': df[df['point_id'].duplicated(keep=False)].index.tolist()
    }
    
    return results

# 使用示例
# df = pd.read_csv('inspection_data.csv')
# quality_report = evaluate_data_quality(df)
# for dim, info in quality_report.items():
#     print(f"{dim}: {info['score']}% - {'通过' if info['passed'] else '未通过'}")

3.6 维度清洗策略矩阵

针对不同维度的质量问题，需要采取差异化的清洗策略。下表总结了常见问题类型及对应的处理方法。