影刀RPA 从零搭建一个完整自动化流程构思、开发、调试、上线的全流程实战作者林焱很多新手学影刀的时候是这样学了一堆指令看了不少教程真到自己上手写流程的时候却盯着编辑器不知道从哪下手。即使磕磕绊绊写出来了一跑就红一片改了半天还是不对。折腾一天发现不如自己手动操作来得快。问题出在哪不是你不会用指令是你不懂得把一个需求拆成一个完整的自动化流程。从想法到能稳定运行的成品中间有构思、开发、调试、上线四个阶段每个阶段都有固定的套路。这篇文章用一个真实案例——每日自动采集竞品价格并生成对比报告——带你完整走一遍这四个阶段。读完之后你拿到任何一个自动化需求都知道怎么下手。一、构思阶段把需求翻译成自动化步骤1.1 需求是什么先把这个案例的需求定清楚每天早上9点自动打开三个竞品网站的商品列表页采集每个网站上某类产品的名称和价格和我们自己的价格做对比把对比结果写入Excel标红价格高于我们的产品把Excel文件通过邮件发给自己如果你拿到的需求只有一句话帮我自动对比竞品价格你会怎么做直接开写不第一步永远是拆步骤。1.2 需求拆解把一个模糊需求拆成可执行的操作步骤这个能力比会写代码重要十倍。拆的时候问自己三个问题这个流程从哪开始起点需要经过哪些中间步骤路径最终产出什么终点我们这个案例拆出来是这样起点每天早上9点触发 ↓ 步骤1打开竞品网站A → 搜关键词 → 采集商品名价格 步骤2打开竞品网站B → 搜关键词 → 采集商品名价格 步骤3打开竞品网站C → 搜关键词 → 采集商品名价格 ↓ 步骤4读取我们自己的价格表Excel ↓ 步骤5把三家的数据和自家的对比算出价差 ↓ 步骤6生成对比报表Excel超标的产品标红 ↓ 步骤7把报表发邮件给自己这个拆解看着简单但很多人就是省了这一步导致写着写着发现逻辑丢了流程断了然后又重头补。1.3 确定技术方案步骤拆清楚之后再决定每一步用什么技术实现步骤技术方案为什么选它定时触发影刀计划中心影刀自带不需要额外的Windows计划任务配置打开网页采集数据【打开网页】【获取元素列表】/【获取元素文本】三个网站结构各不相同列表采集最灵活搜索商品【填写输入框】【点击按钮】【等待元素】搜索后需要等结果加载等待策略要配好读取自家价格表【打开Excel】/【读取Excel区域】自家数据源头固定每次读取最新即可数据对比Python节点pandas多数据源合并对比pandas最方便生成报表标红Python节点openpyxl写入条件格式标红原生指令做不到发送邮件【发送邮件】/Python smtplib带附件用Python更灵活结果通知【日志输出】记录执行情况出问题好查这里面有个关键决策什么时候用影刀原生指令什么时候用Python节点原则很简单能用指令完成的优先用指令指令做不了或者做起来很笨重的用Python。比如元素采集用指令因为它本来就是影刀的强项数据对比和Excel标红用Python因为pandas和openpyxl就是干这个的。1.4 构思阶段的常见坑坑1一上来就写代码拿到需求直接打开影刀编辑器开始拖指令拖到一半发现流程结构不对又删掉重来。构思阶段花30分钟画个流程图能省开发阶段3个小时的返工。坑2技术选型太理想化有人觉得全用Python节点写才高级结果写出来的流程调试困难日志不清晰团队接手根本看不懂。选技术的原则是简单可靠优先不是炫技优先。坑3没考虑异常情况构思的时候只想了正常流程没想如果网站打不开怎么办“如果搜索结果为空怎么办”“如果Excel文件被占用怎么办”。这些问题不在构思阶段想清楚后面遇到一个卡一个。我在构思阶段通常会在每步后面标注可能出问题的地方步骤1打开竞品网站A → 异常1网站打不开 → 等待30秒重试超过3次跳过 → 异常2搜索无结果 → 记录无数据继续下一个 → 异常3页面加载超时 → 等待60秒超时记录日志拼多多店群自动化报活动上架二、开发阶段从小模块到大流程2.1 先写最小可用模块很多人喜欢一口气把全流程写完再跑这是最大的错误。正确做法是写完一个模块就跑一个模块确认没问题再接下一个。以我们这个案例为例先写竞品网站A的采集子流程。子流程采集竞品A数据主流程调用此子流程传入关键词返回采集结果列表 操作步骤 1. 【打开网页】url https://www.site-a.com/search 2. 【填写输入框】定位搜索框 → 输入关键词 3. 【点击】点击搜索按钮 4. 【等待元素】等待搜索结果列表出现超时30秒 5. 【获取元素列表】获取商品名称列表 6. 【获取元素列表】获取价格列表 7. 【关闭网页】 8. 输出结果商品名价格的列表子流程写完先用一个测试主流程调用它只传一个关键词跑一遍主流程测试用 1. 【运行子流程】运行采集竞品A传入关键词蓝牙耳机 2. 【日志输出】输出采集结果跑通了数据正常再写采集B和采集C的子流程。2.2 数据处理模块用Python三个网站的数据都采集完之后用Python节点做数据处理。这个节点是整个流程的核心逻辑。importpandasaspdfromdatetimeimportdatetime# 把影刀采集的数据转成DataFrame # 假设影刀传入的数据格式三组列表每组是[{名称, 价格}, ...]# 在影刀Python节点中用输入变量接收site_a_datapd.DataFrame(get_variable(site_a_result))site_b_datapd.DataFrame(get_variable(site_b_result))site_c_datapd.DataFrame(get_variable(site_c_result))my_datapd.DataFrame(get_variable(my_price_list))# 添加来源标记site_a_data[来源]竞品Asite_b_data[来源]竞品Bsite_c_data[来源]竞品Cmy_data[来源]我方# 合并所有数据 all_datapd.concat([site_a_data,site_b_data,site_c_data,my_data],ignore_indexTrue)# 按商品名做对比 # 先找出所有商品的中位数价格作为参考pivotall_data.pivot_table(index商品名,columns来源,values价格,aggfuncfirst).reset_index()# 计算我方价格与各竞品的价差forcolin[竞品A,竞品B,竞品C]:ifcolinpivot.columns:pivot[f{col}_价差]pivot[我方]-pivot[col]# 标记需要关注的产品我方价格高于任一竞品pivot[需要关注]Falseforcolin[竞品A_价差,竞品B_价差,竞品C_价差]:ifcolinpivot.columns:pivot[需要关注]pivot[需要关注]|(pivot[col]0)set_variable(comparison_result,pivot.to_dict(records))2.3 报表生成对比结果出来后写入Excel并标红。importopenpyxlfromopenpyxl.stylesimportPatternFill,Font,Alignment,Border,Sidefromopenpyxl.utilsimportget_column_letter result_dataget_variable(comparison_result)wbopenpyxl.Workbook()wswb.active ws.title竞品价格对比# 写表头 headers[商品名,我方价格,竞品A,竞品A价差,竞品B,竞品B价差,竞品C,竞品C价差,关注]forcol_idx,headerinenumerate(headers,1):cellws.cell(row1,columncol_idx,valueheader)cell.fontFont(boldTrue,size11)cell.fillPatternFill(start_color4472C4,end_color4472C4,fill_typesolid)cell.fontFont(boldTrue,colorFFFFFF,size11)cell.alignmentAlignment(horizontalcenter)# 写数据并标红 red_fillPatternFill(start_colorFF6B6B,end_colorFF6B6B,fill_typesolid)forrow_idx,iteminenumerate(result_data,2):ws.cell(rowrow_idx,column1,valueitem.get(商品名,))ws.cell(rowrow_idx,column2,valueitem.get(我方,))price_columns{3:竞品A,4:竞品A_价差,5:竞品B,6:竞品B_价差,7:竞品C,8:竞品C_价差,}forcol,keyinprice_columns.items():valitem.get(key)ifvalisnotNone:ws.cell(rowrow_idx,columncol,valueval)# 需要关注的行整行标红ws.cell(rowrow_idx,column9,value⚠ 关注ifitem.get(需要关注)else正常)ifitem.get(需要关注):forcolinrange(1,10):ws.cell(rowrow_idx,columncol).fillred_fill# 自适应列宽 forcol_idxinrange(1,10):max_len0forrow_idxinrange(1,len(result_data)2):cell_valws.cell(rowrow_idx,columncol_idx).valueifcell_val:max_lenmax(max_len,len(str(cell_val)))ws.column_dimensions[get_column_letter(col_idx)].widthmax_len4# 保存todaydatetime.now().strftime(%Y%m%d)filepathfD:\\竞品分析报表\\竞品价格对比_{today}.xlsxwb.save(filepath)set_variable(report_filepath,filepath)2.4 组装主流程所有子流程都单独测试通过后组装主流程主流程每日竞品价格对比 1. 【日志输出】 开始执行每日竞品价格对比 2. 【运行子流程】采集竞品A → site_a_result 3. 【日志输出】竞品A采集完成共XX条 4. 【运行子流程】采集竞品B → site_b_result 5. 【日志输出】竞品B采集完成共XX条 6. 【运行子流程】采集竞品C → site_c_result 7. 【日志输出】竞品C采集完成共XX条 8. 【打开Excel】D:\价格表\我方价格表.xlsx 9. 【读取Excel区域】→ my_price_list 10. 【关闭Excel】 11. Python节点数据对比与报表生成 12. 【日志输出】对比报表已生成{report_filepath} 13. Python节点/【发送邮件】发送报表附件 14. 【日志输出】 执行完毕 每个子流程都在外层套一个Try…CatchTry 【运行子流程】采集竞品A Catch 【日志输出】竞品A采集失败{错误信息}跳过继续 【赋值为空列表】site_a_result [] End Try这样做的好处是就算一个网站挂了其他两个网站的采集不受影响流程不会整体崩。2.5 开发阶段的避坑清单坑1子流程变量不传回去子流程里用【设置流程变量】设置了变量以为主流程能拿到结果拿不到。子流程返回数据要么用【输出变量】要么用Python节点的set_variable。流程变量只在当前流程作用域内有效子流程出来就没了。坑2Excel文件打开后忘了关采集数据时【打开Excel】→ 读数据 → Python节点处理 → 然后Python节点里又用openpyxl打开同一个文件写报表。结果报错文件被占用。影刀打开的文件一直占着锁必须【关闭Excel】之后才能用其他方式操作这个文件。坑3URL硬编码直接在主流程里写死https://www.site-a.com/search?q蓝牙耳机。过两天要搜别的产品得改代码。正确做法是用流程参数或配置文件传关键词。三、调试阶段从一团红到稳定运行3.1 分层调试流程全部写完一跑大概率不会一次成功。这时候不要慌按层级来排查第一层子流程级单独跑每个子流程确认每个子流程本身没问题。如果子流程都跑不通主流程一定跑不通。调试方法右键子流程 → “以调试模式运行” → 逐步骤看变量的值。第二层数据传递级子流程都通过了但主流程报错。重点看子流程的输出和主流程拿到的数据是否对得上。调试方法在每个子流程调用后加【日志输出】打印返回数据的类型和长度。第三层Python节点级数据传递没错但Python节点报错。把Python代码拷贝出来在本地Jupyter或IDLE里用样例数据跑逐行看哪一行报错。调试方法在Python节点中加print语句跑完看影刀的输出面板。3.2 最常见的调试场景场景1元素找不到报错未能找到目标元素 排查 → 打开网页后先手动F12看看元素在不在 → 确认元素在iframe里这是最常见的原因 → 确认没有Shadow DOM → 确认等待时间够不够有时候要等3-5秒元素才渲染出来场景2获取的元素列表是空的报错IndexError 或 取不到元素文本 排查 → 【日志输出】打印列表长度确认到底有没有元素 → 检查选择器是不是写对了XPath/CSS选择器 → 页面是否加载完成加一个【等待元素】 → 是不是页面有不同的状态登录态/未登录态显示不同场景3Python节点变量类型对不上报错NoneType object has no attribute ... 排查 → 在Python节点开头加一行 print(type(get_variable(xxx))) → 确认传入的是列表/字典/字符串不是你预期的类型 → 影刀的【获取元素列表】返回的是列表【获取元素文本】返回的是字符串3.3 调试工具箱每次调试都加这三样东西会省很多时间工具1关键节点后加【日志输出】【运行子流程】采集竞品A → result 【日志输出】竞品A返回数据类型{type(result)}, 长度{len(result)}一行日志抵你盯屏幕十分钟。工具2子流程开头加测试开关在子流程最前面加一个流程参数debug_mode如果 debug_mode True 【日志输出】【调试模式】子流程采集竞品A开始执行 【日志输出】参数keyword{keyword} 调试时传True上线后传False不污染生产日志。工具3Python节点里用try包裹关键逻辑try:resultprocess_data(raw_data)set_variable(status,success)exceptExceptionase:set_variable(status,failed)set_variable(error_msg,str(e))# 返回空数据别让流程整个崩set_variable(comparison_result,[])3.4 调试阶段的心态说点实在的第一次写完流程大概率跑不通这太正常了不等于你不行。我写了几百个流程第一个版本也很少一次通过的。关键是不要反复改全流程再跑那是浪费时间。我的做法是跑一次看第一个报错在哪只修这一个错误单独跑出错的那个模块修到通过后再跑全流程循环直到没有报错四、上线阶段从能跑到稳定运行4.1 上线前的收敛工作TEMU店群矩阵自动化运营核价报活动流程调通了不等于能上线。上线前至少做这些事情收敛1清理调试代码把所有只在调试时用的日志输出、中间变量、注释掉的代码清理掉。只保留对后续排查有帮助的日志。收敛2统一变量命名检查主流程和子流程的变量名是否一致、有意义。不要出现变量1、temp、aaa这种名字。收敛3参数外提把可能变化的参数提到流程参数或配置文件里关键词 → 流程参数网站URL → 配置文件文件路径 → 配置文件用相对路径或可配置的绝对路径邮件收件人 → 配置收敛4增加容错每个可能失败的节点前加容错处理1. 文件是否存在检查 2. 网络连接检查访问失败重试3次 3. 数据为空的兜底处理 4. 超时保护每个网络操作设置超时时间4.2 发布到影刀计划中心上线后不需要手动触发配到影刀计划中心让它在后台跑在影刀客户端点击计划中心新建计划 → 选择要执行的流程设置执行频率每天早上9:00设置执行设备选择你的电脑注意电脑不能关机失败通知勾选执行失败时通知我关于定时任务的细节可以参考我之前写的《影刀RPA 定时任务配置实战Windows计划任务与影刀计划中心》。这里只强调一点电脑必须保持开机且影刀客户端在线。计划中心依赖影刀客户端运行如果电脑休眠了或者影刀退出了计划不会执行。4.3 上线后观察期刚上线的流程前3天要特别关注第一天手动跑一遍确认在真实网络环境下数据正常第二天看计划中心是否在预定时间准时触发第三天检查邮件收到的报表确认内容完整、标红正确如果前三天都稳定基本可以放心了。但建议每周随机抽查一次因为网站结构随时可能变。4.4 长期维护要点网站改版监控每个月手动检查一次三个竞品网站的结构有没有变化。如果采集到的数据突然归零大概率是网站改版了重新定位一下元素。异常处理积累每次出问题在流程里补上这个异常的捕获。比如某次竞品A返回了空列表但流程没报错结果报表里竞品A的数据全是空的。那就加一个检测如果列表为空日志标记可能网站改版。版本管理每次改流程先复制一份出来改不要直接在跑着计划的那个版本上改。改完测试通过后再替换。五、全流程框架模板把这个案例的项目结构抽成一个模板以后做任何自动化项目都可以套用项目名称XX自动化流程 │ ├── 主流程.flow │ 负责编排子流程、处理整体异常 │ ├── 子流程/ │ ├── 数据采集_来源A.flow │ ├── 数据采集_来源B.flow │ ├── 数据处理_核心逻辑.flowPython节点集中在这里 │ └── 结果输出_发送通知.flow │ ├── 配置文件/ │ └── config.jsonURL、关键词、路径、邮件配置 │ ├── 测试流程/ │ ├── 测试_采集子流程.flow │ └── 测试_Python节点.flow │ └── README.md流程说明、运行条件、已知问题六、从这出发你可以做什么这篇文章的案例是一个竞品价格对比但框架是完全通用的。你可以把这个框架套到任何场景竞品价格对比 →每日舆情监控采集新闻 → 提取关键词 → 生成日报竞品价格对比 →多平台库存同步采集各平台库存 → 对比 → 标差异竞品价格对比 →招聘信息汇总采集各招聘网站 → 去重 → 生成汇总表竞品价格对比 →多平台评论区采集采集评论 → 情感分析 → 报告核心思路就是构思 → 开发 → 调试 → 上线四步循环。拿到需求先拆步骤写一个模块测一个模块分层调试逐层排查上线前收敛加固。别一上来就想着写一个几百步的完美流程那不现实。小模块拼大流程稳稳当当。