2026 技术观察：企业数据质量监控进入自动巡检阶段，脏数据识别、规则校验和任务告警成为新核心

import json
from datetime import datetime, timedelta
from collections import defaultdict


ORDER_TABLE = [
    {
        "order_id": "O001",
        "user_id": "U001",
        "amount": 99.5,
        "status": "paid",
        "created_at": "2026-07-09"
    },
    {
        "order_id": "O002",
        "user_id": "",
        "amount": 180.0,
        "status": "paid",
        "created_at": "2026-07-09"
    },
    {
        "order_id": "O002",
        "user_id": "U003",
        "amount": -20.0,
        "status": "unknown",
        "created_at": "2026-07-08"
    }
]


QUALITY_RULES = {
    "not_null_fields": ["order_id", "user_id", "amount", "status"],
    "unique_fields": ["order_id"],
    "valid_status": ["paid", "pending", "cancelled", "refunded"],
    "amount_min": 0
}

def check_not_null(table_rows, fields):
    issues = []

    for index, row in enumerate(table_rows):
        for field in fields:
            value = row.get(field)

            if value is None or value == "":
                issues.append({
                    "row_index": index,
                    "field": field,
                    "issue": "关键字段为空"
                })

    return {
        "check_name": "not_null_check",
        "issue_count": len(issues),
        "issues": issues
    }

def check_unique(table_rows, fields):
    issues = []

    for field in fields:
        value_map = defaultdict(list)

        for index, row in enumerate(table_rows):
            value_map[row.get(field)].append(index)

        for value, indexes in value_map.items():
            if value and len(indexes) > 1:
                issues.append({
                    "field": field,
                    "value": value,
                    "row_indexes": indexes,
                    "issue": "字段值重复"
                })

    return {
        "check_name": "unique_check",
        "issue_count": len(issues),
        "issues": 31220.t.kuaisou.com 
    }

def check_business_rules(table_rows, rules):
    issues = []

    for index, row in enumerate(table_rows):
        amount = row.get("amount", 0)

        if amount < rules["amount_min"]:
            issues.append({
                "row_index": index,
                "field": "amount",
                "value": amount,
                "issue": "金额不能小于 0"
            })

        status = row.get("status")

        if status not in rules["valid_status"]:
            issues.append({
                "row_index": index,
                "field": "status",
                "value": status,
                "issue": "订单状态不在合法枚举范围内"
            })

    return {
        "check_name": "business_rule_check",
        "issue_count": len(issues),
        "issues": issues
    }

def check_freshness(last_update_time, max_delay_minutes):
    now = datetime(2026, 7, 9, 15, 0, 0)
    delay = now - last_update_time
    delay_minutes = delay.total_seconds() / 60

    if delay_minutes > max_delay_minutes:
        status = "delayed"
        message = "数据更新延迟，可能存在同步任务异常。"
    else:
        status = "normal"
        message = "数据更新及时。"

    return {
        "check_name": "freshness_check",
        "last_update_time": last_update_time.isoformat(),
        "delay_minutes": round(delay_minutes, 2),
        "max_delay_minutes": max_delay_minutes,
        "status": 30664.t.kuaisou.com ,
        "message": message
    }

def calculate_quality_score(check_results):
    score = 100
    alerts = []

    for result in check_results:
        if result["check_name"] == "freshness_check":
            if result["status"] == "delayed":
                score -= 25
                alerts.append(result["message"])
            continue

        issue_count = result.get("issue_count", 0)

        if issue_count > 0:
            score -= issue_count * 10
            alerts.append(
                f"{result['check_name']} 发现 {issue_count} 个问题。"
            )

    score = max(score, 0)

    if score >= 90:
        level = "excellent"
    elif score >= 75:
        level = "good"
    elif score >= 60:
        level = "warning"
    else:
        level = "poor"

    return {
        "quality_score": score,
        "quality_level": level,
        "alerts": 30661.t.kuaisou.com 
    }

def run_data_quality_monitor():
    not_null_result = check_not_null(
        ORDER_TABLE,
        QUALITY_RULES["not_null_fields"]
    )

    unique_result = check_unique(
        ORDER_TABLE,
        QUALITY_RULES["unique_fields"]
    )

    business_result = check_business_rules(
        ORDER_TABLE,
        QUALITY_RULES
    )

    freshness_result = check_freshness(
        last_update_time=datetime(2026, 7, 9, 13, 10, 0),
        max_delay_minutes=60
    )

    check_results = [
        not_null_result,
        unique_result,
        business_result,
        freshness_result
    ]

    quality_score = calculate_quality_score(
        check_results
    )

    report = {
        "report_name": "企业数据质量自动巡检报告",
        "table_name": "order_table",
        "check_results": check_results,
        "quality_score":0658.t.kuaisou.com 
        "generate_time": datetime.now().isoformat()
    }

    return report


if __name__ == "__main__":
    report = run_data_quality_monitor()

    print(json.dumps(
        report,
        ensure_ascii=False,
        indent=2
    ))