密钥管理系统与数据质量平台的深度集成策略

kms-dq-integration/
├── config/
│   └── fields_meta.yaml       # 数据字段元数据配置（是否加密、密钥ID）
├── core/
│   ├── __init__.py
│   ├── kms_client.py          # 模拟KMS客户端
│   ├── dq_engine.py           # 数据质量规则引擎
│   └── kms_aware_checker.py   # 集成KMS的检查器入口
├── rules/
│   ├── __init__.py
│   ├── base_rule.py           # 抽象规则基类
│   ├── non_null_rule.py       # 非空检查规则
│   └── email_format_rule.py   # 邮箱格式规则
├── main.py                    # 主程序入口
├── requirements.txt           # 项目依赖
└── sample_data.json          # 示例数据（含加密与明文字段）

table_name: "user_profiles"
fields:

  - name: "user_id"
    encrypted: false

  - name: "email"
    encrypted: false

  - name: "phone_number"
    encrypted: true
    key_id: "key_phone_v1"  # 在模拟KMS中对应的密钥标识

  - name: "ssn"
    encrypted: true
    key_id: "key_ssn_v1"

import hashlib
from typing import Optional, Dict

class SimulatedKMSClient:
    """模拟的KMS客户端。仅用于演示集成逻辑。"""

    def __init__(self):
        # 模拟密钥库：key_id -> "密钥"。实际应为加密密钥材料。
        # 此处使用简单的字符串模拟，并使用一个固定的盐值进行"加密/解密"模拟。
        self._key_store: Dict[str, str] = {
            "key_phone_v1": "simulated-secret-key-phone-123",
            "key_ssn_v1": "simulated-secret-key-ssn-456",
        }
        self._salt = "kms-dq-integration-salt"

    def get_key_material(self, key_id: str) -> Optional[str]:
        """模拟从KMS获取密钥材料。实际应用中返回的可能是密钥句柄或需调用解密API。"""
        print(f"[SimulatedKMS] Requesting key material for Key ID: {key_id}")
        return self._key_store.get(key_id)

    def decrypt_field(self, encrypted_value: str, key_id: str) -> Optional[str]:
        """
        模拟解密过程。
        注意：真实场景中，解密应发生在KMS服务端或使用安全的本地密码库。
        此处仅为逻辑演示，使用可逆的简单哈希操作模拟。
        """
        key_material = self.get_key_material(key_id)
        if not key_material or not encrypted_value:
            return None

        # !!! 警告：这是不安全的模拟解密，仅用于演示流程 !!!
        # 模拟逻辑：假设"加密"是 `value + key + salt` 的MD5。
        # 为了"解密"，我们尝试逆向查找。真实场景是密码学解密。
        print(f"[SimulatedKMS] Attempting to decrypt value for Key ID: {key_id}")
        for possible_original in [f"test-{i}" for i in range(1000)] + ["123-45-6789", "john.doe@example.com"]:
            simulated_encrypted = hashlib.md5(
                (possible_original + key_material + self._salt).encode()
            ).hexdigest()
            if simulated_encrypted == encrypted_value:
                return possible_original
        return None

    def health_check(self) -> bool:
        """模拟KMS健康检查。"""
        return True

from abc import ABC, abstractmethod
from typing import Any, Dict

class DataQualityRule(ABC):
    """数据质量规则抽象基类。"""

    def __init__(self, field_name: str):
        self.field_name = field_name

    @abstractmethod
    def validate(self, value: Any, record: Dict[str, Any]) -> tuple[bool, str]:
        """
        验证字段值。
        返回: (是否通过, 错误信息)
        """
        pass

    def __str__(self):
        return f"{self.__class__.__name__}(field={self.field_name})"

from .base_rule import DataQualityRule

class NonNullRule(DataQualityRule):
    """检查字段值非空（不为None或空字符串）。"""

    def validate(self, value: Any, record: Dict[str, Any]) -> tuple[bool, str]:
        is_valid = value is not None and str(value).strip() != ''
        message = "" if is_valid else f"字段 '{self.field_name}' 为空。"
        return is_valid, message

import re
from .base_rule import DataQualityRule

class EmailFormatRule(DataQualityRule):
    """简单的邮箱格式验证规则。"""

    # 简单的邮箱正则，生产环境应使用更严谨的校验
    EMAIL_REGEX = re.compile(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$')

    def validate(self, value: Any, record: Dict[str, Any]) -> tuple[bool, str]:
        if value is None or str(value).strip() == '':
            return True, ""  # 空值由NotNullRule检查，本规则跳过

        is_valid = bool(self.EMAIL_REGEX.match(str(value).strip()))
        message = "" if is_valid else f"字段 '{self.field_name}' 的邮箱格式无效：'{value}'。"
        return is_valid, message

from typing import List, Dict, Any
from rules.base_rule import DataQualityRule

class DataQualityEngine:
    """数据质量规则执行引擎。"""

    def __init__(self):
        self.rules: List[DataQualityRule] = []

    def register_rule(self, rule: DataQualityRule):
        """向引擎注册一条质量规则。"""
        self.rules.append(rule)
        print(f"[DQ Engine] 规则已注册: {rule}")

    def run_checks(self, data: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        对数据集运行所有已注册的质量检查。
        返回汇总报告。
        """
        report = {
            "total_records": len(data),
            "passed_records": 0,
            "failed_records": 0,
            "record_details": []
        }

        for i, record in enumerate(data):
            record_passed = True
            errors = []

            for rule in self.rules:
                field_value = record.get(rule.field_name)
                is_valid, error_msg = rule.validate(field_value, record)
                if not is_valid:
                    record_passed = False
                    errors.append(error_msg)

            record_detail = {
                "record_id": i,
                "record_data": record,
                "passed": record_passed,
                "errors": errors
            }
            report["record_details"].append(record_detail)

            if record_passed:
                report["passed_records"] += 1
            else:
                report["failed_records"] += 1
                print(f"[DQ Engine] 记录 {i} 检查失败。错误: {errors}")

        print(f"[DQ Engine] 检查完成。通过: {report['passed_records']}, 失败: {report['failed_records']}")
        return report

import yaml
from typing import List, Dict, Any
from .kms_client import SimulatedKMSClient
from .dq_engine import DataQualityEngine

class KMSAwareDqChecker:
    """集成了KMS能力的数据质量检查器。"""

    def __init__(self, meta_config_path: str):
        self.meta_config = self._load_meta_config(meta_config_path)
        self.kms_client = SimulatedKMSClient()
        self.dq_engine = DataQualityEngine()

    def _load_meta_config(self, path: str) -> Dict[str, Any]:
        """加载字段元数据配置文件。"""
        with open(path, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)

    def _preprocess_record(self, record: Dict[str, Any]) -> Dict[str, Any]:
        """
        预处理单条记录：对加密字段进行解密。
        返回一个新的、包含解密后明文数据的记录字典，用于质量检查。
        """
        processed_record = record.copy()
        for field_info in self.meta_config.get('fields', []):
            field_name = field_info['name']
            if field_info.get('encrypted', False) and field_name in processed_record:
                encrypted_value = processed_record[field_name]
                key_id = field_info.get('key_id')
                if encrypted_value and key_id:
                    # 关键集成点：调用KMS客户端进行解密
                    decrypted_value = self.kms_client.decrypt_field(encrypted_value, key_id)
                    if decrypted_value is not None:
                        # 将解密后的值替换原加密值，供后续规则检查
                        processed_record[field_name] = decrypted_value
                        print(f"[KMS-Aware Checker] 字段 '{field_name}' 解密成功。")
                    else:
                        print(f"[KMS-Aware Checker] 警告：字段 '{field_name}' 解密失败，将使用原始值检查。")
        return processed_record

    def register_rules(self, rules: List[DataQualityRule]):
        """将规则注册到底层引擎。"""
        for rule in rules:
            self.dq_engine.register_rule(rule)

    def run(self, data: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        执行完整的KMS感知的数据质量检查流程。

        1. 预处理（解密）每条记录。
        2. 对预处理后的数据运行质量规则。
        """
        if not self.kms_client.health_check():
            raise ConnectionError("KMS健康检查失败，无法继续。")

        print(f"[KMS-Aware Checker] 开始处理 {len(data)} 条记录，包含加密字段预处理。")
        processed_data = [self._preprocess_record(record) for record in data]

        print(f"[KMS-Aware Checker] 开始执行数据质量规则检查...")
        report = self.dq_engine.run_checks(processed_data)
        return report

import json
from core.kms_aware_checker import KMSAwareDqChecker
from rules.non_null_rule import NonNullRule
from rules.email_format_rule import EmailFormatRule

def main():
    # 1. 初始化集成检查器，加载元数据配置
    checker = KMSAwareDqChecker('config/fields_meta.yaml')

    # 2. 注册数据质量规则
    # 规则应用于解密后的明文字段
    checker.register_rules([
        NonNullRule("user_id"),
        EmailFormatRule("email"),
        NonNullRule("phone_number"), # 对解密后的phone_number进行非空检查
        NonNullRule("ssn"),          # 对解密后的ssn进行非空检查
    ])

    # 3. 加载示例数据（其中包含加密字段）
    with open('sample_data.json', 'r', encoding='utf-8') as f:
        sample_data = json.load(f)

    # 4. 运行集成检查
    print("\n" + "="*50)
    print("启动 KMS 与数据质量平台集成检查演示")
    print("="*50)
    final_report = checker.run(sample_data)

    # 5. 打印最终报告摘要
    print("\n" + "="*50)
    print("最终检查报告摘要")
    print("="*50)
    print(f"数据表: {checker.meta_config.get('table_name')}")
    print(f"总记录数: {final_report['total_records']}")
    print(f"通过记录: {final_report['passed_records']}")
    print(f"失败记录: {final_report['failed_records']}")
    print("="*50)

    # 简单判定
    if final_report['failed_records'] == 0:
        print("结果: 所有数据质量检查通过！")
    else:
        print(f"结果: 发现 {final_report['failed_records']} 条记录存在质量问题。详情见上方日志。")

if __name__ == "__main__":
    main()

[
  {
    "user_id": 1,
    "email": "alice@example.com",
    "phone_number": "7f1c41c8b0b3c3d3e4f5a6b7c8d9e0f1",
    "ssn": "a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6"
  },
  {
    "user_id": 2,
    "email": "bob.notvalid-email",
    "phone_number": "123-45-6789",
    "ssn": "e7f8a9b0c1d2e3f4a5b6c7d8e9f0a1b2"
  },
  {
    "user_id": 3,
    "email": "charlie@example.org",
    "phone_number": "",
    "ssn": "c3d4e5f6a7b8c9d0e1f2a3b4c5d6e7f8"
  }
]

pyyaml>=6.0

pip install -r requirements.txt

python main.py