Python脱敏工具案例如何封装数据脱敏

本文目录导读：

1. 基础脱敏工具类封装
2. 增强版脱敏工具类
3. 使用示例
4. 配置化脱敏封装
5. 单元测试
6. 使用建议

我来为您介绍一个Python数据脱敏工具的封装案例,包含完整的类设计和多种脱敏策略。

基础脱敏工具类封装

import re
import random
from typing import Any, Dict, List, Union, Optional
from datetime import datetime, timedelta
import json
import hashlib
class DataMasker:
    """
    数据脱敏工具类
    支持多种脱敏策略：手机号、邮箱、身份证、姓名、地址、银行卡号、密码等
    """
    def __init__(self, mask_char: str = '*', keep_prefix: int = 3, keep_suffix: int = 4):
        """
        初始化脱敏工具
        Args:
            mask_char: 脱敏替换字符
            keep_prefix: 保留前缀长度
            keep_suffix: 保留后缀长度
        """
        self.mask_char = mask_char
        self.keep_prefix = keep_prefix
        self.keep_suffix = keep_suffix
    def mask_phone(self, phone: str) -> str:
        """
        手机号脱敏：保留前3后4位
        Args:
            phone: 手机号码
        Returns:
            脱敏后的手机号
        """
        if not phone or len(phone) != 11:
            return phone
        return phone[:3] + self.mask_char * 4 + phone[-4:]
    def mask_email(self, email: str) -> str:
        """
        邮箱脱敏：@前保留前2位，其余用*代替
        Args:
            email: 邮箱地址
        Returns:
            脱敏后的邮箱
        """
        if not email or '@' not in email:
            return email
        name, domain = email.split('@')
        if len(name) <= 2:
            masked_name = name[0] + self.mask_char * (len(name) - 1)
        else:
            masked_name = name[:2] + self.mask_char * (len(name) - 2)
        return f"{masked_name}@{domain}"
    def mask_id_card(self, id_card: str) -> str:
        """
        身份证号脱敏：保留前6后4位
        Args:
            id_card: 身份证号
        Returns:
            脱敏后的身份证号
        """
        if not id_card or len(id_card) < 10:
            return id_card
        return id_card[:6] + self.mask_char * 8 + id_card[-4:]
    def mask_name(self, name: str) -> str:
        """
        姓名脱敏：保留姓，名用*代替
        Args:
            name: 姓名
        Returns:
            脱敏后的姓名
        """
        if not name or len(name) < 2:
            return name
        if len(name) == 2:
            return name[0] + self.mask_char
        elif len(name) == 3:
            return name[0] + self.mask_char * 2
        else:
            return name[0] + self.mask_char * (len(name) - 2) + name[-1]
    def mask_address(self, address: str, keep_level: int = 2) -> str:
        """
        地址脱敏：保留前N级地址
        Args:
            address: 详细地址
            keep_level: 保留的地址层级数（省/市/区）
        Returns:
            脱敏后的地址
        """
        if not address:
            return address
        # 简单的地址层级拆分
        levels = re.split(r'[省市区县街道镇乡]', address)
        if len(levels) > keep_level:
            masked = ''.join(levels[:keep_level])
            # 补全层级标识
            markers = ['省', '市', '区', '县', '街道', '镇', '乡']
            for i in range(min(keep_level, len(markers))):
                if i < len(masked):
                    pass
            return masked + self.mask_char * 4
        return address
    def mask_bank_card(self, card_number: str) -> str:
        """
        银行卡号脱敏：保留前6后4位
        Args:
            card_number: 银行卡号
        Returns:
            脱敏后的银行卡号
        """
        if not card_number or len(card_number) < 10:
            return card_number
        return card_number[:6] + self.mask_char * (len(card_number) - 10) + card_number[-4:]
    def mask_password(self, password: str) -> str:
        """
        密码脱敏：全部替换为*
        Args:
            password: 密码
        Returns:
            脱敏后的密码
        """
        if not password:
            return password
        return self.mask_char * len(password)
    def mask_ip(self, ip_address: str) -> str:
        """
        IP地址脱敏：保留前两段
        Args:
            ip_address: IP地址
        Returns:
            脱敏后的IP地址
        """
        if not ip_address:
            return ip_address
        parts = ip_address.split('.')
        if len(parts) == 4:
            return f"{parts[0]}.{parts[1]}.{self.mask_char * 3}.{self.mask_char * 3}"
        return ip_address

增强版脱敏工具类

class AdvancedDataMasker(DataMasker):
    """
    高级脱敏工具类，增加更多功能和配置
    """
    def __init__(self, config: Optional[Dict] = None):
        """
        初始化高级脱敏工具
        Args:
            config: 脱敏配置字典
        """
        super().__init__(
            mask_char=config.get('mask_char', '*') if config else '*',
            keep_prefix=config.get('keep_prefix', 3) if config else 3,
            keep_suffix=config.get('keep_suffix', 4) if config else 4
        )
        self.config = config or {}
        self.custom_rules = {}
    def hash_mask(self, data: str, algorithm: str = 'sha256') -> str:
        """
        哈希脱敏：使用哈希算法进行不可逆脱敏
        Args:
            data: 待脱敏数据
            algorithm: 哈希算法（md5, sha1, sha256）
        Returns:
            哈希后的字符串
        """
        if not data:
            return data
        hash_funcs = {
            'md5': hashlib.md5,
            'sha1': hashlib.sha1,
            'sha256': hashlib.sha256
        }
        hash_func = hash_funcs.get(algorithm, hashlib.sha256)
        return hash_func(data.encode()).hexdigest()
    def random_mask(self, data: str, keep_length: bool = True) -> str:
        """
        随机替换脱敏
        Args:
            data: 待脱敏数据
            keep_length: 是否保持原长度
        Returns:
            随机替换后的字符串
        """
        if not data:
            return data
        if keep_length:
            return ''.join(random.choice('abcdefghijklmnopqrstuvwxyz0123456789') 
                          for _ in range(len(data)))
        else:
            return self.mask_char * len(data)
    def date_mask(self, date_str: str, format: str = '%Y-%m-%d', 
                 offset_days: int = 0) -> str:
        """
        日期脱敏：偏移或部分隐藏
        Args:
            date_str: 日期字符串
            format: 日期格式
            offset_days: 日期偏移天数（0表示只隐藏部分）
        Returns:
            脱敏后的日期
        """
        if not date_str or len(date_str) < 4:
            return date_str
        if offset_days != 0:
            # 日期偏移
            try:
                date_obj = datetime.strptime(date_str, format)
                new_date = date_obj + timedelta(days=offset_days)
                return new_date.strftime(format)
            except:
                return date_str
        else:
            # 保留年份，隐藏月和日
            if len(date_str) >= 10:
                return date_str[:4] + '-' + self.mask_char * 2 + '-' + self.mask_char * 2
            return date_str
    def auto_detect_and_mask(self, data: str) -> str:
        """
        自动检测数据类型并脱敏
        Args:
            data: 待检测和脱敏的数据
        Returns:
            脱敏后的数据
        """
        # 定义检测规则和对应的脱敏方法
        rules = [
            (r'^1[3-9]\d{9}$', self.mask_phone),
            (r'^[\w\.-]+@[\w\.-]+\.\w+$', self.mask_email),
            (r'^\d{17}[\dXx]$', self.mask_id_card),
            (r'^\d{16,19}$', self.mask_bank_card),
        ]
        for pattern, mask_func in rules:
            if re.match(pattern, data):
                return mask_func(data)
        # 默认返回原数据
        return data
    def mask_dict(self, data_dict: Dict, fields: List[str]) -> Dict:
        """
        对字典中的指定字段进行脱敏
        Args:
            data_dict: 数据字典
            fields: 需要脱敏的字段列表
        Returns:
            脱敏后的字典
        """
        masked_dict = data_dict.copy()
        for field in fields:
            if field in masked_dict and masked_dict[field]:
                field_type = self._detect_field_type(field)
                if field_type:
                    mask_method = getattr(self, f'mask_{field_type}', None)
                    if mask_method:
                        masked_dict[field] = mask_method(masked_dict[field])
                else:
                    # 自动检测
                    masked_dict[field] = self.auto_detect_and_mask(masked_dict[field])
        return masked_dict
    def mask_json(self, json_str: str, fields: List[str]) -> str:
        """
        对JSON字符串中的指定字段进行脱敏
        Args:
            json_str: JSON字符串
            fields: 需要脱敏的字段列表
        Returns:
            脱敏后的JSON字符串
        """
        try:
            data = json.loads(json_str)
            masked_data = self.mask_dict(data, fields)
            return json.dumps(masked_data, ensure_ascii=False)
        except:
            return json_str
    def _detect_field_type(self, field_name: str) -> Optional[str]:
        """
        根据字段名检测数据类型
        Args:
            field_name: 字段名
        Returns:
            检测到的数据类型
        """
        field_mapping = {
            'phone': 'phone',
            'mobile': 'phone',
            'tel': 'phone',
            'email': 'email',
            'mail': 'email',
            'id_card': 'id_card',
            'idcard': 'id_card',
            'name': 'name',
            'username': 'name',
            'address': 'address',
            'bank_card': 'bank_card',
            'card_number': 'bank_card',
            'password': 'password',
            'pwd': 'password',
            'ip': 'ip',
            'ip_address': 'ip',
        }
        # 字段名匹配
        if field_name.lower() in field_mapping:
            return field_mapping[field_name.lower()]
        # 字段名包含关键词
        for keyword, field_type in field_mapping.items():
            if keyword in field_name.lower():
                return field_type
        return None
    def add_custom_rule(self, field_pattern: str, mask_method: callable):
        """
        添加自定义脱敏规则
        Args:
            field_pattern: 字段匹配模式（正则表达式）
            mask_method: 脱敏方法
        """
        self.custom_rules[field_pattern] = mask_method
    def batch_mask(self, data_list: List[Dict], fields: List[str]) -> List[Dict]:
        """
        批量脱敏
        Args:
            data_list: 数据列表
            fields: 需要脱敏的字段
        Returns:
            脱敏后的数据列表
        """
        return [self.mask_dict(item, fields) for item in data_list]

使用示例

def demo_usage():
    """脱敏工具使用示例"""
    # 基本使用
    masker = DataMasker()
    phone = "13812345678"
    email = "zhangsan@example.com"
    id_card = "110101199001011234"
    name = "张三丰"
    print("基本脱敏示例：")
    print(f"手机号: {masker.mask_phone(phone)}")           # 138****5678
    print(f"邮箱: {masker.mask_email(email)}")             # zh****@example.com
    print(f"身份证: {masker.mask_id_card(id_card)}")      # 110101********6789
    print(f"姓名: {masker.mask_name(name)}")               # 张**
    # 高级使用
    config = {
        'mask_char': '*',
        'keep_prefix': 3,
        'keep_suffix': 4
    }
    advanced_masker = AdvancedDataMasker(config)
    # 字典脱敏
    user_data = {
        'name': '李四',
        'phone': '13987654321',
        'email': 'lisi@test.com',
        'id_card': '320102198801012345'
    }
    print("\n字典脱敏示例：")
    masked_user = advanced_masker.mask_dict(user_data, ['name', 'phone', 'email', 'id_card'])
    print(f"原始数据: {user_data}")
    print(f"脱敏数据: {masked_user}")
    # JSON脱敏
    json_data = json.dumps(user_data, ensure_ascii=False)
    print(f"\nJSON脱敏示例：")
    print(f"脱敏前: {json_data}")
    print(f"脱敏后: {advanced_masker.mask_json(json_data, ['phone', 'email'])}")
    # 哈希脱敏
    print(f"\n哈希脱敏示例：")
    print(f"MD5: {advanced_masker.hash_mask('password123', 'md5')}")
    print(f"SHA256: {advanced_masker.hash_mask('password123', 'sha256')}")
    # 批量脱敏
    user_list = [
        {'name': '张三', 'phone': '13812345678'},
        {'name': '李四', 'phone': '13987654321'},
        {'name': '王五', 'phone': '13711112222'}
    ]
    print(f"\n批量脱敏示例：")
    masked_list = advanced_masker.batch_mask(user_list, ['phone'])
    for original, masked in zip(user_list, masked_list):
        print(f"{original['name']}: {original['phone']} -> {masked['phone']}")
if __name__ == "__main__":
    demo_usage()

配置化脱敏封装

class ConfigurableMasker:
    """
    可配置的脱敏管理器
    """
    def __init__(self, config_file: Optional[str] = None):
        """
        初始化配置脱敏管理器
        Args:
            config_file: 配置文件路径（JSON格式）
        """
        self.masker = AdvancedDataMasker()
        self.rules = {}
        if config_file:
            self.load_config(config_file)
    def load_config(self, config_file: str):
        """
        从配置文件加载脱敏规则
        Args:
            config_file: 配置文件路径
        """
        try:
            with open(config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
            for rule in config.get('rules', []):
                field = rule.get('field')
                method = rule.get('method')
                params = rule.get('params', {})
                if field and method:
                    self.rules[field] = {
                        'method': method,
                        'params': params
                    }
        except Exception as e:
            print(f"加载配置文件失败: {e}")
    def process_data(self, data: Dict) -> Dict:
        """
        根据配置处理数据
        Args:
            data: 待处理数据
        Returns:
            处理后的数据
        """
        result = data.copy()
        for field, rule in self.rules.items():
            if field in result and result[field]:
                method_name = rule['method']
                params = rule['params']
                # 获取脱敏方法
                mask_method = getattr(self.masker, method_name, None)
                if mask_method:
                    try:
                        result[field] = mask_method(result[field], **params)
                    except:
                        result[field] = mask_method(result[field])
        return result
    def save_config(self, config_file: str):
        """
        保存配置到文件
        Args:
            config_file: 配置文件路径
        """
        config = {
            'rules': [
                {
                    'field': field,
                    'method': rule['method'],
                    'params': rule['params']
                }
                for field, rule in self.rules.items()
            ]
        }
        with open(config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, ensure_ascii=False, indent=2)
# 配置文件示例 (mask_config.json)
"""
{
    "rules": [
        {
            "field": "phone",
            "method": "mask_phone",
            "params": {}
        },
        {
            "field": "email",
            "method": "mask_email",
            "params": {}
        },
        {
            "field": "password",
            "method": "hash_mask",
            "params": {"algorithm": "sha256"}
        }
    ]
}
"""

单元测试

import unittest
class TestDataMasker(unittest.TestCase):
    """脱敏工具单元测试"""
    def setUp(self):
        self.masker = DataMasker()
        self.advanced_masker = AdvancedDataMasker()
    def test_phone_mask(self):
        """测试手机号脱敏"""
        result = self.masker.mask_phone("13812345678")
        self.assertEqual(result, "138****5678")
        # 测试异常情况
        self.assertEqual(self.masker.mask_phone(""), "")
        self.assertEqual(self.masker.mask_phone("12345"), "12345")
    def test_email_mask(self):
        """测试邮箱脱敏"""
        result = self.masker.mask_email("zhangsan@example.com")
        self.assertIn("@example.com", result)
        self.assertTrue(result.startswith("zh****"))
        # 测试短用户名
        result = self.masker.mask_email("ab@test.com")
        self.assertEqual(result, "a*@test.com")
    def test_id_card_mask(self):
        """测试身份证脱敏"""
        result = self.masker.mask_id_card("110101199001011234")
        self.assertEqual(result[:6], "110101")
        self.assertEqual(result[-4:], "1234")
        self.assertTrue("********" in result)
    def test_dict_mask(self):
        """测试字典脱敏"""
        data = {
            'phone': '13812345678',
            'email': 'test@example.com'
        }
        result = self.advanced_masker.mask_dict(data, ['phone', 'email'])
        self.assertEqual(result['phone'], '138****5678')
        self.assertIn('@example.com', result['email'])

使用建议

1. 选择合适的脱敏策略：根据业务需求和数据敏感性选择合适的脱敏方法
2. 配置管理：使用配置文件管理脱敏规则，便于维护和修改
3. 性能优化：对于大量数据，考虑使用批量处理和缓存
4. 安全性：避免在生产环境中使用可逆的脱敏方法
5. 审计日志：记录脱敏操作的日志，便于追踪和审计

这个封装案例提供了完整的脱敏工具类,可以根据实际需求进行扩展和定制。