影刀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 接单实战：从需求对接到交付验收全流程

更多资讯

揭秘电商智能系统：从选品到私域的全链路优化

MoE混合专家模型：稀疏激活、路由机制与推理部署优化全解析

AI Agent专用数据库MemoryDB架构与实战

云端计算在气象干旱监测中的应用与SPI分析实践

Zcode OSS上传风波复盘：从抓包误读到harness诊断机制的技术真相

curl命令分层实战：接口调试、耗时打点与自动化脚本

今日更新

AI辅助技术栈评估：提升研发决策效率60%

从生成Token到直接决策：Jev如何重塑AI Agent工程实践

32G统一内存跑大模型：内存账、TPS实测与端云分流

本周精选

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

本月热门

面向复杂光照与角度的宠物面部识别鲁棒性增强策略

基于DRV8818与STM32的双极步进电机驱动方案实战解析

DRV8818PWPR+PIC18F45K80双极步进电机控制方案实战

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