怎样实现数据掩码脚本

wen 实用脚本 2026-06-29 29

本文目录导读：

怎样实现数据掩码脚本

Python实现 - 通用数据掩码
批量数据处理脚本
数据库敏感数据掩码脚本
命令行工具实现
配置文件示例 (mask_config.json)
使用建议

Python实现 - 通用数据掩码

import re
import random
import string
class DataMasker:
    """数据掩码工具类"""
    @staticmethod
    def mask_email(email):
        """掩码邮箱地址"""
        parts = email.split('@')
        if len(parts) != 2:
            return email
        name, domain = parts
        # 只显示前两个字符和最后一个字符
        if len(name) > 2:
            masked_name = name[:2] + '*' * (len(name) - 2)
        else:
            masked_name = name[0] + '*'
        return f"{masked_name}@{domain}"
    @staticmethod
    def mask_phone(phone):
        """掩码手机号"""
        # 保留前3位和后4位
        if len(phone) >= 7:
            return phone[:3] + '****' + phone[-4:]
        return phone
    @staticmethod
    def mask_id_card(id_card):
        """掩码身份证号"""
        if len(id_card) >= 10:
            return id_card[:4] + '********' + id_card[-4:]
        return id_card
    @staticmethod
    def mask_name(name):
        """掩码姓名"""
        if len(name) == 0:
            return name
        # 保留第一个字，其他用*代替
        return name[0] + '*' * (len(name) - 1)
    @staticmethod
    def mask_card_number(card_num):
        """掩码银行卡号"""
        if len(card_num) >= 8:
            return card_num[:4] + '****' + card_num[-4:]
        return card_num
# 使用示例
masker = DataMasker()
print(masker.mask_email("example@email.com"))  # ex*****@email.com
print(masker.mask_phone("13800138000"))        # 138****8000
print(masker.mask_id_card("110101199001011234")) # 1101********1234
print(masker.mask_name("张三"))                 # 张*
print(masker.mask_card_number("6222021234567890")) # 6222****7890

批量数据处理脚本

import pandas as pd
import json
import csv
class BatchDataMasker:
    """批量数据掩码处理器"""
    def __init__(self, config_file=None):
        self.config = self._load_config(config_file) or {}
        self.masker = DataMasker()
    def _load_config(self, config_file):
        """加载掩码配置"""
        if config_file:
            with open(config_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return None
    def mask_dataframe(self, df, columns_config):
        """
        掩码DataFrame中的数据
        columns_config: {column_name: mask_type}
        mask_type: 'email', 'phone', 'id_card', 'name', 'card_number'
        """
        masked_df = df.copy()
        for column, mask_type in columns_config.items():
            if column in masked_df.columns:
                mask_method = getattr(self.masker, f'mask_{mask_type}')
                masked_df[column] = masked_df[column].apply(mask_method)
        return masked_df
    def process_csv(self, input_file, output_file, columns_config):
        """处理CSV文件"""
        df = pd.read_csv(input_file)
        masked_df = self.mask_dataframe(df, columns_config)
        masked_df.to_csv(output_file, index=False)
        print(f"处理完成，结果已保存到: {output_file}")
    def process_json(self, json_data, fields_config):
        """处理JSON数据"""
        if isinstance(json_data, str):
            json_data = json.loads(json_data)
        def mask_recursive(data, path=""):
            if isinstance(data, dict):
                for key, value in data.items():
                    current_path = f"{path}.{key}" if path else key
                    if key in fields_config:
                        mask_type = fields_config[key]
                        mask_method = getattr(self.masker, f'mask_{mask_type}')
                        data[key] = mask_method(str(value))
                    elif isinstance(value, (dict, list)):
                        mask_recursive(value, current_path)
            elif isinstance(data, list):
                for item in data:
                    mask_recursive(item, path)
        masked_data = json.loads(json.dumps(json_data))
        mask_recursive(masked_data)
        return masked_data
# 使用示例
batch_masker = BatchDataMasker()
# 处理CSV文件
columns_config = {
    'email': 'email',
    'phone': 'phone',
    'name': 'name',
    'id_card': 'id_card'
}
batch_masker.process_csv('input.csv', 'output_masked.csv', columns_config)
# 处理JSON数据
json_data = {
    "users": [
        {"name": "张三", "email": "zhangsan@email.com", "phone": "13800138000"},
        {"name": "李四", "email": "lisi@email.com", "phone": "13900139000"}
    ]
}
result = batch_masker.process_json(json_data, {'name': 'name', 'email': 'email', 'phone': 'phone'})
print(json.dumps(result, ensure_ascii=False, indent=2))

数据库敏感数据掩码脚本

import pymysql
from sqlalchemy import create_engine
import logging
class DatabaseDataMasker:
    """数据库数据掩码工具"""
    def __init__(self, db_config, mask_config):
        """
        db_config: 数据库连接配置
        mask_config: 掩码配置 {table: {column: mask_type}}
        """
        self.db_config = db_config
        self.mask_config = mask_config
        self.masker = DataMasker()
        self.logger = self._setup_logger()
    def _setup_logger(self):
        """设置日志"""
        logger = logging.getLogger('DataMasker')
        logger.setLevel(logging.INFO)
        handler = logging.FileHandler('data_masking.log')
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        handler.setFormatter(formatter)
        logger.addHandler(handler)
        return logger
    def connect(self):
        """连接数据库"""
        return pymysql.connect(**self.db_config)
    def mask_table_column(self, connection, table, column, mask_type):
        """掩码单个表的指定列"""
        try:
            with connection.cursor() as cursor:
                # 获取所有数据
                cursor.execute(f"SELECT id, {column} FROM {table}")
                rows = cursor.fetchall()
                # 逐行更新
                mask_method = getattr(self.masker, f'mask_{mask_type}')
                for row in rows:
                    row_id = row[0]
                    original_value = row[1]
                    masked_value = mask_method(str(original_value))
                    update_sql = f"UPDATE {table} SET {column}=%s WHERE id=%s"
                    cursor.execute(update_sql, (masked_value, row_id))
                connection.commit()
                self.logger.info(f"成功掩码 {table}.{column} - {len(rows)} 条记录")
        except Exception as e:
            connection.rollback()
            self.logger.error(f"掩码失败 {table}.{column}: {str(e)}")
            raise
    def run_masking(self):
        """执行所有掩码任务"""
        connection = self.connect()
        try:
            for table, columns in self.mask_config.items():
                for column, mask_type in columns.items():
                    self.mask_table_column(connection, table, column, mask_type)
            self.logger.info("所有数据掩码完成")
        finally:
            connection.close()
# 使用示例
db_config = {
    'host': 'localhost',
    'user': 'username',
    'password': 'password',
    'database': 'test_db',
    'charset': 'utf8mb4'
}
mask_config = {
    'users': {
        'email': 'email',
        'phone': 'phone',
        'name': 'name'
    },
    'orders': {
        'customer_name': 'name',
        'card_number': 'card_number'
    }
}
# 执行掩码
masker = DatabaseDataMasker(db_config, mask_config)
masker.run_masking()

命令行工具实现

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import argparse
import sys
import os
def create_cli():
    """创建命令行工具"""
    parser = argparse.ArgumentParser(description='数据掩码工具')
    parser.add_argument('input', help='输入文件路径')
    parser.add_argument('output', help='输出文件路径')
    parser.add_argument('--format', choices=['csv', 'json', 'txt'], 
                       default='csv', help='文件格式')
    parser.add_argument('--columns', nargs='+', 
                       help='需要掩码的列（格式：column:type）')
    parser.add_argument('--config', help='掩码配置文件（JSON格式）')
    return parser
def main():
    """主函数"""
    parser = create_cli()
    args = parser.parse_args()
    # 解析掩码配置
    columns_config = {}
    if args.columns:
        for col_config in args.columns:
            if ':' in col_config:
                column, mask_type = col_config.split(':')
                columns_config[column] = mask_type
    # 执行掩码
    batch_masker = BatchDataMasker(args.config)
    if args.format == 'csv':
        batch_masker.process_csv(args.input, args.output, columns_config)
    elif args.format == 'json':
        with open(args.input, 'r', encoding='utf-8') as f:
            data = json.load(f)
        masked_data = batch_masker.process_json(data, columns_config)
        with open(args.output, 'w', encoding='utf-8') as f:
            json.dump(masked_data, f, ensure_ascii=False, indent=2)
    print(f"数据掩码完成，结果已保存到: {args.output}")
if __name__ == '__main__':
    main()

配置文件示例 (mask_config.json)

{
    "email": {
        "mask_type": "partial",
        "prefix_length": 2,
        "suffix_length": 1,
        "mask_char": "*"
    },
    "phone": {
        "mask_type": "partial",
        "prefix_length": 3,
        "suffix_length": 4,
        "mask_char": "*"
    },
    "id_card": {
        "mask_type": "partial",
        "prefix_length": 4,
        "suffix_length": 4,
        "mask_char": "*"
    },
    "name": {
        "mask_type": "custom",
        "mask_char": "*"
    },
    "card_number": {
        "mask_type": "partial",
        "prefix_length": 4,
        "suffix_length": 4,
        "mask_char": "*"
    }
}

使用建议

安全性考虑：
- 在生产环境执行前先在测试环境验证
- 做好数据备份
- 记录详细的日志
性能优化：
- 大批量数据处理时使用批处理
- 考虑使用并行处理
- 数据库操作使用事务
不同类型的掩码：
- 静态掩码：固定替换规则
- 动态掩码：根据用户权限显示不同级别的掩码
- 一致性掩码：相同输入产生相同输出（用于测试数据关联）
合规要求：
- 确保符合数据保护法规（GDPR、CCPA等）
- 保留必要的审计日志
- 定期检查掩码效果

这个脚本可以满足大多数数据脱敏需求,你可以根据实际情况调整具体的掩码规则和实现方式。