Python应用数据脱敏怎么实现

wen python案例 2026-07-10 331

本文目录导读：

Python应用数据脱敏怎么实现

基础字符串脱敏
高级脱敏方案
结构化数据脱敏
数据库查询结果脱敏
使用第三方库
性能优化与最佳实践
总结与建议

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)

总结与建议

选择合适的脱敏策略：
- 固定替换：适用于简单的敏感信息
- 部分保留：适合需要保持数据格式的情况
- 数据加密：适合需要可恢复的场景
- 数据替换：适合测试环境
性能考虑：
- 使用缓存机制避免重复计算
- 批量处理数据时考虑并发
- 大型数据集考虑流式处理
安全建议：
- 不要在日志中输出原始敏感数据
- 脱敏规则应该配置化，便于管理和修改
- 考虑使用正则表达式确保数据格式正确
测试覆盖：
- 编写单元测试验证脱敏效果
- 测试边界情况（空值、异常格式等）

通过合理使用这些方法,你可以根据实际需求选择最适合的数据脱敏方案。

本文地址： https://www.bmjup.com/post/18673.html