测试数据脱敏：生产数据脱敏后用于测试，3分钟搞定自动化报告推送到Slack_千锋教育

import random

def mask_phone(phone: str) -> str:
    # 保留前3位，后4位随机生成
    prefix = phone[:3]
    suffix = ''.join([str(random.randint(0,9)) for _ in range(8)])
    return prefix + suffix

# 测试
print(mask_phone("13812345678"))  # 输出：138XXXXXXX

from cryptography.fernet import Fernet

# 生成密钥（生产环境必须保存到密钥管理服务）
key = Fernet.generate_key()
cipher = Fernet(key)

def encrypt_credit_card(card_no: str) -> bytes:
    return cipher.encrypt(card_no.encode())

def decrypt_credit_card(encrypted: bytes) -> str:
    return cipher.decrypt(encrypted).decode()

# 测试
encrypted = encrypt_credit_card("4111111111111111")
print(encrypted)  # 乱码
print(decrypt_credit_card(encrypted))  # 4111111111111111

import random
import numpy as np

def perturb_age(age: int, noise_range: tuple = (-3, 3)) -> int:
    noise = random.randint(*noise_range)
    return max(0, age + noise)

# 批量处理
ages = [25, 30, 35, 40]
perturbed_ages = [perturb_age(a) for a in ages]
print(f"原始均值: {np.mean(ages):.2f}, 扰动后均值: {np.mean(perturbed_ages):.2f}")

-- 只抽取最近7天的订单数据
SELECT * FROM orders WHERE created_at >= NOW() - INTERVAL '7 DAYS';

import pandas as pd
from your_masking_lib import mask_phone, mask_email, perturb_age

def mask_dataframe(df: pd.DataFrame) -> pd.DataFrame:
    df['phone'] = df['phone'].apply(mask_phone)
    df['email'] = df['email'].apply(mask_email)
    df['age'] = df['age'].apply(perturb_age)
    return df

# 读取生产数据（假设已导出为CSV）
df = pd.read_csv('production_orders.csv')
masked_df = mask_dataframe(df)
masked_df.to_csv('test_orders.csv', index=False)

def generate_report(masked_count: int, errors: list) -> str:
    report = f"""
    <html>
    <head><title>数据脱敏报告</title></head>
    <body>
        <h1>脱敏完成</h1>
        <p>脱敏记录数: {masked_count}</p>
        <p>错误数: {len(errors)}</p>
        <ul>
            {"".join([f"<li>{e}</li>" for e in errors])}
        </ul>
        <p>完成时间: {datetime.now()}</p>
    </body>
    </html>
    """
    return report

import requests
import json

def send_to_slack(report_url: str, webhook_url: str):
    payload = {
        "text": f"✅ 数据脱敏完成！报告地址：{report_url}",
        "attachments": [{
            "title": "脱敏报告",
            "text": f"共处理 {masked_count} 条记录，错误 {len(errors)} 条",
            "color": "#36a64f" if len(errors) == 0 else "#ff0000"
        }]
    }
    requests.post(webhook_url, data=json.dumps(payload))

# 每天凌晨2点运行
0 2 * * * /usr/bin/python3 /path/to/masking_pipeline.py

def validate_masked_data(df: pd.DataFrame):
    assert df['id_card'].str.len().eq(18).all(), "身份证号长度不为18位"
    assert df['phone'].str.len().eq(11).all(), "手机号长度不为11位"
    assert df['email'].str.contains('@').all(), "邮箱格式错误"

import hashlib

def deterministic_mask(value: str, salt: str = "your_salt") -> str:
    hash_val = hashlib.sha256((value + salt).encode()).hexdigest()
    return hash_val[:10]  # 取前10位作为脱敏值

sensitive_fields = ['phone', 'email', 'id_card', 'credit_card', 'address']

def check_masking_coverage(db_schema: dict) -> float:
    total = 0
    masked = 0
    for table, columns in db_schema.items():
        for col in columns:
            if col in sensitive_fields:
                total += 1
                if col in masked_tables.get(table, []):
                    masked += 1
    return masked / total if total > 0 else 1.0

from scipy.stats import pearsonr

def check_reversibility(original: list, masked: list) -> bool:
    # 如果相关性接近1，说明数据可逆，脱敏失败
    corr, _ = pearsonr(original, masked)
    return abs(corr) < 0.1  # 阈值可调整

def generate_compliance_report():
    report = f"""
    # GDPR合规报告
    - 脱敏覆盖率: {check_masking_coverage(db_schema):.2%}
    - 可逆性检查: {'通过' if check_reversibility(orig, mask) else '失败'}
    - 检查时间: {datetime.now()}
    """
    with open('compliance_report.md', 'w') as f:
        f.write(report)
    send_to_slack(report)