行业资讯
影刀RPA 流程日志系统搭建：从print到结构化日志的进化
影刀RPA 流程日志系统搭建从print到结构化日志的进化作者林焱什么情况用这个大部分影刀新手的日志系统就是这样的每个Python节点里撒几行print()需要排查问题的时候要么打开影刀的运行日志一条条翻要么干脆找不到——因为print输出的东西运行完就没了。当你的流程从跑给自己看进化到每天自动跑、出问题要回溯的阶段一套规范的日志系统就是必需品——不是锦上添花是救命稻草。这篇文章教你从零搭建一套影刀流程的日志系统从最原始的print到结构化的文件日志再到多渠道告警。每个阶段都给你可以直接用的代码。怎么做第一步日志的四个等级——先搞清楚记录什么不是所有信息都值得记录。区分四个等级# 日志等级体系LEVELS{DEBUG:0,# 调试信息当前URL是xxx跳过了第3条INFO:1,# 关键节点开始采集第2页、共采集50条数据WARNING:2,# 异常但可恢复第5条数据格式异常已跳过ERROR:3,# 出错需要关注登录失败、API超时}每个等级记录什么等级内容谁看例子DEBUG过程中的状态变更开发者调试时“元素已找到classbtn-primary”INFO关键步骤的完成和结果日常巡检“第3页采集完成50条”店群矩阵自动化突破运营极限| WARNING | 异常但流程能继续 | 需要定期检查 | “第12行金额为空已跳过” || ERROR | 流程中断或关键失败 | 需要立即处理 | “数据库连接失败” |第二步从print到文件日志最简单的进化把print改成写入文件。fromdatetimeimportdatetimeimportosclassSimpleLogger:最简单的文件日志器def__init__(self,log_dirD:/rpa_logs,process_namedefault):self.process_nameprocess_name self.log_dirlog_dir os.makedirs(log_dir,exist_okTrue)# 每次运行创建一个日志文件timestampdatetime.now().strftime(%Y%m%d_%H%M%S)self.log_fileos.path.join(log_dir,f{process_name}_{timestamp}.log)def_write(self,level,message):写入日志nowdatetime.now().strftime(%Y-%m-%d %H:%M:%S)linef[{now}] [{level}]{message}\nwithopen(self.log_file,a,encodingutf-8)asf:f.write(line)# 同时输出到控制台影刀的运行日志中也能看到print(f[{level}]{message})defdebug(self,msg):self._write(DEBUG,msg)definfo(self,msg):self._write(INFO,msg)defwarning(self,msg):self._write(WARN,msg)deferror(self,msg):self._write(ERROR,msg)# 在流程中使用 # 在流程的第一个Python节点中初始化loggerSimpleLogger(process_name每日销售数据采集)# 在后续Python节点中使用logger.info(流程启动)logger.info(f开始采集第{page}页)logger.warning(f第{idx}条数据价格为空已跳过)logger.error(f采集失败:{str(e)})logger.info(流程完成共采集386条数据)第三步结构化日志——能检索、能统计纯文本日志只能看结构化日志能查importjsonclassStructuredLogger:结构化日志器JSON格式def__init__(self,log_dirD:/rpa_logs,process_namedefault):self.process_nameprocess_name self.log_dirlog_dir os.makedirs(log_dir,exist_okTrue)timestampdatetime.now().strftime(%Y%m%d_%H%M%S)self.log_fileos.path.join(log_dir,f{process_name}_{timestamp}.jsonl)# JSONL格式每行一个JSON对象def_log(self,level,message,**extra):写入结构化日志entry{timestamp:datetime.now().strftime(%Y-%m-%d %H:%M:%S),level:level,process:self.process_name,message:message,**extra# 额外字段}withopen(self.log_file,a,encodingutf-8)asf:f.write(json.dumps(entry,ensure_asciiFalse)\n)definfo(self,msg,**extra):self._log(INFO,msg,**extra)defwarning(self,msg,**extra):self._log(WARN,msg,**extra)deferror(self,msg,**extra):self._log(ERROR,msg,**extra)# 使用示例 loggerStructuredLogger(process_name销售数据采集)# 记录关键指标logger.info(开始采集,page1,urlhttps://example.com/list)logger.info(采集完成,page1,count50,duration_seconds3.2)logger.warning(数据异常,row12,field金额,value,reason为空)logger.error(API超时,urlhttps://api.example.com/data,timeout10,retry3)结构化日志的价值在于你可以用Python快速分析# 分析日志文件importjsondefanalyze_log(log_file):分析日志文件——统计错误、计算耗时errors[]warnings[]total_records0withopen(log_file,r,encodingutf-8)asf:forlineinf:entryjson.loads(line)total_records1ifentry[level]ERROR:errors.append(entry)elifentry[level]WARN:warnings.append(entry)print(f总日志条数:{total_records})print(f警告数:{len(warnings)})print(f错误数:{len(errors)})iferrors:print(\n错误详情:)foreinerrors:print(f - [{e[timestamp]}]{e[message]})第四步日志分级输出——开发时看详细生产时看关键classLogManager:分级日志管理器def__init__(self,envdev,log_dirD:/rpa_logs):self.envenv self.file_loggerStructuredLogger(log_dirlog_dir)# 不同环境的日志级别self.level_map{dev:0,# 记录所有test:1,# 记录INFO及以上prod:2,# 只记录WARNING和ERROR}self.min_levelself.level_map.get(env,1)self.level_values{DEBUG:0,INFO:1,WARN:2,ERROR:3}defshould_log(self,level):判断当前级别是否需要记录returnself.level_values.get(level,0)self.min_leveldefdebug(self,msg,**extra):ifself.should_log(DEBUG):self.file_logger._log(DEBUG,msg,**extra)definfo(self,msg,**extra):ifself.should_log(INFO):self.file_logger._log(INFO,msg,**extra)defwarning(self,msg,**extra):ifself.should_log(WARN):self.file_logger._log(WARN,msg,**extra)deferror(self,msg,**extra):ifself.should_log(ERROR):self.file_logger._log(ERROR,msg,**extra)# 生产环境的错误额外发通知ifself.envprod:self._send_alert(msg,extra)# 开发环境详细日志loggerLogManager(envdev)logger.debug(准备点击第3个按钮)# 生产环境只看关键loggerLogManager(envprod)logger.info(日常采集完成)# 记logger.debug(循环到了第15次)# 不记级别太低第五步异常自动记录在try-catch中自动记录错误上下文importtracebackdeflog_error_with_context(logger,func_name,**context):在异常处理中记录完整上下文error_infotraceback.format_exc()logger.error(f{func_name}执行失败,errorerror_info,**context)# 在流程中使用defprocess_order(order_id):try:# 处理订单的逻辑passexceptExceptionase:log_error_with_context(logger,process_order,order_idorder_id,current_pagecurrent_page)# 决定是重试还是跳过raise第六步日志的轮转与清理importglobdefcleanup_old_logs(log_dir,keep_days7):清理过期日志cutoffdatetime.now().timestamp()-(keep_days*86400)forlog_fileinglob.glob(os.path.join(log_dir,*.jsonl)):ifos.path.getmtime(log_file)cutoff:os.remove(log_file)print(f已删除过期日志:{log_file})forlog_fileinglob.glob(os.path.join(log_dir,*.log)):ifos.path.getmtime(log_file)cutoff:os.remove(log_file)# 在每次流程启动时执行一次清理cleanup_old_logs(D:/rpa_logs,keep_days7)有什么坑坑一生产环境打DEBUG日志撑爆磁盘现象用了半个月C盘红了——一看日志文件夹50GB。原因每条循环都打了DEBUG日志一天几万条。解决生产环境只用INFO及以上级别。给日志目录设大小限制。坑二日志文件名为中文导致编码问题现象日志文件名是销售数据采集_20250315.log在某些系统上无法读取。temu店群自动化报活动案例解决文件名只用英文数字下划线process_namesales_collector# 不用销售数据采集坑三不区分业务日志和系统日志现象所有的信息都混在一个日志里排查问题时在几千行日志中大海捞针。解决分离关注点——业务日志记录采集了多少数据系统日志记录操作是否成功logger_bizStructuredLogger(process_namebiz)# 业务日志logger_sysStructuredLogger(process_namesys)# 系统日志logger_sys.info(页面加载完成,urlurl,duration1.5)logger_biz.info(采集完成,count50,total_amount12500)坑四只在Python节点中打日志现象Python节点中的逻辑有日志但影刀内置指令的操作没有日志——这个【点击】到底点没点不知道。解决在影刀关键指令前后用【打印日志】指令【打印日志】→ 开始处理第N个商品 【点击元素】→ 详情链接 【打印日志】→ 已打开详情页 【等待元素】→ 商品标题 【打印日志】→ 详情页加载完成坑五错误日志不记录输入数据现象日志里只有处理失败不知道处理的是什么数据导致失败。解决错误日志必须带上当时的输入数据try:process_item(item_data)exceptExceptionase:logger.error(处理失败,itemitem_data,# ← 关键记录出错的数据errorstr(e),tracebacktraceback.format_exc())总结日志系统不需要一开始就很复杂。分三步走第一步把print改成写文件每条带时间戳第二步改成JSON格式方便检索第三步加上分级和环境切换。即使在第二步停住你的流程维护成本也已经大幅降低了。记住日志存在的唯一价值是出问题的时候帮你快速定位——能让你10分钟内找到原因的日志就是好日志。
郑州网站建设
网页设计
企业官网
