影刀RPA 数据脱敏与隐私保护敏感信息处理作者林焱什么情况用你的影刀流程采集了大量用户数据——手机号、身份证号、邮箱、银行卡号——需要导出给第三方分析但直接导出违反隐私法规你想在测试环境中使用真实数据但又不能暴露敏感信息需要把日志中的密码和密钥脱敏后才能记录数据脱敏是数据处理中不可忽视的一环。无论是个人信息保护法PIPL还是企业内部合规要求敏感数据在存储、传输、展示时都需要脱敏处理。本文讲清楚常见敏感数据的脱敏规则、在影刀RPA中怎么实现批量脱敏、脱敏后怎么保持数据可用性。怎么做拼多多店群自动化上架方案一、常见脱敏规则importreimporthashlibclassDataMasker:数据脱敏工具staticmethoddefmask_phone(phone):手机号脱敏138****1234phonestr(phone).strip()iflen(phone)11:returnphone[:3]****phone[7:]returnphonestaticmethoddefmask_id_card(id_card):身份证号脱敏110101********1234id_cardstr(id_card).strip()iflen(id_card)18:returnid_card[:6]********id_card[14:]eliflen(id_card)15:returnid_card[:6]******id_card[12:]returnid_cardstaticmethoddefmask_email(email):邮箱脱敏z***example.comemailstr(email).strip()ifinemail:name,domainemail.split(,1)iflen(name)1:masked_namenameeliflen(name)3:masked_namename[0]**(len(name)-1)else:masked_namename[0]***name[-1]returnf{masked_name}{domain}returnemailstaticmethoddefmask_bank_card(card):银行卡号脱敏6222****1234cardstr(card).replace( ,).strip()iflen(card)12:returncard[:4]****card[-4:]returncardstaticmethoddefmask_name(name):姓名脱敏张*三 / 欧阳**namestr(name).strip()iflen(name)1:returnnameeliflen(name)2:returnname[0]*eliflen(name)3:returnname[0]*name[2]eliflen(name)4:returnname[:2]**else:returnname[0]**(len(name)-2)name[-1]staticmethoddefmask_address(address):地址脱敏保留省市详细地址用*替代addressstr(address).strip()# 保留前6个字符通常包含省市后面脱敏iflen(address)6:returnaddressreturnaddress[:6]**min(len(address)-6,20)staticmethoddefmask_password(password):密码脱敏全部替换为*return**len(str(password))staticmethoddefhash_value(value,algorithmmd5):哈希脱敏不可逆valuestr(value).encode(utf-8)ifalgorithmmd5:returnhashlib.md5(value).hexdigest()elifalgorithmsha256:returnhashlib.sha256(value).hexdigest()else:returnhashlib.md5(value).hexdigest()staticmethoddefmask_custom(value,keep_start0,keep_end0,mask_char*):自定义脱敏保留前N位和后N位中间用*替代valuestr(value).strip()iflen(value)keep_startkeep_end:returnmask_char*len(value)masked_lengthlen(value)-keep_start-keep_endreturnvalue[:keep_start]mask_char*masked_lengthvalue[-keep_end:]ifkeep_end0elsevalue[:keep_start]mask_char*masked_length# 使用示例mDataMasker()print(m.mask_phone(13812345678))# 138****5678print(m.mask_id_card(110101199001011234))# 110101********1234print(m.mask_email(zhangsanexample.com))# z***example.comprint(m.mask_bank_card(6222021234567890123))# 6222****0123print(m.mask_name(张三丰))# 张*丰print(m.mask_name(欧阳修))# 欧** (3字名)print(m.mask_address(北京市朝阳区建国路88号))# 北京市朝阳区*************print(m.mask_password(mypassword123))# ************print(m.hash_value(13812345678))# 哈希值二、Excel批量脱敏importopenpyxlimportosdefmask_excel_file(input_file,output_file,mask_config): 批量脱敏Excel文件 mask_config: {列名: 脱敏类型} 字典 示例: {手机号: phone, 身份证: id_card, 邮箱: email} maskerDataMasker()# 脱敏函数映射mask_functions{phone:masker.mask_phone,id_card:masker.mask_id_card,email:masker.mask_email,bank_card:masker.mask_bank_card,name:masker.mask_name,address:masker.mask_address,password:masker.mask_password,hash:masker.hash_value,}wbopenpyxl.load_workbook(input_file)forsheet_nameinwb.sheetnames:wswb[sheet_name]# 获取表头headers{}forcolinrange(1,ws.max_column1):headerws.cell(row1,columncol).valueifheader:headers[str(header).strip()]col# 找到需要脱敏的列mask_columns{}forcol_name,mask_typeinmask_config.items():ifcol_nameinheaders:mask_columns[headers[col_name]]mask_typeprint(f Sheet {sheet_name}: 列{col_name} →{mask_type}脱敏)# 逐行脱敏forrowinrange(2,ws.max_row1):forcol,mask_typeinmask_columns.items():cellws.cell(rowrow,columncol)ifcell.value:mask_funcmask_functions.get(mask_type)ifmask_func:cell.valuemask_func(str(cell.value))# 保存脱敏后的文件wb.save(output_file)print(f脱敏完成:{input_file}→{output_file})# 使用示例mask_config{手机号:phone,身份证号:id_card,邮箱:email,银行卡号:bank_card,客户姓名:name,家庭住址:address,}mask_excel_file(data/customers.xlsx,data/customers_masked.xlsx,mask_config)三、CSV批量脱敏importcsvimportosdefmask_csv_file(input_file,output_file,mask_config,encodingutf-8):批量脱敏CSV文件maskerDataMasker()mask_functions{phone:masker.mask_phone,id_card:masker.mask_id_card,email:masker.mask_email,bank_card:masker.mask_bank_card,name:masker.mask_name,address:masker.mask_address,password:masker.mask_password,}withopen(input_file,r,encodingencoding,newline)asfin:readercsv.DictReader(fin)# 确定需要脱敏的字段mask_fields{f:tforf,tinmask_config.items()iffinreader.fieldnames}withopen(output_file,w,encodingencoding,newline)asfout:writercsv.DictWriter(fout,fieldnamesreader.fieldnames)writer.writeheader()row_count0forrowinreader:forfield,mask_typeinmask_fields.items():ifrow[field]:mask_funcmask_functions.get(mask_type)ifmask_func:row[field]mask_func(row[field])writer.writerow(row)row_count1print(f脱敏完成:{row_count}行,{input_file}→{output_file})# 使用mask_csv_file(data/users.csv,data/users_masked.csv,{phone:phone,email:email,name:name})四、日志脱敏importreimportloggingclassSensitiveDataFilter(logging.Filter):日志敏感信息过滤器# 敏感数据正则模式PATTERNS{phone:(re.compile(r1[3-9]\d{9}),lambdam:m.group()[:3]****m.group()[7:]),id_card:(re.compile(r\d{17}[\dXx]),lambdam:m.group()[:6]********m.group()[14:]),email:(re.compile(r[\w.][\w.]),lambdam:m.group()[0]***m.group().split()[1]),bank_card:(re.compile(r\d{16,19}),lambdam:m.group()[:4]****m.group()[-4:]),}deffilter(self,record):msgstr(record.getMessage())forname,(pattern,replacer)inself.PATTERNS.items():msgpattern.sub(replacer,msg)record.msgmsgreturnTrue# 使用loggerlogging.getLogger(rpa)logger.setLevel(logging.DEBUG)handlerlogging.StreamHandler()handler.addFilter(SensitiveDataFilter())logger.addHandler(handler)# 日志中的敏感信息会自动脱敏logger.info(f用户手机号13812345678登录成功)# 输出: 用户手机号138****5678登录成功logger.info(f身份证号110101199001011234已验证)# 输出: 身份证号110101********1234已验证五、可逆脱敏加密脱敏importhashlibimportbase64fromcryptography.fernetimportFernetimportjsonclassReversibleMasker:可逆脱敏加密解密def__init__(self,keyNone):ifkeyisNone:keyFernet.generate_key()self.keykey self.cipherFernet(key)defencrypt(self,value):加密脱敏value_bytesstr(value).encode(utf-8)encryptedself.cipher.encrypt(value_bytes)returnencrypted.decode(utf-8)defdecrypt(self,encrypted_value):解密还原encrypted_bytesencrypted_value.encode(utf-8)decryptedself.cipher.decrypt(encrypted_bytes)returndecrypted.decode(utf-8)defencrypt_dict(self,data,fields):加密字典中的指定字段resultdata.copy()forfieldinfields:iffieldinresult:result[field]self.encrypt(result[field])returnresultdefdecrypt_dict(self,data,fields):解密字典中的指定字段resultdata.copy()forfieldinfields:iffieldinresult:try:result[field]self.decrypt(result[field])except:pass# 解密失败保持原值returnresult# 使用maskerReversibleMasker(keybyour-secret-key-here-32-bytes!!!)# 原始数据user_data{name:张三,phone:13812345678,id_card:110101199001011234}# 加密脱敏masked_datamasker.encrypt_dict(user_data,[phone,id_card])print(f脱敏后:{masked_data})# {name: 张三, phone: gAAAAABm..., id_card: gAAAAABm...}# 解密还原original_datamasker.decrypt_dict(masked_data,[phone,id_card])print(f还原后:{original_data})# {name: 张三, phone: 13812345678, id_card: 110101199001011234}有什么坑坑1脱敏不彻底导致信息泄露现象手机号脱敏了但日志里还出现了完整号码。或者Excel中某列漏配了脱敏规则。解决建立脱敏配置清单——列出所有包含敏感信息的字段及其脱敏规则。脱敏后做验证——搜索脱敏后的数据中是否还有未脱敏的敏感信息模式如11位数字、18位身份证号。日志中统一使用敏感数据过滤器。TEMU店群如何管理运营坑2脱敏后数据无法用于开发测试现象手机号脱敏成138****5678后开发测试时无法发送短信验证码。解决对开发测试环境使用假数据替换而非遮盖脱敏。把真实手机号替换为测试号码如13800000000。保留数据格式但替换为假值——保持11位手机号格式但使用假号码。用Faker库生成逼真的假数据。坑3脱敏规则不一致现象同一个手机号在不同报表中脱敏方式不同——有的显示138****5678有的显示138***5678。解决统一脱敏规则封装成工具类如上面的DataMasker。所有流程都调用同一个脱敏工具不各自实现。建立脱敏规则文档明确每种数据类型的脱敏格式。坑4哈希脱敏后的碰撞问题现象用MD5哈希手机号不同手机号可能产生相同的哈希值碰撞导致数据关联错误。解决MD5碰撞概率极低实际中几乎不会出问题。但为了更安全用SHA256代替MD5。哈希时加salt随机盐值防止彩虹表攻击hashlib.sha256((value salt).encode()).hexdigest()。不同字段的salt不同。坑5可逆脱敏的密钥管理现象用加密做可逆脱敏但密钥硬编码在代码中泄露了等于没脱敏。解决密钥不能硬编码。通过环境变量传入key os.environ.get(MASK_KEY)。或者用密钥管理服务KMS。密钥和加密数据分开存储。定期轮换密钥。开发、测试、生产环境使用不同的密钥。如果不需要还原优先使用不可逆脱敏遮盖或哈希。