Python脚本如何操作数据库静态脱敏

本文目录导读：

1. 基础架构
2. 常用脱敏方法
3. 数据库操作核心类
4. 完整脱敏流程
5. 使用示例
6. 性能优化版本
7. 配置文件方式
8. 注意事项

我来详细介绍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

注意事项

1. 备份数据：脱敏前务必备份原始数据
2. 分批处理：避免大数据量时内存溢出
3. 事务管理：确保数据一致性
4. 并行处理：多线程处理不同表
5. 验证结果：脱敏后验证数据完整性
6. 性能监控：监控处理进度和性能

这个框架可以根据具体需求扩展,支持更多脱敏规则和数据源。