本文目录导读：
我来详细介绍Python操作数据库静态脱敏的方法：
基础架构
import pymysql
import hashlib
import random
import string
from typing import Any, Dict, List
from dataclasses import dataclass
@dataclass
class SensitiveField:
"""敏感字段配置"""
table: str
column: str
mask_type: str # mask, hash, random, simulate
params: dict = None
常用脱敏方法
class DataMasker:
"""数据脱敏器"""
@staticmethod
def mask_phone(phone: str) -> str:
"""手机号脱敏: 138****1234"""
if len(phone) == 11:
return phone[:3] + "****" + phone[7:]
return phone
@staticmethod
def mask_id_card(id_card: str) -> str:
"""身份证脱敏: 110101****1234"""
if len(id_card) == 18:
return id_card[:6] + "********" + id_card[-4:]
return id_card
@staticmethod
def mask_email(email: str) -> str:
"""邮箱脱敏: a***@example.com"""
if "@" in email:
name, domain = email.split("@")
return name[0] + "***" + "@" + domain
return email
@staticmethod
def mask_name(name: str) -> str:
"""姓名脱敏: 张*"""
if len(name) >= 2:
return name[0] + "*" * (len(name) - 1)
return name
@staticmethod
def hash_data(data: str, algorithm: str = 'sha256') -> str:
"""哈希脱敏"""
hash_func = getattr(hashlib, algorithm)
return hash_func(data.encode()).hexdigest()
@staticmethod
def random_string(length: int = 10) -> str:
"""生成随机字符串"""
return ''.join(random.choices(string.ascii_letters + string.digits, k=length))
数据库操作核心类
class DatabaseMasker:
"""数据库脱敏处理器"""
def __init__(self, db_config: dict):
self.db_config = db_config
self.connection = None
def connect(self):
"""连接数据库"""
self.connection = pymysql.connect(**self.db_config)
def disconnect(self):
"""关闭连接"""
if self.connection:
self.connection.close()
def get_table_data(self, table: str, columns: List[str],
batch_size: int = 1000) -> List[Dict]:
"""分批获取表数据"""
cursor = self.connection.cursor(pymysql.cursors.DictCursor)
columns_str = ', '.join(columns)
offset = 0
while True:
sql = f"""
SELECT {columns_str}
FROM {table}
LIMIT {batch_size} OFFSET {offset}
"""
cursor.execute(sql)
rows = cursor.fetchall()
if not rows:
break
yield rows
offset += batch_size
cursor.close()
def update_data(self, table: str, data: List[Dict],
primary_key: str = 'id'):
"""更新脱敏后的数据"""
cursor = self.connection.cursor()
for row in data:
updates = []
params = []
for column, value in row.items():
if column != primary_key:
updates.append(f"{column} = %s")
params.append(value)
params.append(row[primary_key])
sql = f"""
UPDATE {table}
SET {', '.join(updates)}
WHERE {primary_key} = %s
"""
cursor.execute(sql, params)
self.connection.commit()
cursor.close()
完整脱敏流程
class DataDesensitization:
"""数据脱敏主流程"""
def __init__(self, db_config: dict, sensitive_fields: List[SensitiveField]):
self.masker = DatabaseMasker(db_config)
self.sensitive_fields = sensitive_fields
self.mask_method = DataMasker()
def execute(self):
"""执行脱敏"""
try:
self.masker.connect()
for field in self.sensitive_fields:
print(f"正在脱敏: {field.table}.{field.column}")
self._process_field(field)
print("脱敏完成!")
except Exception as e:
print(f"脱敏失败: {e}")
self.masker.connection.rollback()
finally:
self.masker.disconnect()
def _process_field(self, field: SensitiveField):
"""处理单个字段"""
for batch_data in self.masker.get_table_data(
field.table,
['id', field.column]
):
for row in batch_data:
original_value = row[field.column]
if original_value:
row[field.column] = self._apply_mask(
original_value,
field.mask_type,
field.params
)
self.masker.update_data(
field.table,
batch_data,
primary_key='id'
)
def _apply_mask(self, value: Any, mask_type: str, params: dict = None) -> Any:
"""应用脱敏规则"""
if not value:
return value
if mask_type == 'phone':
return self.mask_method.mask_phone(str(value))
elif mask_type == 'id_card':
return self.mask_method.mask_id_card(str(value))
elif mask_type == 'email':
return self.mask_method.mask_email(str(value))
elif mask_type == 'name':
return self.mask_method.mask_name(str(value))
elif mask_type == 'hash':
algorithm = params.get('algorithm', 'sha256') if params else 'sha256'
return self.mask_method.hash_data(str(value), algorithm)
elif mask_type == 'random':
length = params.get('length', 10) if params else 10
return self.mask_method.random_string(length)
elif mask_type == 'faker':
return self._fake_data(value, params)
else:
raise ValueError(f"不支持的脱敏类型: {mask_type}")
使用示例
# 数据库配置
db_config = {
'host': 'localhost',
'port': 3306,
'user': 'root',
'password': 'password',
'database': 'test_db',
'charset': 'utf8mb4'
}
# 配置敏感字段
sensitive_fields = [
SensitiveField(table='users', column='phone', mask_type='phone'),
SensitiveField(table='users', column='email', mask_type='email'),
SensitiveField(table='users', column='id_card', mask_type='id_card'),
SensitiveField(table='users', column='password',
mask_type='hash', params={'algorithm': 'sha256'}),
SensitiveField(table='address', column='name', mask_type='name'),
]
# 执行脱敏
desensitization = DataDesensitization(db_config, sensitive_fields)
desensitization.execute()
性能优化版本
class OptimizedDatabaseMasker(DatabaseMasker):
"""优化版的数据库脱敏器"""
def __init__(self, db_config: dict):
super().__init__(db_config)
self.batch_update_size = 500
def batch_update(self, table: str, data: List[Dict],
primary_key: str = 'id'):
"""批量更新优化"""
if not data:
return
cursor = self.connection.cursor()
# 构建批量更新SQL
update_columns = [col for col in data[0].keys() if col != primary_key]
case_statements = []
for column in update_columns:
case_sql = f"{column} = CASE {primary_key}\n"
for row in data:
case_sql += f" WHEN %s THEN %s\n"
case_sql += "END"
case_statements.append(case_sql)
sql = f"""
UPDATE {table}
SET {', '.join(case_statements)}
WHERE {primary_key} IN ({','.join(['%s'] * len(data))})
"""
# 准备参数
params = []
for column in update_columns:
for row in data:
params.append(row[primary_key])
params.append(row[column])
params.extend([row[primary_key] for row in data])
cursor.execute(sql, params)
self.connection.commit()
cursor.close()
配置文件方式
# mask_config.yaml
database:
host: localhost
port: 3306
user: root
password: password
database: test_db
mask_rules:
- table: users
columns:
- name: phone
type: phone
- name: email
type: email
- name: password
type: hash
params:
algorithm: sha256
- table: orders
columns:
- name: receiver_name
type: name
- name: receiver_phone
type: phone
注意事项
- 备份数据：脱敏前务必备份原始数据
- 分批处理：避免大数据量时内存溢出
- 事务管理：确保数据一致性
- 并行处理：多线程处理不同表
- 验证结果：脱敏后验证数据完整性
- 性能监控：监控处理进度和性能
这个框架可以根据具体需求扩展,支持更多脱敏规则和数据源。
