影刀RPA进阶教程：日志系统搭建——让每个流程都有完整的执行轨迹
影刀RPA日志系统是被大多数新手忽视的基础设施。流程跑着跑着挂了，不知道在哪一步挂的、处理了多少条数据、为什么挂——因为没有日志。本文帮你从零搭建一套标准化日志系统，让流程的执行过程透明可控。
作者：林焱 | 影刀RPA进阶系列
没有日志的痛苦
假设你周一早上发现周末两天的采集数据都没入库。没有日志的情况下：
不知道流程什么时候停的（周六上午还是周日晚上？）
不知道停了的时候处理了多少条数据（要从头重跑还是接着跑？）
不知道停的原因（网络超时？登录过期？服务器重启？）
只能全部重跑，浪费时间和资源
有日志的情况下：看眼日志文件，5秒定位问题。
店群矩阵自动化突破运营极限！
日志的三个等级
不要什么信息都记，日志要分层：
| 等级 | 内容 | 示例 |
|---|---|---|
| INFO | 正常流程节点 | “开始采集第3页”、“登录成功” |
| WARNING | 不影响运行但需关注 | “第5页数据为空，跳过”、“重试第2次” |
| ERROR | 流程失败/异常 | “登录失败：账号密码错误”、“元素#submit未找到” |
日志系统设计
importloggingimportosfromdatetimeimportdatetimeclassFlowLogger:""" 影刀流程专用日志系统 """def__init__(self,flow_name,log_dir="C:/logs/"):self.flow_name=flow_name# 按日期创建日志目录today=datetime.now().strftime('%Y-%m-%d')self.log_dir=os.path.join(log_dir,today)os.makedirs(self.log_dir,exist_ok=True)# 日志文件路径self.log_file=os.path.join(self.log_dir,f"{flow_name}_{datetime.now().strftime('%H%M')}.log")# 配置loggerself.logger=logging.getLogger(f"{flow_name}_{datetime.now().timestamp()}")self.logger.setLevel(logging.INFO)# 文件处理器file_handler=logging.FileHandler(self.log_file,encoding='utf-8')file_handler.setLevel(logging.DEBUG)# 控制台处理器（可选）console_handler=logging.StreamHandler()console_handler.setLevel(logging.INFO)# 日志格式formatter=logging.Formatter('%(asctime)s | %(levelname)-7s | %(message)s',datefmt='%Y-%m-%d %H:%M:%S')file_handler.setFormatter(formatter)console_handler.setFormatter(formatter)self.logger.addHandler(file_handler)self.logger.addHandler(console_handler)# 记录启动self.logger.info(f"=========={flow_name}开始执行 ==========")definfo(self,msg):self.logger.info(msg)defwarning(self,msg):self.logger.warning(msg)deferror(self,msg):self.logger.error(msg)defstep(self,step_name,step_num=None):"""记录流程步骤"""prefix=f"[步骤{step_num}] "ifstep_numelse""self.logger.info(f"{prefix}{step_name}")defsummary(self,stats_dict):"""记录统计摘要"""self.logger.info("---------- 执行摘要 ----------")forkey,valueinstats_dict.items():self.logger.info(f"{key}:{value}")self.logger.info("-------------------------------")defclose(self,success=True):"""记录结束"""status="成功"ifsuccesselse"失败"self.logger.info(f"=========={self.flow_name}执行{status}==========\n")在流程中集成日志
以采集流程为例，每个关键节点都记日志：
# 初始化日志log=FlowLogger("淘宝竞品采集")try:# 登录log.step("开始登录",1)login_result=login_taobao(account,password)iflogin_result['success']:log.info(f"登录成功，账号：{account[:3]}***")else:log.error(f"登录失败：{login_result['error']}")log.close(success=False)exit()# 搜索关键词列表keywords=['蓝牙耳机','无线鼠标','机械键盘']log.info(f"关键词列表：{len(keywords)}个")total_collected=0forkeywordinkeywords:log.step(f"开始采集关键词：{keyword}")# 搜索search_result=search_keyword(keyword)log.info(f"搜索结果：{search_result['total']}条")# 翻页采集page=1keyword_count=0whilepage<=10:log.info(f" 采集第{page}页")page_data=collect_page(keyword,page)ifnotpage_data:log.warning(f" 第{page}页无数据，停止翻页")breakkeyword_count+=len(page_data)save_data(page_data)page+=1total_collected+=keyword_count log.info(f"关键词'{keyword}'采集完成：{keyword_count}条")# 摘要log.summary({'采集关键词数':len(keywords),'总数据量':total_collected,'执行耗时':f"{elapsed_seconds}秒",'保存文件':output_file})log.close(success=True)exceptExceptionase:log.error(f"流程异常终止：{str(e)}")# 记录完整的错误堆栈importtraceback log.error(traceback.format_exc())log.close(success=False)日志分析：快速定位问题
日志积累多了以后，用脚本快速分析：
importpandasaspdimportreimportosdefanalyze_logs(log_dir,days=7):"""分析最近的日志，找出异常"""all_errors=[]# 遍历日志文件forfilenameinos.listdir(log_dir):ifnotfilename.endswith('.log'):continuefilepath=os.path.join(log_dir,filename)withopen(filepath,'r',encoding='utf-8')asf:content=f.read()# 查找ERROR行errors=re.findall(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) \| ERROR.*?\| (.*)',content)fortimestamp,msginerrors:flow_name=filename.split('_')[0]all_errors.append({'时间':timestamp,'流程':flow_name,'错误':msg,'文件':filename})ifall_errors:df=pd.DataFrame(all_errors)df.to_excel("日志异常汇总.xlsx",index=False)# 统计各流程运行次数flow_stats={}forfilenameinos.listdir(log_dir):iffilename.endswith('.log'):withopen(os.path.join(log_dir,filename),'r',encoding='utf-8')asf:content=f.read()flow_name=filename.split('_')[0]success_count=content.count('执行成功')fail_count=content.count('执行失败')flow_stats[flow_name]={'成功':success_count,'失败':fail_count,'成功率':f"{success_count/(success_count+fail_count)*100:.1f}%"if(success_count+fail_count)>0else'N/A'}returnpd.DataFrame(flow_stats).T日志存储策略
日志会随着时间越积越多，要有清理策略：
temu店群自动化报活动案例
# 日志保留规则 近7天：保留所有日志 8-30天：保留ERROR和WARNING级日志，删除INFO级 30天以上：压缩存档到 archive/ 目录  90天以上：删除可以用影刀流程本身来做日志清理——再写一个小流程，每周日自动清理过期日志。
小结
日志系统的核心价值不在于"记了什么"，而在于"出问题时能多快定位"。几个原则：
关键节点必须记（登录、翻页、保存、异常）
日志要带上下文（第几页、第几条、什么关键词）
异常要记堆栈（出错后至少知道哪一行代码崩了）
定期分析日志（从日志里发现流程的薄弱环节）
花半小时搭建好日志模块，以后排查问题节省的时间是百倍千倍。
内容标签：影刀RPA / 日志系统 / 流程监控 / 错误排查 / Python logging
作者：林焱
本文是「影刀RPA从零到实战」系列文章之一，持续更新实操技巧与避坑经验。
