本文目录导读:

我来为你展示几个Python数据脱敏的实用案例:
常用脱敏工具库
# 安装:pip install faker
from faker import Faker
import random
import re
fake = Faker('zh_CN') # 中文环境
基础脱敏函数
class DataMasker:
"""数据脱敏工具类"""
@staticmethod
def mask_phone(phone: str) -> str:
"""手机号脱敏:138****1234"""
if len(phone) != 11:
return phone
return phone[:3] + '****' + phone[7:]
@staticmethod
def mask_id_card(id_card: str) -> str:
"""身份证脱敏:1101011990****1234"""
if len(id_card) != 18:
return id_card
return id_card[:6] + '********' + id_card[14:]
@staticmethod
def mask_name(name: str) -> str:
"""姓名脱敏:张**"""
if len(name) <= 1:
return name
return name[0] + '*' * (len(name) - 1)
@staticmethod
def mask_email(email: str) -> str:
"""邮箱脱敏:abc***@example.com"""
if '@' not in email:
return email
name, domain = email.split('@')
if len(name) <= 3:
masked_name = name[:1] + '***'
else:
masked_name = name[:3] + '***'
return f"{masked_name}@{domain}"
@staticmethod
def mask_address(address: str) -> str:
"""地址脱敏:北京市朝阳区***"""
if len(address) <= 6:
return address[:3] + '***'
return address[:6] + '***'
@staticmethod
def mask_bank_card(card_no: str) -> str:
"""银行卡号脱敏:6222 **** **** 1234"""
if len(card_no) < 4:
return card_no
return card_no[:4] + ' **** **** ' + card_no[-4:]
实际应用案例
def demo_basic_masking():
"""基础脱敏演示"""
masker = DataMasker()
test_data = {
'姓名': '张三丰',
'手机号': '13812345678',
'身份证': '110101199001011234',
'邮箱': 'zhangsan@example.com',
'地址': '北京市朝阳区建国路88号',
'银行卡': '6222021234567890'
}
print("=== 基础脱敏演示 ===")
print(f"原始姓名: {test_data['姓名']} -> 脱敏后: {masker.mask_name(test_data['姓名'])}")
print(f"原始手机: {test_data['手机号']} -> 脱敏后: {masker.mask_phone(test_data['手机号'])}")
print(f"原始身份证: {test_data['身份证']} -> 脱敏后: {masker.mask_id_card(test_data['身份证'])}")
print(f"原始邮箱: {test_data['邮箱']} -> 脱敏后: {masker.mask_email(test_data['邮箱'])}")
print(f"原始地址: {test_data['地址']} -> 脱敏后: {masker.mask_address(test_data['地址'])}")
print(f"原始银行卡: {test_data['银行卡']} -> 脱敏后: {masker.mask_bank_card(test_data['银行卡'])}")
demo_basic_masking()
使用Faker生成假数据替换
def demo_faker_masking():
"""使用Faker生成假数据替换"""
print("\n=== Faker假数据替换 ===")
# 原始数据
original = {
'name': '李四',
'phone': '13912345678',
'email': 'lisi@company.com',
'company': '某某科技有限公司'
}
# 生成替代数据
masked = {
'name': fake.name(),
'phone': fake.phone_number(),
'email': fake.email(),
'company': fake.company()
}
print("原始数据:", original)
print("脱敏数据:", masked)
demo_faker_masking()
批量数据脱敏处理
import pandas as pd
def batch_masking_example():
"""批量数据处理"""
# 创建示例数据
data = {
'name': ['张三', '李四', '王五', '赵六'],
'phone': ['13800138001', '13900139002', '13700137003', '13600136004'],
'id_card': ['110101199001011111', '110101199002022222',
'110101199003033333', '110101199004044444'],
'salary': [15000, 20000, 18000, 22000]
}
df = pd.DataFrame(data)
masker = DataMasker()
print("=== 批量数据脱敏 ===")
print("原始数据:")
print(df)
# 批量脱敏
df['name'] = df['name'].apply(masker.mask_name)
df['phone'] = df['phone'].apply(masker.mask_phone)
df['id_card'] = df['id_card'].apply(masker.mask_id_card)
# 薪资四舍五入处理
df['salary'] = df['salary'].apply(lambda x: round(x, -3)) # 保留千位
print("\n脱敏后数据:")
print(df)
batch_masking_example()
高级脱敏模式(差分隐私)
import numpy as np
class DifferentialPrivacyMasker:
"""差分隐私脱敏"""
@staticmethod
def add_laplace_noise(data, epsilon=1.0, sensitivity=1.0):
"""
添加拉普拉斯噪声实现差分隐私
epsilon: 隐私预算(越小隐私保护越强)
sensitivity: 敏感度
"""
noise = np.random.laplace(0, sensitivity/epsilon, len(data))
return [x + n for x, n in zip(data, noise)]
def mask_numerical_data(self, data, epsilon=0.5):
"""数值数据脱敏"""
return self.add_laplace_noise(data, epsilon)
def demo_differential_privacy():
"""差分隐私演示"""
masker = DifferentialPrivacyMasker()
# 原始薪资数据
salaries = [15000, 20000, 18000, 22000, 16000]
print("\n=== 差分隐私脱敏 ===")
print(f"原始薪资: {salaries}")
# 不同隐私保护级别
for eps in [0.1, 0.5, 1.0]:
masked = masker.mask_numerical_data(salaries, epsilon=eps)
print(f"隐私预算 ε={eps}: {[round(x) for x in masked]}")
demo_differential_privacy()
正则表达式智能脱敏
class SmartMasker:
"""智能识别并脱敏"""
def __init__(self):
self.patterns = {
'phone': (r'1[3-9]\d{9}', lambda m: m.group()[:3] + '****' + m.group()[7:]),
'email': (r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
lambda m: m.group().split('@')[0][:2] + '***@' + m.group().split('@')[1]),
'id_card': (r'\d{17}[\dXx]', lambda m: m.group()[:6] + '********' + m.group()[14:]),
'bank_card': (r'\d{16,19}', lambda m: m.group()[:4] + ' **** **** ' + m.group()[-4:])
}
def smart_mask(self, text):
"""智能识别并脱敏文本中的敏感信息"""
result = text
for pattern_type, (pattern, mask_func) in self.patterns.items():
result = re.sub(pattern, mask_func, result)
return result
def demo_smart_masking():
"""智能脱敏演示"""
masker = SmartMasker()
text = """
用户信息:
姓名:张三
手机:13812345678
邮箱:zhangsan@example.com
身份证:110101199001011234
银行卡:6222021234567890
"""
print("=== 智能脱敏 ===")
print("原始文本:")
print(text)
print("\n脱敏后:")
print(masker.smart_mask(text))
demo_smart_masking()
完整的数据脱敏流程
class DataMaskingPipeline:
"""完整的数据脱敏流水线"""
def __init__(self, config=None):
self.config = config or {
'name': 'mask_name',
'phone': 'mask_phone',
'id_card': 'mask_id_card',
'email': 'mask_email',
'address': 'mask_address',
'salary': 'round_to_thousand'
}
self.masker = DataMasker()
def mask_record(self, record):
"""脱敏单条记录"""
masked = {}
for field, value in record.items():
method = self.config.get(field)
if method and hasattr(self.masker, method):
masked[field] = getattr(self.masker, method)(value)
elif method == 'round_to_thousand':
masked[field] = round(value, -3) if isinstance(value, (int, float)) else value
else:
masked[field] = value
return masked
def mask_dataset(self, dataset):
"""脱敏整个数据集"""
return [self.mask_record(record) for record in dataset]
def demo_complete_pipeline():
"""完整流水线演示"""
pipeline = DataMaskingPipeline()
# 示例数据
dataset = [
{'name': '张三', 'phone': '13812345678', 'id_card': '110101199001011234',
'email': 'zhangsan@example.com', 'salary': 15000},
{'name': '李四', 'phone': '13987654321', 'id_card': '110101199002022345',
'email': 'lisi@example.com', 'salary': 20000}
]
print("\n=== 完整脱敏流程 ===")
print("原始数据集:")
for record in dataset:
print(record)
print("\n脱敏后数据集:")
masked_dataset = pipeline.mask_dataset(dataset)
for record in masked_dataset:
print(record)
demo_complete_pipeline()
使用建议
-
选择合适的脱敏策略:
- 静态数据:使用固定替换或格式保留加密
- 动态数据:使用实时脱敏或差分隐私
-
性能优化:对大规模数据使用批量处理
-
安全存储:脱敏密钥和规则单独存储和管理
-
合规要求:根据GDPR、个人信息保护法等法规要求实施
这些案例涵盖了从基础到高级的脱敏技术,可以根据实际需求灵活组合使用。