本文目录导读：
Python实现数据脱敏（Data Masking）有多种方式，具体取决于你的数据格式、脱敏场景和需求,下面我为你介绍几种常见的实现方法。
基础字符串脱敏
固定长度脱敏
def mask_string(text, start=0, end=None, mask_char='*'):
"""将字符串的指定部分替换为掩码字符"""
if not text:
return text
text = str(text)
if end is None:
end = len(text)
return text[:start] + mask_char * (end - start) + text[end:]
# 示例
phone = "13800138000"
masked_phone = mask_string(phone, start=3, end=7)
print(masked_phone) # 138****8000
email = "zhangsan@example.com"
# 只保留邮箱前缀的第一个字符和域名
name, domain = email.split('@')
masked_email = name[0] + "****" + domain
print(masked_email) # z****example.com
id_card = "110101199001011234"
masked_id = mask_string(id_card, start=6, end=14)
print(masked_id) # 110101********1234
正则表达式脱敏
import re
def regex_mask(text, pattern, replacement):
"""使用正则表达式进行脱敏"""
return re.sub(pattern, replacement, text)
# 手机号脱敏
phone_pattern = r'(\d{3})\d{4}(\d{4})'
phone_replacement = r'\1****\2'
phones = ["13800138000", "13912345678"]
masked_phones = [regex_mask(phone, phone_pattern, phone_replacement)
for phone in phones]
print(masked_phones) # ['138****8000', '139****5678']
# 邮箱脱敏
email_pattern = r'(\w)[^@]*@'
email_replacement = r'\1****@'
emails = ["zhangsan@example.com", "lisi@test.org"]
masked_emails = [regex_mask(email, email_pattern, email_replacement)
for email in emails]
print(masked_emails) # ['z****@example.com', 'l****@test.org']
高级脱敏方案
类封装实现
from typing import Union, Optional
import re
import random
import hashlib
class DataMasker:
"""数据脱敏器"""
def __init__(self, mask_char: str = '*'):
self.mask_char = mask_char
def phone(self, phone: str) -> str:
"""手机号脱敏：保留前3后4"""
if not phone or len(phone) < 7:
return phone
return phone[:3] + self.mask_char * 4 + phone[-4:]
def email(self, email: str) -> str:
"""邮箱脱敏：保留用户名首字符和域名"""
if not email or '@' not in email:
return email
name, domain = email.split('@')
if len(name) <= 1:
return email
return name[0] + self.mask_char * 5 + '@' + domain
def id_card(self, id_card: str) -> str:
"""身份证号脱敏：保留前6后4"""
if not id_card or len(id_card) < 10:
return id_card
return id_card[:6] + self.mask_char * 8 + id_card[-4:]
def bank_card(self, card_no: str) -> str:
"""银行卡号脱敏：保留前4后4"""
if not card_no or len(card_no) < 8:
return card_no
# 去除可能的空格
card_no = card_no.replace(' ', '')
return card_no[:4] + ' ' + self.mask_char * 8 + ' ' + card_no[-4:]
def name(self, name: str) -> str:
"""姓名脱敏：保留姓氏"""
if not name:
return name
if len(name) == 2:
return name[0] + self.mask_char
elif len(name) >= 3:
return name[0] + self.mask_char * (len(name) - 2) + name[-1]
return name
def address(self, address: str, keep_start: int = 6) -> str:
"""地址脱敏：保留前几个字符"""
if not address or len(address) <= keep_start:
return address
return address[:keep_start] + self.mask_char * min(10, len(address) - keep_start)
# 使用示例
masker = DataMasker()
user_data = {
"name": "张三丰",
"phone": "13800138000",
"email": "zhangsan@example.com",
"id_card": "110101199001011234",
"bank_card": "6222021234567890",
"address": "北京市朝阳区xxx街道xxx小区"
}
masked_data = {
"name": masker.name(user_data["name"]),
"phone": masker.phone(user_data["phone"]),
"email": masker.email(user_data["email"]),
"id_card": masker.id_card(user_data["id_card"]),
"bank_card": masker.bank_card(user_data["bank_card"]),
"address": masker.address(user_data["address"])
}
print(masked_data)
# {'name': '张*丰', 'phone': '138****8000', 'email': 'z****@example.com',
# 'id_card': '110101********1234', 'bank_card': '6222 ******** 7890',
# 'address': '北京市**********'}
结构化数据脱敏
字典/JSON数据脱敏
from typing import Dict, Any, List
import json
class StructuredDataMasker:
"""结构化数据脱敏"""
def __init__(self):
self.masker = DataMasker()
# 定义不同字段的脱敏规则
self.mask_rules = {
'phone': self.masker.phone,
'mobile': self.masker.phone,
'tel': self.masker.phone,
'email': self.masker.email,
'id_card': self.masker.id_card,
'id_number': self.masker.id_card,
'bank_card': self.masker.bank_card,
'bank_card_no': self.masker.bank_card,
'name': self.masker.name,
'address': self.masker.address,
'password': lambda x: self.mask_char * 8,
'token': lambda x: x[:10] + '...' if x else x,
}
self.mask_char = '*'
def mask_dict(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""脱敏字典数据"""
masked_data = {}
for key, value in data.items():
if key in self.mask_rules:
masked_data[key] = self.mask_rules[key](value)
elif isinstance(value, dict):
masked_data[key] = self.mask_dict(value)
elif isinstance(value, list):
masked_data[key] = [self.mask_dict(item) if isinstance(item, dict)
else item for item in value]
else:
masked_data[key] = value
return masked_data
def mask_json(self, json_str: str) -> str:
"""脱敏JSON字符串"""
data = json.loads(json_str)
masked_data = self.mask_dict(data)
return json.dumps(masked_data, ensure_ascii=False)
# 使用示例
masker = StructuredDataMasker()
# 字典数据
user_info = {
"name": "李四",
"phone": "13912345678",
"email": "lisi@test.com",
"id_card": "320102198501012345",
"address": "上海市浦东新区xxx路xxx号",
"company": {
"name": "XX科技公司",
"address": "北京市海淀区中关村"
},
"hobbies": ["reading", "sports"]
}
masked_info = masker.mask_dict(user_info)
print(json.dumps(masked_info, ensure_ascii=False, indent=2))
Pandas DataFrame脱敏
import pandas as pd
import numpy as np
def mask_dataframe(df: pd.DataFrame, columns: Dict[str, str]) -> pd.DataFrame:
"""
对DataFrame指定列进行脱敏
Args:
df: 原始DataFrame
columns: 列名和脱敏类型的映射，如 {'phone': 'phone', 'name': 'name'}
Returns:
脱敏后的DataFrame
"""
masker = DataMasker()
masked_df = df.copy()
mask_functions = {
'phone': masker.phone,
'email': masker.email,
'id_card': masker.id_card,
'name': masker.name,
'address': masker.address
}
for col, mask_type in columns.items():
if col in masked_df.columns and mask_type in mask_functions:
masked_df[col] = masked_df[col].apply(mask_functions[mask_type])
return masked_df
# 创建示例数据
df = pd.DataFrame({
'name': ['张三', '李四', '王五'],
'phone': ['13800138000', '13912345678', '13798765432'],
'email': ['a@example.com', 'b@test.com', 'c@demo.org'],
'salary': [8000, 12000, 15000]
})
# 指定需要脱敏的列
mask_columns = {
'name': 'name',
'phone': 'phone',
'email': 'email'
}
masked_df = mask_dataframe(df, mask_columns)
print("原始数据:")
print(df)
print("\n脱敏后数据:")
print(masked_df)
数据库查询结果脱敏
from typing import List, Dict, Any
import sqlite3 # 示例使用SQLite，实际项目中可替换为其他数据库
class DatabaseMasker:
"""数据库查询结果脱敏"""
def __init__(self, db_path: str):
self.conn = sqlite3.connect(db_path)
self.masker = DataMasker()
self.mask_rules = {
'phone': self.masker.phone,
'email': self.masker.email,
'name': self.masker.name
}
def query_with_mask(self, sql: str, mask_columns: List[str]) -> List[Dict[str, Any]]:
"""执行SQL查询并对指定列脱敏"""
cursor = self.conn.cursor()
cursor.execute(sql)
# 获取列名
columns = [description[0] for description in cursor.description]
# 构建结果
results = []
for row in cursor.fetchall():
row_dict = {}
for i, col in enumerate(columns):
value = row[i]
if col in mask_columns and col in self.mask_rules:
value = self.mask_rules[col](value)
row_dict[col] = value
results.append(row_dict)
return results
def close(self):
"""关闭数据库连接"""
self.conn.close()
# 使用示例
# db_masker = DatabaseMasker('example.db')
# users = db_masker.query_with_mask(
# "SELECT name, phone, email, salary FROM users",
# mask_columns=['phone', 'email', 'name']
# )
# db_masker.close()
使用第三方库
faker库生成模拟数据
from faker import Faker
fake = Faker('zh_CN')
def generate_masked_data(real_data: str, data_type: str) -> str:
"""使用faker生成脱敏数据"""
fake_data_generators = {
'name': fake.name,
'phone': fake.phone_number,
'email': fake.email,
'address': fake.address,
'company': fake.company,
'id_card': fake.ssn # 中国身份证号
}
if data_type in fake_data_generators:
return fake_data_generators[data_type]()
return real_data
# 使用示例
real_name = "张三"
masked_name = generate_masked_data(real_name, 'name')
print(f"原始姓名: {real_name}, 脱敏后: {masked_name}")
性能优化与最佳实践
import functools
import logging
class DataMaskingManager:
"""数据脱敏管理器（带缓存和日志）"""
def __init__(self):
self.masker = DataMasker()
self.mask_cache = {}
logger = logging.getLogger(__name__)
@functools.lru_cache(maxsize=1000)
def mask_phone_cached(self, phone: str) -> str:
"""带缓存的手机号脱敏"""
return self.masker.phone(phone)
def mask_sensitive_data(self, data: Any, fields: List[str]) -> Any:
"""批量脱敏敏感数据"""
if isinstance(data, dict):
return {
key: self.mask_field(key, value) if key in fields
else self.mask_sensitive_data(value, fields)
if isinstance(value, (dict, list)) else value
for key, value in data.items()
}
elif isinstance(data, list):
return [self.mask_sensitive_data(item, fields) for item in data]
return data
def mask_field(self, field_name: str, value: str) -> str:
"""根据字段名选择合适的脱敏方法"""
mask_methods = {
'phone': self.mask_phone_cached,
'mobile': self.mask_phone_cached,
'email': self.masker.email,
'name': self.masker.name
}
if field_name in mask_methods and value:
try:
return mask_methods[field_name](value)
except Exception as e:
logging.error(f"脱敏失败: {field_name}={value}, 错误: {e}")
return value
return value
# 使用示例
manager = DataMaskingManager()
# 批量处理
sensitive_data = {
"users": [
{"name": "张三", "phone": "13800138000", "email": "a@test.com"},
{"name": "李四", "phone": "13912345678", "email": "b@test.com"}
]
}
masked = manager.mask_sensitive_data(sensitive_data, ['phone', 'email', 'name'])
print(masked)
总结与建议
-
选择合适的脱敏策略：
- 固定替换：适用于简单的敏感信息
- 部分保留：适合需要保持数据格式的情况
- 数据加密：适合需要可恢复的场景
- 数据替换：适合测试环境
-
性能考虑：
- 使用缓存机制避免重复计算
- 批量处理数据时考虑并发
- 大型数据集考虑流式处理
-
安全建议：
- 不要在日志中输出原始敏感数据
- 脱敏规则应该配置化，便于管理和修改
- 考虑使用正则表达式确保数据格式正确
-
测试覆盖：
- 编写单元测试验证脱敏效果
- 测试边界情况（空值、异常格式等）
通过合理使用这些方法,你可以根据实际需求选择最适合的数据脱敏方案。
