影刀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 批量处理的进度追踪让长任务可视化作者林焱什么情况用这个你在跑一个处理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 类。核心原则是让等待的人知道发生了什么——进度数字比进度条重要预估时间比百分比重要。
