影刀RPA 流程日志系统搭建：从print到结构化日志的进化

影刀RPA 流程日志系统搭建：从print到结构化日志的进化

作者：林焱

什么情况用这个

大部分影刀新手的日志系统就是这样的：每个Python节点里撒几行print()，需要排查问题的时候，要么打开影刀的运行日志一条条翻，要么干脆找不到——因为print输出的东西运行完就没了。

当你的流程从"跑给自己看"进化到"每天自动跑、出问题要回溯"的阶段，一套规范的日志系统就是必需品——不是锦上添花，是救命稻草。

这篇文章教你从零搭建一套影刀流程的日志系统，从最原始的print到结构化的文件日志再到多渠道告警。每个阶段都给你可以直接用的代码。

怎么做

第一步：日志的四个等级——先搞清楚记录什么

不是所有信息都值得记录。区分四个等级：

# 日志等级体系LEVELS={"DEBUG":0,# 调试信息："当前URL是xxx"，"跳过了第3条""INFO":1,# 关键节点："开始采集第2页"、"共采集50条数据""WARNING":2,# 异常但可恢复："第5条数据格式异常，已跳过"![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/7e9b10b212134fee8a4360fd6f88e334.png#pic_center)"ERROR":3,# 出错需要关注："登录失败"、"API超时"}

每个等级记录什么：

| WARNING | 异常但流程能继续 | 需要定期检查 | “第12行金额为空，已跳过” |
| ERROR | 流程中断或关键失败 | 需要立即处理 | “数据库连接失败” |

第二步：从print到文件日志

最简单的进化：把print改成写入文件。

fromdatetimeimportdatetimeimportosclassSimpleLogger:"""最简单的文件日志器"""def__init__(self,log_dir="D:/rpa_logs",process_name="default"):self.process_name=process_name self.log_dir=log_dir os.makedirs(log_dir,exist_ok=True)# 每次运行创建一个日志文件timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")self.log_file=os.path.join(log_dir,f"{process_name}_{timestamp}.log")def_write(self,level,message):"""写入日志"""now=datetime.now().strftime("%Y-%m-%d %H:%M:%S")line=f"[{now}] [{level}]{message}\n"withopen(self.log_file,"a",encoding="utf-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节点中初始化logger=SimpleLogger(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_dir="D:/rpa_logs",process_name="default"):self.process_name=process_name self.log_dir=log_dir os.makedirs(log_dir,exist_ok=True)timestamp=datetime.now().strftime("%Y%m%d_%H%M%S")self.log_file=os.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",encoding="utf-8")asf:f.write(json.dumps(entry,ensure_ascii=False)+"\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)# ===== 使用示例 =====logger=StructuredLogger(process_name="销售数据采集")# 记录关键指标logger.info("开始采集",page=1,url="https://example.com/list")logger.info("采集完成",page=1,count=50,duration_seconds=3.2)logger.warning("数据异常",row=12,field="金额",value="",reason="为空")logger.error("API超时",url="https://api.example.com/data",timeout=10,retry=3)

结构化日志的价值在于，你可以用Python快速分析：

# 分析日志文件importjsondefanalyze_log(log_file):"""分析日志文件——统计错误、计算耗时"""errors=[]warnings=[]total_records=0withopen(log_file,"r",encoding="utf-8")asf:forlineinf:entry=json.loads(line)total_records+=1ifentry["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,env="dev",log_dir="D:/rpa_logs"):self.env=env self.file_logger=StructuredLogger(log_dir=log_dir)# 不同环境的日志级别self.level_map={"dev":0,# 记录所有"test":1,# 记录INFO及以上"prod":2,# 只记录WARNING和ERROR}self.min_level=self.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.env=="prod":self._send_alert(msg,extra)# 开发环境：详细日志logger=LogManager(env="dev")logger.debug("准备点击第3个按钮")# 生产环境：只看关键logger=LogManager(env="prod")logger.info("日常采集完成")# 记logger.debug("循环到了第15次")# 不记（级别太低）

第五步：异常自动记录

在try-catch中自动记录错误上下文：

importtracebackdeflog_error_with_context(logger,func_name,**context):"""在异常处理中记录完整上下文"""error_info=traceback.format_exc()logger.error(f"{func_name}执行失败",error=error_info,**context)# 在流程中使用defprocess_order(order_id):try:# 处理订单的逻辑passexceptExceptionase:log_error_with_context(logger,"process_order",order_id=order_id,current_page=current_page)# 决定是重试还是跳过raise

第六步：日志的轮转与清理

importglobdefcleanup_old_logs(log_dir,keep_days=7):"""清理过期日志"""cutoff=datetime.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_days=7)

有什么坑

坑一：生产环境打DEBUG日志撑爆磁盘

现象：用了半个月，C盘红了——一看日志文件夹50GB。

原因：每条循环都打了DEBUG日志，一天几万条。

解决：生产环境只用INFO及以上级别。给日志目录设大小限制。

坑二：日志文件名为中文导致编码问题

现象：日志文件名是"销售数据采集_20250315.log"，在某些系统上无法读取。

解决：文件名只用英文+数字+下划线：

process_name="sales_collector"# 不用"销售数据采集"

坑三：不区分业务日志和系统日志

现象：所有的信息都混在一个日志里，排查问题时在几千行日志中大海捞针。

解决：分离关注点——业务日志记录"采集了多少数据"，系统日志记录"操作是否成功"：

logger_biz=StructuredLogger(process_name="biz")# 业务日志logger_sys=StructuredLogger(process_name="sys")# 系统日志logger_sys.info("页面加载完成",url=url,duration=1.5)logger_biz.info("采集完成",count=50,total_amount=12500)

坑四：只在Python节点中打日志

现象：Python节点中的逻辑有日志，但影刀内置指令的操作没有日志——"这个【点击】到底点没点？"不知道。

解决：在影刀关键指令前后用【打印日志】指令：

【打印日志】→ "开始处理第N个商品" 【点击元素】→ 详情链接 【打印日志】→ "已打开详情页" 【等待元素】→ 商品标题 【打印日志】→ "详情页加载完成"

坑五：错误日志不记录输入数据

现象：日志里只有"处理失败"，不知道处理的是什么数据导致失败。

解决：错误日志必须带上当时的输入数据：

try:process_item(item_data)exceptExceptionase:logger.error("处理失败",item=item_data,# ← 关键：记录出错的数据![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/e3193deda15f40c399c51c9ed79c1223.png#pic_center)error=str(e),traceback=traceback.format_exc())

总结：日志系统不需要一开始就很复杂。分三步走：第一步，把print改成写文件（每条带时间戳）；第二步，改成JSON格式方便检索；第三步，加上分级和环境切换。即使在第二步停住，你的流程维护成本也已经大幅降低了。记住，日志存在的唯一价值是出问题的时候帮你快速定位——能让你10分钟内找到原因的日志，就是好日志。