1. 治理导向设计：基于数据域、数据实体和数据属性的三层结构，符合企业数据治理的层级划分，便于与数据资产目录关联。

2. 质量维度体系化：覆盖数据治理核心关注的五个质量维度（完整性、一致性、准确性、唯一性、及时性），每个维度都有具体校验规则。

3. 阈值与状态机制：引入阈值（Threshold）和稽核状态（PASS/WARN/FAIL），支持不同严重程度的问题分级处理，符合数据治理的风险分级原则。

4. 业务规则深度融合：将业务规则（如价格限制、金额计算逻辑）编码为可执行的 SQL 校验，确保数据符合业务运营规范。

5. 可追溯性：通过批次 ID 和详细的错误样本，支持数据质量问题的溯源和整改跟踪，为数据治理的闭环管理提供基础。

-- 数据治理场景下的数据质量稽核脚本
-- 设计目标：满足企业级数据治理要求，确保数据符合标准、规则和业务逻辑
-- 适用范围：数据仓库、业务系统核心表的定期质量稽核

-- 1. 创建稽核结果表（正式表，用于长期跟踪质量趋势）
CREATE TABLE IF NOT EXISTS data_governance_audit_result (
    audit_id INT AUTO_INCREMENT PRIMARY KEY COMMENT '稽核记录ID',
    audit_batch_id VARCHAR(50) COMMENT '稽核批次ID',
    data_domain VARCHAR(50) COMMENT '数据域（如用户域、订单域）',
    data_entity VARCHAR(50) COMMENT '数据实体（表名）',
    data_attribute VARCHAR(50) COMMENT '数据属性（字段名）',
    quality_dimension VARCHAR(20) COMMENT '质量维度：完整性/一致性/准确性/唯一性/及时性',
    audit_rule VARCHAR(200) COMMENT '稽核规则描述',
    total_records BIGINT COMMENT '总记录数',
    error_records BIGINT COMMENT '错误记录数',
    error_rate DECIMAL(10,6) COMMENT '错误率',
    error_samples TEXT COMMENT '错误样本ID（最多10个）',
    audit_status VARCHAR(20) COMMENT '稽核状态：PASS/WARN/FAIL',
    threshold DECIMAL(10,6) COMMENT '阈值',
    audit_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '稽核时间',
    auditor VARCHAR(50) COMMENT '稽核执行人'
);

-- 2. 初始化稽核批次
SET @batch_id = CONCAT('AUDIT_', DATE_FORMAT(NOW(), '%Y%m%d%H%i%s'));
SET @auditor = 'DATA_GOVERNANCE_BOT';

-- 3. 数据标准符合性校验（元数据维度）
-- 3.1 字段长度合规性（基于数据标准定义）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '用户域',
    'user_info',
    'user_name',
    '准确性',
    '用户名长度必须在2-20个字符之间（符合数据标准STD-USER-001）',
    COUNT(*) AS total_records,
    SUM(CASE WHEN LENGTH(user_name) < 2 OR LENGTH(user_name) > 20 THEN 1 ELSE 0 END) AS error_records,
    SUM(CASE WHEN LENGTH(user_name) < 2 OR LENGTH(user_name) > 20 THEN 1 ELSE 0 END)/COUNT(*) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN LENGTH(user_name) < 2 OR LENGTH(user_name) > 20 THEN user_id END SEPARATOR ','), ',', 10) AS error_samples,
    0.01,  -- 1%阈值
    CASE WHEN SUM(CASE WHEN LENGTH(user_name) < 2 OR LENGTH(user_name) > 20 THEN 1 ELSE 0 END)/COUNT(*) > 0.01 THEN 'FAIL'
         WHEN SUM(CASE WHEN LENGTH(user_name) < 2 OR LENGTH(user_name) > 20 THEN 1 ELSE 0 END) > 0 THEN 'WARN'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM user_info;

-- 3.2 数据类型格式合规性（如身份证号、邮箱）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '用户域',
    'user_info',
    'id_card',
    '准确性',
    '身份证号必须符合18位标准格式（最后一位可为X）',
    COUNT(*) AS total_records,
    SUM(CASE WHEN id_card IS NOT NULL AND id_card NOT REGEXP '^[1-9]\\d{5}(18|19|20)\\d{2}((0[1-9])|(1[0-2]))(([0-2][1-9])|10|20|30|31)\\d{3}[0-9Xx]$' THEN 1 ELSE 0 END) AS error_records,
    SUM(CASE WHEN id_card IS NOT NULL AND id_card NOT REGEXP '^[1-9]\\d{5}(18|19|20)\\d{2}((0[1-9])|(1[0-2]))(([0-2][1-9])|10|20|30|31)\\d{3}[0-9Xx]$' THEN 1 ELSE 0 END)/COUNT(*) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN id_card IS NOT NULL AND id_card NOT REGEXP '^[1-9]\\d{5}(18|19|20)\\d{2}((0[1-9])|(1[0-2]))(([0-2][1-9])|10|20|30|31)\\d{3}[0-9Xx]$' THEN user_id END SEPARATOR ','), ',', 10) AS error_samples,
    0.005,  -- 0.5%阈值
    CASE WHEN SUM(CASE WHEN id_card IS NOT NULL AND id_card NOT REGEXP '^[1-9]\\d{5}(18|19|20)\\d{2}((0[1-9])|(1[0-2]))(([0-2][1-9])|10|20|30|31)\\d{3}[0-9Xx]$' THEN 1 ELSE 0 END)/COUNT(*) > 0.005 THEN 'FAIL'
         WHEN SUM(CASE WHEN id_card IS NOT NULL AND id_card NOT REGEXP '^[1-9]\\d{5}(18|19|20)\\d{2}((0[1-9])|(1[0-2]))(([0-2][1-9])|10|20|30|31)\\d{3}[0-9Xx]$' THEN 1 ELSE 0 END) > 0 THEN 'WARN'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM user_info;

-- 4. 跨表数据一致性校验（主外键与业务实体关系）
-- 4.1 订单与用户的参照完整性（数据血缘一致性）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '订单域',
    'order_main',
    'user_id',
    '一致性',
    '订单关联的用户ID必须在用户主表中存在（数据血缘关系：order_main.user_id → user_info.user_id）',
    COUNT(*) AS total_records,
    SUM(CASE WHEN u.user_id IS NULL THEN 1 ELSE 0 END) AS error_records,
    SUM(CASE WHEN u.user_id IS NULL THEN 1 ELSE 0 END)/COUNT(*) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN u.user_id IS NULL THEN o.order_id END SEPARATOR ','), ',', 10) AS error_samples,
    0,  -- 不允许有错误
    CASE WHEN SUM(CASE WHEN u.user_id IS NULL THEN 1 ELSE 0 END) > 0 THEN 'FAIL'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM order_main o
LEFT JOIN user_info u ON o.user_id = u.user_id;

-- 4.2 订单子表与主表金额一致性（业务规则一致性）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '订单域',
    'order_main',
    'total_amount',
    '一致性',
    '订单总金额必须等于子表明细金额之和（业务规则BR-ORDER-002）',
    COUNT(DISTINCT o.order_id) AS total_records,
    SUM(CASE WHEN ABS(o.total_amount - IFNULL(d.detail_sum, 0)) > 0.01 THEN 1 ELSE 0 END) AS error_records,
    SUM(CASE WHEN ABS(o.total_amount - IFNULL(d.detail_sum, 0)) > 0.01 THEN 1 ELSE 0 END)/COUNT(DISTINCT o.order_id) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN ABS(o.total_amount - IFNULL(d.detail_sum, 0)) > 0.01 THEN o.order_id END SEPARATOR ','), ',', 10) AS error_samples,
    0,  -- 不允许有错误
    CASE WHEN SUM(CASE WHEN ABS(o.total_amount - IFNULL(d.detail_sum, 0)) > 0.01 THEN 1 ELSE 0 END) > 0 THEN 'FAIL'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM order_main o
LEFT JOIN (
    SELECT order_id, SUM(item_amount) AS detail_sum
    FROM order_detail
    GROUP BY order_id
) d ON o.order_id = d.order_id;

-- 5. 业务规则合规性校验
-- 5.1 价格规则校验（商品售价不能低于成本价的90%）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '商品域',
    'product_info',
    'sale_price,cost_price',
    '合理性',
    '商品售价不能低于成本价的90%（防止低价倾销，业务规则BR-PROD-001）',
    COUNT(*) AS total_records,
    SUM(CASE WHEN sale_price < cost_price * 0.9 THEN 1 ELSE 0 END) AS error_records,
    SUM(CASE WHEN sale_price < cost_price * 0.9 THEN 1 ELSE 0 END)/COUNT(*) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN sale_price < cost_price * 0.9 THEN product_id END SEPARATOR ','), ',', 10) AS error_samples,
    0.001,  -- 0.1%阈值
    CASE WHEN SUM(CASE WHEN sale_price < cost_price * 0.9 THEN 1 ELSE 0 END)/COUNT(*) > 0.001 THEN 'FAIL'
         WHEN SUM(CASE WHEN sale_price < cost_price * 0.9 THEN 1 ELSE 0 END) > 0 THEN 'WARN'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM product_info
WHERE cost_price > 0;  -- 排除赠品等无成本商品

-- 5.2 数据时效性校验（订单状态变更时间不能晚于当前时间）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '订单域',
    'order_status_log',
    'status_time',
    '及时性',
    '订单状态时间不能晚于当前系统时间（防止未来时间）',
    COUNT(*) AS total_records,
    SUM(CASE WHEN status_time > NOW() THEN 1 ELSE 0 END) AS error_records,
    SUM(CASE WHEN status_time > NOW() THEN 1 ELSE 0 END)/COUNT(*) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(CASE WHEN status_time > NOW() THEN log_id END SEPARATOR ','), ',', 10) AS error_samples,
    0,  -- 不允许有错误
    CASE WHEN SUM(CASE WHEN status_time > NOW() THEN 1 ELSE 0 END) > 0 THEN 'FAIL'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM order_status_log;

-- 6. 数据唯一性校验（基于业务主键）
-- 6.1 用户手机号唯一性（业务唯一键）
INSERT INTO data_governance_audit_result(
    audit_batch_id, data_domain, data_entity, data_attribute, 
    quality_dimension, audit_rule, total_records, error_records, 
    error_rate, error_samples, threshold, audit_status, auditor
)
SELECT 
    @batch_id,
    '用户域',
    'user_info',
    'phone',
    '唯一性',
    '用户手机号必须唯一（业务唯一键规则UK-USER-001）',
    COUNT(*) AS total_records,
    SUM(dup_count) - COUNT(dup_count) AS error_records,  -- 重复记录数=总记录数-唯一记录数
    (SUM(dup_count) - COUNT(dup_count))/COUNT(*) AS error_rate,
    SUBSTRING_INDEX(GROUP_CONCAT(phone SEPARATOR ','), ',', 10) AS error_samples,
    0,  -- 不允许有重复
    CASE WHEN SUM(CASE WHEN dup_count > 1 THEN 1 ELSE 0 END) > 0 THEN 'FAIL'
         ELSE 'PASS' END AS audit_status,
    @auditor
FROM (
    SELECT phone, COUNT(*) AS dup_count
    FROM user_info
    WHERE phone IS NOT NULL
    GROUP BY phone
) t;

-- 7. 生成本次稽核汇总报告
SELECT 
    audit_batch_id,
    data_domain,
    COUNT(DISTINCT data_entity) AS checked_tables,
    SUM(CASE WHEN audit_status = 'PASS' THEN 1 ELSE 0 END) AS pass_count,
    SUM(CASE WHEN audit_status = 'WARN' THEN 1 ELSE 0 END) AS warn_count,
    SUM(CASE WHEN audit_status = 'FAIL' THEN 1 ELSE 0 END) AS fail_count,
    ROUND(SUM(CASE WHEN audit_status = 'PASS' THEN 1 ELSE 0 END)/COUNT(*) * 100, 2) AS pass_rate,
    MAX(audit_time) AS audit_complete_time
FROM data_governance_audit_result
WHERE audit_batch_id = @batch_id
GROUP BY audit_batch_id, data_domain;

-- 8. （可选）将FAIL级别的问题推送到数据治理平台
/*
INSERT INTO data_quality_issue(issue_id, batch_id, domain, entity, attribute, rule, severity, status, create_time)
SELECT 
    CONCAT('ISSUE_', @batch_id, '_', audit_id),
    audit_batch_id,
    data_domain,
    data_entity,
    data_attribute,
    audit_rule,
    'CRITICAL',
    'UNRESOLVED',
    NOW()
FROM data_governance_audit_result
WHERE audit_batch_id = @batch_id
    AND audit_status = 'FAIL';
*/