影刀RPA 流程输入验证防御性编程的第一道防线作者林焱什么情况用这个你的流程跑了一个月都没事突然某天崩溃了。排查半天发现Excel里的金额列不知什么时候多了个元字原来的数字199.00变成了199.00元。于是所有计算全崩。这不是你流程逻辑的Bug是你没有在数据入口做好验证。防御性编程的核心思想是不信任任何外部输入。输入的Excel、网页数据、API返回、用户参数——所有进入流程的数据在使用的第一行就要验证。如果有问题第一时间发现并拒绝而不是让它跑完整条链路再报一个无头无脑的错误。怎么做第一步输入验证的三个层次层次1格式验证 → 数据是预期的格式吗类型、长度、范围 层次2逻辑验证 → 数据在业务上合理吗金额不能为负、日期不能是未来 层次3关联验证 → 多条数据之间的关系对吗合计明细之和原则越早验证损失越小。一个数据在流程入口被拒绝比跑了半小时处理后再崩溃要好100倍。第二步格式验证——最基础也最重要拼多多店群自动化报活动上架classInputValidator:输入验证器staticmethoddefvalidate_string(value,field_name,min_len1,max_len500,allow_emptyFalse):验证字符串ifvalueisNone:ifallow_empty:returnTrue,returnFalse,f{field_name}不能为空valuestr(value).strip()ifnotallow_emptyandvalue:returnFalse,f{field_name}不能为空字符串iflen(value)min_len:returnFalse,f{field_name}长度至少{min_len}当前{len(value)}iflen(value)max_len:returnFalse,f{field_name}长度最多{max_len}当前{len(value)}returnTrue,valuestaticmethoddefvalidate_number(value,field_name,min_valueNone,max_valueNone):验证数字try:numfloat(str(value).replace(,,).replace(¥,).strip())except(ValueError,TypeError):returnFalse,f{field_name}不是有效数字:{value}ifmin_valueisnotNoneandnummin_value:returnFalse,f{field_name}不能小于{min_value}当前{num}ifmax_valueisnotNoneandnummax_value:returnFalse,f{field_name}不能大于{max_value}当前{num}returnTrue,numstaticmethoddefvalidate_date(value,field_name,allow_futureFalse):验证日期ifnotvalue:returnFalse,f{field_name}为空fromdatetimeimportdatetime date_strstr(value).strip()formats[%Y-%m-%d,%Y/%m/%d,%Y%m%d,%Y年%m月%d日]forfmtinformats:try:dtdatetime.strptime(date_str,fmt)ifnotallow_futureanddtdatetime.now():returnFalse,f{field_name}不能是未来日期:{date_str}returnTrue,dt.strftime(%Y-%m-%d)exceptValueError:continuereturnFalse,f{field_name}格式无效:{date_str}接受格式: YYYY-MM-DDstaticmethoddefvalidate_email(value,field_name):验证邮箱importre patternr^[a-zA-Z0-9._%-][a-zA-Z0-9.-]\.[a-zA-Z]{2,}$ifnotvalue:returnFalse,f{field_name}为空ifre.match(pattern,str(value)):returnTrue,str(value)returnFalse,f{field_name}不是有效邮箱:{value}staticmethoddefvalidate_phone(value,field_name):验证手机号中国大陆importre patternr^1[3-9]\d{9}$ifnotvalue:returnFalse,f{field_name}为空phonestr(value).strip()ifre.match(pattern,phone):returnTrue,phonereturnFalse,f{field_name}不是有效手机号:{value}# 使用示例 validatorInputValidator()# 批量验证一条数据的所有字段defvalidate_order_record(record,row_num):验证订单记录的完整性errors[]# 订单号ok,valvalidator.validate_string(record.get(订单号),订单号,min_len5)ifnotok:errors.append(f第{row_num}行{val})else:record[订单号]val# 金额ok,valvalidator.validate_number(record.get(金额),金额,min_value0)ifnotok:errors.append(f第{row_num}行{val})else:record[金额]val# 日期ok,valvalidator.validate_date(record.get(日期),日期)ifnotok:errors.append(f第{row_num}行{val})else:record[日期]val# 手机号ok,valvalidator.validate_phone(record.get(手机号),手机号)ifnotok:errors.append(f第{row_num}行{val})else:record[手机号]valreturnerrors# 空列表 验证通过第三步批量数据验证——一条不对整批停defvalidate_batch(records,rules): 批量数据验证 rules: {字段名: 验证函数} 返回: (clean_records, error_report) clean_records[]error_report[]forrow_num,recordinenumerate(records,start1):row_errors[]clean_record{}forfield,validate_funcinrules.items():valuerecord.get(field)ok,resultvalidate_func(value,field)ifnotok:row_errors.append(result)else:clean_record[field]resultifrow_errors:error_report.append({row:row_num,errors:row_errors,original:record})else:clean_records.append(clean_record)# 汇总报告print(f验证结果:{len(clean_records)}条通过,{len(error_report)}条异常)iferror_report:print(\n异常记录:)forerrinerror_report[:5]:# 只显示前5条print(f 第{err[row]}行:{; .join(err[errors])})returnclean_records,error_report# 使用rules{订单号:lambdav,f:InputValidator.validate_string(v,f,min_len5),金额:lambdav,f:InputValidator.validate_number(v,f,min_value0),日期:lambdav,f:InputValidator.validate_date(v,f),}clean_data,errorsvalidate_batch(raw_data,rules)iferrors:# 有错误是继续处理clean_data还是全部中止error_ratelen(errors)/len(raw_data)iferror_rate0.05:# 错误率超过5%全部中止raiseException(f数据错误率过高({error_rate:.1%})终止处理)第四步业务逻辑验证格式对了不代表数据合理classBusinessValidator:业务逻辑验证staticmethoddefvalidate_total_vs_detail(total,details):验证合计等于明细之和detail_sumsum(float(d.get(金额,0))fordindetails)ifabs(float(total)-detail_sum)0.01:# 容忍1分钱误差returnFalse,f合计{total}与明细之和{detail_sum}不匹配returnTrue,Nonestaticmethoddefvalidate_date_range(start_date,end_date):验证日期范围的合理性fromdatetimeimportdatetime startdatetime.strptime(start_date,%Y-%m-%d)enddatetime.strptime(end_date,%Y-%m-%d)ifstartend:returnFalse,开始日期不能晚于结束日期if(end-start).days365:returnFalse,日期范围不能超过一年returnTrue,Nonestaticmethoddefvalidate_no_duplicate(items,key_field):验证无重复主键唯一seenset()duplicates[]foriteminitems:keyitem.get(key_field)ifkeyinseen:duplicates.append(key)seen.add(key)ifduplicates:returnFalse,f发现重复的{key_field}:{duplicates[:5]}returnTrue,None第五步流程入口的统一验证在流程的第一个Python节点中做入口验证# 影刀流程入口验证节点 classFlowEntryValidator:流程入口验证器——流程启动时执行一次staticmethoddefvalidate_all():errors[]# 1. 检查必需的配置文件required_files[D:/rpa_config/config.json,D:/data/source.xlsx]forfinrequired_files:ifnotos.path.exists(f):errors.append(f缺少文件:{f})# 2. 检查磁盘空间importshutil free_gbshutil.disk_usage(D:/).free/(1024**3)iffree_gb1:# 少于1GBerrors.append(f磁盘空间不足:{free_gb:.1f}GB)# 3. 检查网络连接try:importrequests requests.get(https://www.baidu.com,timeout5)except:errors.append(网络连接异常)# 4. 检查输入数据ifos.path.exists(D:/data/source.xlsx):importopenpyxl wbopenpyxl.load_workbook(D:/data/source.xlsx)wswb.activeifws.max_row1:errors.append(源数据为空仅有表头)iferrors:error_msg流程启动验证失败:\n\n.join(f -{e}foreinerrors)raiseException(error_msg)print(✓ 流程启动验证通过)returnTrue# 执行验证FlowEntryValidator.validate_all()有什么坑坑一验证太严格导致正常数据被拒绝现象金额字段可能带元或¥你的验证把它当错误拒绝了但其实去除符号后数据是对的。解决验证时先做数据清洗把合法但不标准的值转换后再判断defclean_and_validate_amount(raw_value):先清洗再验证# 去掉货币符号、中文单位cleanedstr(raw_value).replace(¥,).replace(元,).replace(,,).strip()try:returnfloat(cleaned)exceptValueError:raiseValueError(f无法格式化为金额:{raw_value})坑二验证逻辑铺满每个Python节点现象每个Python节点都写了一遍金额验证、日期验证。解决把验证函数封装到一个公共模块每个节点引用即可。或者只在流程入口做一次集中验证。坑三验证通过后的数据没有再做类型保障现象入口验证通过了但后续节点中数据类型还是乱比如金额还是字符串。TEMU店群矩阵自动化运营核价报活动解决验证通过的同时做类型转换返回的是转换后的值ok,valuevalidator.validate_number(199.00,金额)# okTrue, value199.0已经是float不只是验证通过坑四异常数据被跳过后不通知人现象验证发现了异常数据流程默默跳过了没人知道数据不完整。解决确定一个告警阈值。跳过率超过阈值时一定要通知skip_ratelen(errors)/len(total)ifskip_rate0.02:# 超过2%send_alert(f数据异常率{skip_rate:.1%}请检查,errors[:10])坑五忽略了数据为0和None的区别现象金额为0免费商品被当成了无效数据拒绝了。解决区分这个值不存在和这个值就是0defvalidate_amount_with_null(value):ifvalueisNoneorstr(value).strip():returnFalse,金额为空# 真的是空的try:amountfloat(str(value).replace(,,))ifamount0:returnFalse,金额不能为负returnTrue,amount# 0也是合法值except:returnFalse,f无效金额:{value}总结输入验证是写起来最枯燥、出事时最救命的代码。不要在流程的各处零散地验证——在流程入口集中做一次数据体检不合格的数据要么拒绝、要么清洗、要么标记。这会让你后续几百行代码都不用再担心数据质量问题。