影刀RPA 流程性能分析：定位瓶颈与优化

影刀RPA 流程性能分析定位瓶颈与优化署名林焱什么情况用你的RPA流程跑了2小时才出结果老板问能不能快一点。你说已经很快了。但到底慢在哪一步是网页加载慢、Excel写入慢、还是某个Python脚本卡了不测量就没法优化。性能分析就是给流程做体检——找到最慢的那几个步骤针对性优化。80%的性能问题出在20%的代码里关键是找到那20%。典型场景流程整体运行时间超过预期需要找瓶颈流程从开发环境搬到生产环境后变慢数据量增大后流程越来越慢需要向领导证明已经优化到位了怎么做1. 步骤级计时店群矩阵自动化突破运营极限在影刀流程的关键节点前后插入【执行Python代码】记录时间戳importtimeimportjsonimportos# 性能记录文件perf_log_pathrC:\RPAData\perf_log.jsondefrecord_step(step_name,start_time,detailsNone):记录步骤耗时elapsedtime.time()-start_time log_entry{step:step_name,elapsed_sec:round(elapsed,2),timestamp:time.strftime(%Y-%m-%d %H:%M:%S),details:detailsor{}}# 追加到日志文件logs[]ifos.path.exists(perf_log_path):withopen(perf_log_path,r)asf:logsjson.load(f)logs.append(log_entry)withopen(perf_log_path,w)asf:json.dump(logs,f,indent2,ensure_asciiFalse)print(f[{step_name}] 耗时{elapsed:.2f}秒)returntime.time()# 返回当前时间作为下一步的开始时间# 使用示例ttime.time()# 步骤1打开网页 # 影刀【打开网页】组件trecord_step(打开网页,t)# 步骤2采集数据 # 影刀【采集网页数据】组件trecord_step(采集数据,t,{rows:500})# 步骤3数据处理 # 影刀【执行Python代码】组件trecord_step(数据处理,t)# 步骤4写入Excel # 影刀【写入Excel文件】组件trecord_step(写入Excel,t,{rows:500})# 步骤5发送邮件 # 影刀【发送邮件】组件trecord_step(发送邮件,t)2. 生成性能报告importjsonimportosimportpandasaspd perf_log_pathrC:\RPAData\perf_log.jsonreport_pathrC:\RPAData\性能分析报告.xlsx# 读取日志withopen(perf_log_path,r)asf:logsjson.load(f)dfpd.DataFrame(logs)total_timedf[elapsed_sec].sum()# 计算占比df[占比](df[elapsed_sec]/total_time*100).round(1).astype(str)%# 排序dfdf.sort_values(elapsed_sec,ascendingFalse)# 输出报告print(f\n{*60})print(f流程性能分析报告)print(f{*60})print(f总耗时{total_time:.1f}秒 ({total_time/60:.1f}分钟))print(f步骤数{len(df)})print(f\n耗时排行从慢到快)print(f{步骤名称:15}{耗时(秒):10}{占比:8})print(f{-*40})for_,rowindf.iterrows():print(f{row[step]:15}{row[elapsed_sec]:10}{row[占比]:8})# 找出瓶颈耗时超过总时间20%的步骤bottlenecksdf[df[elapsed_sec]total_time*0.2]ifnotbottlenecks.empty:print(f\n⚠ 性能瓶颈耗时占比20%)for_,rowinbottlenecks.iterrows():print(f{row[step]}:{row[elapsed_sec]}秒 ({row[占比]}))# 导出Excel报告df[[step,elapsed_sec,占比,timestamp]].to_excel(report_path,indexFalse,sheet_name性能分析)print(f\n报告已导出{report_path})3. Python代码级性能分析importtimeimportfunctoolsdefprofile(func):函数性能分析装饰器functools.wraps(func)defwrapper(*args,**kwargs):starttime.time()resultfunc(*args,**kwargs)elapsedtime.time()-startifelapsed0.5:# 只记录耗时超过0.5秒的print(f [慢]{func.__name__}:{elapsed:.3f}秒)returnresultreturnwrapper# 使用profiledefclean_data(data):数据清洗time.sleep(1)# 模拟耗时操作return[dfordindataifd]profiledefdeduplicate(data):去重returnlist(set(data))profiledefsort_data(data):排序returnsorted(data)# 运行并自动输出慢函数datalist(range(10000))clean_data(data)deduplicate(data)sort_data(data)4. 内存使用监控importpsutilimportosdefcheck_memory(label):检查当前进程内存使用processpsutil.Process(os.getpid())memprocess.memory_info()print(f[内存{label}] RSS:{mem.rss/1024/1024:.1f}MB, VMS:{mem.vms/1024/1024:.1f}MB)returnmem.rss/1024/1024# 返回MB# 在关键步骤前后检查check_memory(开始)# ... 读取大文件 ...check_memory(读取文件后)# ... 数据处理 ...check_memory(数据处理后)# ... 写入Excel ...check_memory(写入Excel后)5. 常见优化手段importtimeimportpandasaspd# 优化1批量操作代替循环 datalist(range(10000))# 慢循环单条写入starttime.time()results[]foritemindata:results.append(item*2)print(f循环处理{time.time()-start:.3f}秒)# 快列表推导式starttime.time()results[item*2foritemindata]print(f列表推导式{time.time()-start:.3f}秒)# 更快pandas向量化starttime.time()dfpd.DataFrame({value:data})df[result]df[value]*2print(fpandas向量化{time.time()-start:.3f}秒)# 优化2减少IO操作 # 慢循环内写文件starttime.time()withopen(rC:\RPAData\slow.txt,w)asf:foriinrange(10000):f.write(f{i}\n)print(f循环写入{time.time()-start:.3f}秒)# 快一次性写入starttime.time()content\n.join(str(i)foriinrange(10000))withopen(rC:\RPAData\fast.txt,w)asf:f.write(content)print(f批量写入{time.time()-start:.3f}秒)# 优化3用集合代替列表做查找 big_listlist(range(100000))big_setset(big_list)starttime.time()foriinrange(1000):_99999inbig_list# 列表查找O(n)print(f列表查找1000次{time.time()-start:.3f}秒)starttime.time()foriinrange(1000):_99999inbig_set# 集合查找O(1)print(f集合查找1000次{time.time()-start:.3f}秒)有什么坑坑1只测一次不代表真实性能第一次运行可能有冷启动开销加载库、编译代码。应该测多次取平均importtimedefbenchmark(func,*args,runs5):多次运行取平均times[]for_inrange(runs):starttime.time()func(*args)times.append(time.time()-start)avgsum(times)/len(times)print(f平均耗时{avg:.3f}秒{runs}次运行)print(f最快{min(times):.3f}秒最慢{max(times):.3f}秒)returnavg坑2时间戳精度问题time.time()在Windows上精度可能是15毫秒测量很快的操作1ms会得到0。用time.perf_counter()替代# 测量短时间操作starttime.perf_counter()# ... 快速操作 ...elapsedtime.perf_counter()-start[video(video-f3zGVAvU-1784481982110)(type-csdn)(url-https://live.csdn.net/v/embed/524992)(image-https://v-blog.csdnimg.cn/asset/b59aed2f01d4fe8583467562aaf4dcfd/cover/Cover0.jpg)(title-temu店群自动化报活动案例)]# 精度到微秒坑3优化了不该优化的地方花了2小时优化一个从0.1秒到0.05秒的函数但整个流程跑10分钟。优化前后没区别。原则先找最大的瓶颈优化它再重新测量。不要凭感觉优化——用数据说话。# 性能优化优先级矩阵# 高耗时易优化 → 立即做# 高耗时难优化 → 值得投入# 低耗时易优化 → 顺手做# 低耗时难优化 → 不做坑4网页等待时间占大头但被忽略RPA流程最慢的往往是网页操作——等加载、等元素出现、等AJAX完成。Python代码优化到极致但网页等待从5秒减不到1秒。# 分析网页等待时间占比# 如果网页等待占80%优化Python代码没用# 应该优化等待策略# 1. 把固定等待(sleep 3)改成条件等待(等到元素出现)# 2. 并行操作多个标签页同时采集# 3. 用API接口代替网页操作坑5内存优化和速度优化的矛盾有时候省内存就会慢省时间就会占内存。比如pandas读取大文件# 快但占内存一次性读取dfpd.read_csv(big.csv)# 2GB文件占4GB内存# 省内存但慢分块读取chunkspd.read_csv(big.csv,chunksize10000)# 每次只占几十MBforchunkinchunks:process(chunk)# 根据实际情况选择内存够就追求速度内存不够就分块