影刀RPA 批量处理的进度追踪：让长任务可视化-北京尧图网络

影刀RPA 批量处理的进度追踪让长任务可视化作者林焱什么情况用这个你在跑一个处理5000条数据的流程半个小时过去了窗口没有任何反应——不知道处理到第几条了、还要多久、有没有卡死。你犹豫要不要手动关掉重来但又怕已经处理了4000条关了白跑。这就是没有进度追踪的后果。流程在拼命干活你在干着急。进度追踪不需要复杂的UI几行代码就能让你的长流程变得透明——随时知道进度、预估剩余时间、判断是否正常。怎么做第一步最简单的进度输出totallen(items)foridx,iteminenumerate(items):process_item(item)# 每10条输出一次进度[video(video-F4scrC9o-1784131575534)(type-csdn)(url-https://live.csdn.net/v/embed/525010)(image-https://v-blog.csdnimg.cn/asset/f4faa587144cb7070f19e8b36813806b/cover/Cover0.jpg)(title-店群矩阵自动化突破运营极限)]if(idx1)%100:pct(idx1)/total*100print(f进度:{idx1}/{total}({pct:.1f}%))这样至少你能看到输出在滚动知道流程没死。第二步带预估剩余时间的进度importtimefromdatetimeimportdatetime,timedeltaclassProgressTracker:进度追踪器def__init__(self,total,label处理中):self.totaltotal self.labellabel self.start_timedatetime.now()self.last_updateself.start_time self.processed0self.checkpoints[]# 记录时间点用于计算速度defupdate(self,count1):更新进度每处理完count条调用一次self.processedcount nowdatetime.now()# 每隔一段时间或每隔一定数量输出elapsed(now-self.last_update).total_seconds()ifelapsed2andself.processedself.total:return# 不要太频繁输出self.last_updatenow self.checkpoints.append((self.processed,now))self._print_progress()def_print_progress(self):输出进度信息pctself.processed/self.total*100elapsed(datetime.now()-self.start_time).total_seconds()# 计算速度和预估剩余时间speedself.processed/elapsedifelapsed0else0# 条/秒remainingself.total-self.processed eta_secondsremaining/speedifspeed0else0# 格式化输出bar_len30filledint(bar_len*self.processed/self.total)bar█*filled░*(bar_len-filled)print(f\r{self.label}f[{bar}] f{self.processed}/{self.total}({pct:.1f}%) | f速度:{speed:.1f}条/秒 | f预计剩余:{timedelta(secondsint(eta_seconds))},endifself.processedself.totalelse\n)deffinish(self):完成——输出汇总total_time(datetime.now()-self.start_time).total_seconds()avg_speedself.total/total_timeiftotal_time0else0print(f\n✓{self.label}完成总计{self.total}条耗时{timedelta(secondsint(total_time))}平均{avg_speed:.1f}条/秒)# 使用示例 itemsrange(1000)trackerProgressTracker(len(items),采集数据)foridx,iteminenumerate(items):process_item(item)tracker.update(1)tracker.finish()第三步记录进度到文件——断点续跑的基础importjsonclassFileProgressTracker(ProgressTracker):带文件记录的进度追踪器支持断点续跑def__init__(self,total,label,progress_fileprogress.json):super().__init__(total,label)self.progress_fileprogress_filedefupdate(self,count1):super().update(count)# 每次更新同时写文件self._save()def_save(self):保存进度到文件state{total:self.total,processed:self.processed,pct:round(self.processed/self.total*100,1),start_time:self.start_time.isoformat(),last_update:self.last_update.isoformat(),status:running}withopen(self.progress_file,w,encodingutf-8)asf:json.dump(state,f,ensure_asciiFalse)deffinish(self):super().finish()# 标记完成withopen(self.progress_file,w,encodingutf-8)asf:json.dump({status:completed,processed:self.total},f)staticmethoddefload_checkpoint(progress_file):加载上次的进度ifnotos.path.exists(progress_file):return0withopen(progress_file,r,encodingutf-8)asf:statejson.load(f)ifstate.get(status)completed:returnstate.get(total,0)returnstate.get(processed,0)第四步并行任务的进度聚合classMultiTaskTracker:多任务进度聚合器def__init__(self):self.tasks{}defadd_task(self,task_id,total):self.tasks[task_id]ProgressTracker(total,f任务{task_id})defupdate(self,task_id,count1):iftask_idinself.tasks:self.tasks[task_id].update(count)defget_overall_progress(self):获取总体进度totalsum(t.totalfortinself.tasks.values())processedsum(t.processedfortinself.tasks.values())pctprocessed/total*100iftotal0else0return{total:total,processed:processed,pct:pct}第五步异常检测——进度卡住自动告警classSmartProgressTracker(ProgressTracker):带异常检测的进度追踪器def__init__(self,total,label,stall_threshold60):super().__init__(total,label)self.stall_thresholdstall_threshold# 停滞超过N秒触发告警self.last_progress_timeself.start_time self.last_processed_count0defupdate(self,count1):self.processedcountifself.processedself.last_processed_count:# 有新进度self.last_progress_timedatetime.now()self.last_processed_countself.processed# 检查是否停滞stall_seconds(datetime.now()-self.last_progress_time).total_seconds()ifstall_secondsself.stall_threshold:print(f\n⚠️ 进度停滞{stall_seconds:.0f}秒当前{self.processed}/{self.total})# 这里可以触发告警self._print_progress()有什么坑坑一每处理一条都更新进度太频繁现象处理10000条每一条都输出进度输出本身占了大量时间。解决合理设置更新频率——总数100时每条更新100-1000时每10条1000时按百分比更新每1%或按时间间隔每2秒。坑二进度倒退的错觉现象有的循环中total不确定比如加载更多的场景进度先显示50%数据翻倍后变成25%看起来像倒退了。解决总数不确定时显示已处理N条而非百分比不显示进度条。temu店群自动化报活动案例坑三预估时间与实际差距太大现象显示剩余5分钟结果跑了20分钟。原因前几条处理很快缓存后面越来越慢网络IO。解决用加权平均或只计算最近N条的平均速度# 只取最近100条的速度recent_speedmin(100,self.processed)/recent_elapsed坑四在影刀控制台输出乱码现象进度条的█和░在影刀控制台显示为乱码。解决用ASCII字符代替bar#*filled-*(bar_len-filled)总结进度追踪从一行print(f{idx1}/{total})开始就够了。当你需要预估剩余时间和断点续跑时再升级到 ProgressTracker 类。核心原则是让等待的人知道发生了什么——进度数字比进度条重要预估时间比百分比重要。

影刀RPA 批量处理的进度追踪：让长任务可视化

相关推荐

告别网盘限速烦恼：LinkSwift直链下载助手完整使用指南

基于GNN的服务依赖图异常检测：在生产环境中识别微服务调用链中的隐性故障传播

影刀RPA 接单实战：从需求对接到交付验收全流程

更多资讯

帝国CMS 7.5后台响应式模板V2.4：安装部署与二次开发指南

动态规划解决含手续费股票交易问题

Terra：Rust+WASM重构的3D GIS引擎与真实大气渲染

C语言结构体：自定义数据类型详解与实践

ERP实施效率陷阱解析与破局之道

驰宇微液晶屏选型与定制应用实战指南

今日更新

SpacetimeDB C++ 模块库实战：用 C++20 在数据库内构建 WebAssembly 模块

别踩雷！不是所有 AI 写作工具都靠谱，2026 高校认可工具精选

SSM+MySQL幼儿园管理系统源码解析：从部署到优化

本周精选

超人会飞不算本事：系统稳定依赖清晰规则与边界设计

超人VS蜘蛛侠：拆解超级IP的影响力与传播方法论

基于CNN的调制信号识别：MATLAB实现时频图分类实战

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？