影刀RPA深度实战：电商数据采集系统从0到1完整搭建手册-开发者社区

影刀RPA深度实战：电商数据采集系统从0到1完整搭建手册

作者：林焱

做电商数据采集的时候，最头疼的不是元素抓不到，而是流程跑两天就崩——元素失效、页面加载超时、数据重复、异常中断后无法续跑。我之前做一个小红书采集项目的时候，前3版流程都是跑着跑着就挂了，后来总结出来一套完整的系统搭建方法，照着做能让你的采集系统稳定跑7×24小时。

一、系统架构设计（先做对，再做好）

一个能稳定运行的采集系统，不是把所有指令堆在一个流程里，而是按「采集→清洗→存储→异常处理」四层来拆。

1.1 子流程拆分原则

子流程	职责	复用场景
`子流程_采集入口`	接收关键词/URL列表，调度采集任务	所有采集任务通用
`子流程_元素定位`	封装元素捕获+失败重试逻辑	同平台多个页面复用
`子流程_数据清洗`	去除表情、换行、特殊字符，统一格式	所有数据源通用
`子流程_数据写入`	写入Excel/飞书，含去重校验	所有输出目标通用
`子流程_异常处理`	截图+记录错误日志+决定是否续跑	全局通用

经验：子流程参数用「对象变量」传递，不要散传多个参数，后期改起来要命。

二、元素定位：不失效的核心方法

2.1 不要用「自动捕获」生成的XPath

影刀自动捕获生成的XPath长这样：

/html/body/div[1]/div[2]/div[3]/div/div[1]/div[2]/a[1]/span

这种XPath页面结构一变就失效。

正确做法：用「属性定位法」，先找元素的稳定属性。

2.2 小红书笔记标题的XPath实战

打开小红书搜索页，用Chrome开发者工具（F12）找到标题元素，观察它的属性：

//div[@class='note-item']//a[@class='cover mask']/@href

但这还不够稳，因为小红书的class会带哈希后缀。升级版：

//a[contains(@href, '/explore/') and @target='_blank']

完整可运行的Python代码（用影刀Python节点执行）：

# -*- coding: utf-8 -*-# 功能：用 requests + lxml 解析小红书搜索页，绕过影刀元素捕获的限制# 适用场景：小红书搜索结果页批量获取笔记链接importrequestsfromlxmlimportetreeimporttimeimportrandomdeffetch_xiaohongshu_notes(keyword,max_page=3):""" 采集小红书搜索结果中的笔记链接 :param keyword: 搜索关键词，如"护肤品推荐" :param max_page: 采集页数，默认3页 :return: 笔记链接列表 """notes=[]# Cookie需要替换成你自己的，登录小红书后从开发者工具Network里复制headers={'User-Agent':'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36','Cookie':'your_cookie_here',# ← 替换成你的Cookie'Referer':'https://www.xiaohongshu.com/',}forpageinrange(1,max_page+1):# 随机延迟，避免被限流time.sleep(random.uniform(2,4))url=f'https://www.xiaohongshu.com/search_result?keyword={keyword}&source=web_search_result_notes&page={page}'try:resp=requests.get(url,headers=headers,timeout=10)# 检查状态码，如果不是200说明被限流了ifresp.status_code!=200:print(f'第{page}页请求失败，状态码：{resp.status_code}')continuehtml=etree.HTML(resp.text)# XPath：找所有包含 /explore/ 的链接（小红书笔记链接特征）links=html.xpath('//a[contains(@href, "/explore/")]/@href')# 去重+补全链接forlinkinlinks:full_link='https://www.xiaohongshu.com'+linkiflink.startswith('/')elselinkiffull_linknotinnotes:notes.append(full_link)print(f'第{page}页采集完成，获得{len(links)}条链接，累计{len(notes)}条')exceptExceptionase:print(f'第{page}页采集异常：{str(e)}')# 遇到异常等待更长时间再试time.sleep(10)returnnotes# 调用示例（在影刀Python节点里调用这个函数）# result = fetch_xiaohongshu_notes('护肤品推荐', max_page=2)

2.3 淘宝/天猫商品列表的元素定位策略

淘宝的商品列表用影刀捕获元素经常失败，原因是商品项是动态加载的。正确方法：

指令：获取相似元素列表 靶点：CSS选择器 = .item.J_MouserOnverReq （淘宝商品项的稳定class）

然后用「循环」遍历每个商品项，在循环内用「相对XPath」定位标题、价格、销量：

# 在循环项（当前商品元素）内定位标题 .//div[@class='title']/a/text() # 定位价格（注意价格有整数和小数两部分） .//div[@class='price']/strong/text()

注意点：淘宝价格元素有时候是整数和小数分两个标签，需要用Python拼接：

# 在影刀Python节点中使用price_int=item.xpath('.//strong[@class="price"]/text()')[0].strip()price_decimal=item.xpath('.//span[@class="price-decimal"]/text()')[0].strip()price=price_int+price_decimal# 拼接成完整价格

三、翻页逻辑：三种方案选型

| 翻页类型 | 适用场景 | 推荐方案 | 稳定性 |
|----------|
| 页码翻页（1,2,3…） | 淘宝、京东 | 「点击指定元素」+ 等待元素出现 | ★★★★☆ |
| 滚动加载（无限下拉） | 小红书、抖音 | 「模拟滚动」+ 检测新增元素数量 | ★★★☆☆ |
| 「加载更多」按钮 | 部分移动端页面 | 「点击元素」+ 等待按钮文本变化 | ★★★★☆ |

3.1 滚动加载的翻页代码（小红书实战）

# -*- coding: utf-8 -*-# 功能：模拟滚动加载，检测新增内容数量来判断是否到底defscroll_and_wait(driver,scroll_pause_time=2):""" 模拟页面滚动，返回滚动后新增的元素数量 :param driver: 影刀的浏览器对象（通过参数传入） :param scroll_pause_time: 每次滚动后等待的秒数 :return: 新增笔记数量 """# 先记录当前笔记数量current_count=len(driver.find_elements_by_xpath('//a[contains(@href, "/explore/")]'))# 滚动到页面底部driver.execute_script('window.scrollTo(0, document.body.scrollHeight);')# 等待页面加载importtime time.sleep(scroll_pause_time)# 再记录滚动后的笔记数量new_count=len(driver.find_elements_by_xpath('//a[contains(@href, "/explore/")]'))# 返回新增数量returnnew_count-current_count# 在影刀中的使用方式：# 1. 在「Python节点」中调用 scroll_and_wait(driver)# 2. 如果返回值 <= 0，说明没有新内容了，结束翻页# 3. 如果返回值 > 0，继续滚动

3.2 页码翻页的异常处理

页码翻页最容易出的问题是「点击了但页面没跳转」。解决方法是「点击→等待目标元素出现→验证当前页码」：

指令序列： 1. 【点击元素】-> 下一页按钮 2. 【等待元素出现】-> 当前页码的高亮元素（如 page-current） 3. 【获取元素文本】-> 当前页码，存入变量 current_page 4. 【条件判断】-> 如果 current_page == expected_page，继续；否则截图+重试

四、数据清洗：采集到的脏数据处理

采集到的数据经常有这些问题：表情符号、换行符、多余空格、HTML转义字符。

4.1 用Python节点做数据清洗（完整代码）

# -*- coding: utf-8 -*-# 功能：清洗采集到的电商数据，处理表情、换行、HTML转义等常见问题importreimporthtmldefclean_product_data(raw_data):""" 清洗单条商品数据 :param raw_data: 字典，包含 title/price/ sales等字段 :return: 清洗后的字典 """cleaned=raw_data.copy()# 1. 去除表情符号（正则匹配表情的Unicode范围）emoji_pattern=re.compile('['u'\U0001F600-\U0001F64F'# 表情符号u'\U0001F300-\U0001F5FF'# 符号&象形文字u'\U0001F680-\U0001F6FF'# 交通&地图符号u'\U0001F1E0-\U0001F1FF'# 国旗u'\U00002702-\U000027B0'u'\U000024C2-\U0001F251'']+',flags=re.UNICODE)if'title'incleaned:# 去除表情cleaned['title']=emoji_pattern.sub('',cleaned['title'])# 去除换行符和多余空格cleaned['title']=re.sub(r'\s+',' ',cleaned['title']).strip()# 解码HTML转义（如 &amp; -> &）cleaned['title']=html.unescape(cleaned['title'])# 2. 价格字段：去除￥符号和逗号，转成浮点数if'price'incleaned:price_str=re.sub(r'[￥,，元]','',str(cleaned['price']))try:cleaned['price']=float(price_str)except:cleaned['price']=0.0# 3. 销量字段：把"1万+"转成数字if'sales'incleaned:sales_str=str(cleaned['sales'])if'万'insales_str:num=re.findall(r'[\d.]+',sales_str)[0]cleaned['sales']=int(float(num)*10000)elif'+'insales_str:cleaned['sales']=int(re.findall(r'\d+',sales_str)[0])else:try:cleaned['sales']=int(re.findall(r'\d+',sales_str)[0])except:cleaned['sales']=0returncleaneddefbatch_clean_data(raw_data_list):""" 批量清洗数据列表 :param raw_data_list: 原始数据列表，每个元素是字典 :return: 清洗后的数据列表 """cleaned_list=[]foriteminraw_data_list:try:cleaned=clean_product_data(item)cleaned_list.append(cleaned)exceptExceptionase:print(f'清洗失败，跳过：{item}，错误：{str(e)}')returncleaned_list# 在影刀中的使用：# 1. 把采集到的数据列表存入变量 raw_list# 2. 在Python节点调用：cleaned_list = batch_clean_data(raw_list)# 3. 把 cleaned_list 写入Excel或飞书

五、数据存储：Excel vs 飞书多维表格

5.1 Excel存储的注意事项

用Excel存储数据，最大的坑是「文件被占用」导致写入失败。解决方法：

方案：每次写入前检查文件是否被占用 指令： 1. 【Python节点】-> 尝试 open 文件，如果报错说明被占用 2. 如果被占用，等待3秒重试，最多重试3次 3. 3次都失败，把数据写入备份文件（如 data_backup_日期.xlsx）

5.2 去重逻辑（必须做）

采集系统最怕数据重复。去重要在「写入前」做，不要在写入后处理。

# -*- coding: utf-8 -*-# 功能：在写入前对数据进行去重defdeduplicate_by_field(data_list,unique_field='product_id'):""" 根据唯一字段去重 :param data_list: 数据列表 :param unique_field: 唯一字段名，如 product_id 或 url :return: 去重后的列表 """seen=set()result=[]foritemindata_list:# 取唯一字段的值key=item.get(unique_field,'')ifnotkey:# 如果没有唯一字段，用标题+价格组合作为去重依据key=item.get('title','')+str(item.get('price',''))ifkeynotinseen:seen.add(key)result.append(item)print(f'去重前：{len(data_list)}条，去重后：{len(result)}条，去除重复{len(data_list)-len(result)}条')returnresult

六、异常处理：让系统能自己恢复

6.1 三层异常处理框架

第一层：指令级 try-catch -> 每个关键指令（点击、获取元素）都包一层 try-catch -> 捕获到异常：截图 + 记录日志 + 重试1次 第二层：子流程级异常传播 -> 子流程内部处理不了的异常，抛给调用方 -> 调用方决定是否跳过当前任务继续跑 第三层：全局异常恢复 -> 流程最外层包一个大的 try-catch -> 捕获到未知异常：重启浏览器 + 从断点继续

6.2 断点续跑的实现

断点续跑的核心是「记录进度」。每次处理完一个任务，把进度写入进度文件：

# -*- coding: utf-8 -*-# 功能：实现断点续跑，记录已处理的任务进度importjsonimportosdefsave_progress(progress_file,completed_tasks):""" 保存进度到本地文件 :param progress_file: 进度文件路径，如 C:/rpa/progress.json :param completed_tasks: 已完成任务列表（存URL或任务ID） """withopen(progress_file,'w',encoding='utf-8')asf:json.dump({'completed':completed_tasks,'last_update':time.strftime('%Y-%m-%d %H:%M:%S')},f,ensure_ascii=False,indent=2)defload_progress(progress_file):""" 加载进度，返回已完成的任务列表 :param progress_file: 进度文件路径 :return: 已完成任务列表 """ifnotos.path.exists(progress_file):return[]withopen(progress_file,'r',encoding='utf-8')asf:data=json.load(f)returndata.get('completed',[])deffilter_uncompleted_tasks(all_tasks,completed_tasks):""" 过滤掉已完成的任务，返回待处理任务 :param all_tasks: 所有任务列表 :param completed_tasks: 已完成任务列表 :return: 待处理任务列表 """return[tfortinall_tasksiftnotincompleted_tasks]# 在影刀中的使用流程：# 1. 流程启动时：调用 load_progress() 加载进度# 2. 过滤任务列表：待处理 = filter_uncompleted_tasks(全部任务, 已完成任务)# 3. 处理每个任务后：调用 save_progress() 保存进度# 4. 流程异常中断后重新启动：自动跳过已完成的任务

七、定时运行：影刀定时任务的配置要点

影刀的定时任务看起来简单，但有几个坑要注意：

不要设整点运行：所有人都在整点跑，影刀的云调度队列会排队。设成:07、:23这种非整点时间。
运行时长限制：影刀免费版定时任务单次运行不能超过2小时。长任务要拆分成多个定时任务，用进度文件衔接。
异常自动重启：在定时任务配置里勾选「运行失败自动重试」，重试次数设3次，每次间隔10分钟。

常见问题速查（12条）

#	现象	原因	解决方式
1	元素突然出现「找不到」	页面结构更新，XPath失效	改用CSS选择器或相对稳定属性定位，避免用完整路径XPath
2	流程跑着跑着变慢	内存泄漏，浏览器标签页没关	每处理50条数据关闭当前标签页，或每2小时重启一次浏览器
3	数据重复写入	没有去重逻辑，或去重逻辑有bug	在写入前用`deduplicate_by_field()`做去重，用唯一字段（如商品URL）判断
4	翻页到一半停了	页面没加载完就点了下一页	在点击下一页前，用「等待元素出现」等待页码变化，超时则截图+重试
5	Excel写入失败	文件被Excel软件打开占用	写入前用Python检测文件占用，被占用则写入备份文件，并发送飞书通知
6	采集到的价格是空的	价格是动态加载的，捕获时还没渲染出来	在获取价格前加「等待元素出现」，等待时间设3秒；还不行就用Python直接解析网页源码
7	定时任务到点没跑	电脑休眠了，影刀客户端没在运行	设置电脑「永不休眠」，并勾选影刀「开机自启动」和「客户端关闭后继续运行定时任务」
8	小红书采集几页后限流	IP被识别为爬虫，触发验证码	降低采集频率（每页间隔5-8秒），使用代理IP轮换，遇到验证码停止当前任务
9	数据清洗后中文乱码	Excel打开CSV时编码不对	用`utf-8-sig`编码写CSV（Python：`open(file, 'w', encoding='utf-8-sig')`），Excel打开就不会乱码
10	流程报错但不知道哪里错了	没有日志记录	在关键步骤用「打印日志」指令输出变量值，并配置日志写入本地文件
11	子流程参数传递失败	对象变量的字段名写错了	用「调试模式」运行，在变量面板检查对象变量的实际字段名，确保和子流程参数名一致
12	浏览器启动时很慢	加载了太多扩展插件	在影刀浏览器配置里，勾选「使用干净浏览器环境」，不加载用户插件