影刀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[classnote-item]//a[classcover mask]/href但这还不够稳因为小红书的class会带哈希后缀。升级版//a[contains(href, /explore/) and target_blank]完整可运行的Python代码用影刀Python节点执行# -*- coding: utf-8 -*-# 功能用 requests lxml 解析小红书搜索页绕过影刀元素捕获的限制# 适用场景小红书搜索结果页批量获取笔记链接importrequestsfromlxmlimportetreeimporttimeimportrandomdeffetch_xiaohongshu_notes(keyword,max_page3): 采集小红书搜索结果中的笔记链接 :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,# ← 替换成你的CookieReferer:https://www.xiaohongshu.com/,}forpageinrange(1,max_page1):# 随机延迟避免被限流time.sleep(random.uniform(2,4))urlfhttps://www.xiaohongshu.com/search_result?keyword{keyword}sourceweb_search_result_notespage{page}try:resprequests.get(url,headersheaders,timeout10)# 检查状态码如果不是200说明被限流了ifresp.status_code!200:print(f第{page}页请求失败状态码{resp.status_code})continuehtmletree.HTML(resp.text)# XPath找所有包含 /explore/ 的链接小红书笔记链接特征linkshtml.xpath(//a[contains(href, /explore/)]/href)# 去重补全链接forlinkinlinks:full_linkhttps://www.xiaohongshu.comlinkiflink.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_page2)2.3 淘宝/天猫商品列表的元素定位策略淘宝的商品列表用影刀捕获元素经常失败原因是商品项是动态加载的。正确方法指令获取相似元素列表 靶点CSS选择器 .item.J_MouserOnverReq 淘宝商品项的稳定class然后用「循环」遍历每个商品项在循环内用「相对XPath」定位标题、价格、销量# 在循环项当前商品元素内定位标题 .//div[classtitle]/a/text() # 定位价格注意价格有整数和小数两部分 .//div[classprice]/strong/text()注意点淘宝价格元素有时候是整数和小数分两个标签需要用Python拼接# 在影刀Python节点中使用price_intitem.xpath(.//strong[classprice]/text())[0].strip()price_decimalitem.xpath(.//span[classprice-decimal]/text())[0].strip()priceprice_intprice_decimal# 拼接成完整价格三、翻页逻辑三种方案选型| 翻页类型 | 适用场景 | 推荐方案 | 稳定性 ||----------|| 页码翻页1,2,3… | 淘宝、京东 | 「点击指定元素」 等待元素出现 | ★★★★☆ || 滚动加载无限下拉 | 小红书、抖音 | 「模拟滚动」 检测新增元素数量 | ★★★☆☆ || 「加载更多」按钮 | 部分移动端页面 | 「点击元素」 等待按钮文本变化 | ★★★★☆ |3.1 滚动加载的翻页代码小红书实战# -*- coding: utf-8 -*-# 功能模拟滚动加载检测新增内容数量来判断是否到底defscroll_and_wait(driver,scroll_pause_time2): 模拟页面滚动返回滚动后新增的元素数量 :param driver: 影刀的浏览器对象通过参数传入 :param scroll_pause_time: 每次滚动后等待的秒数 :return: 新增笔记数量 # 先记录当前笔记数量current_countlen(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_countlen(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: 清洗后的字典 cleanedraw_data.copy()# 1. 去除表情符号正则匹配表情的Unicode范围emoji_patternre.compile([u\U0001F600-\U0001F64F# 表情符号u\U0001F300-\U0001F5FF# 符号象形文字u\U0001F680-\U0001F6FF# 交通地图符号u\U0001F1E0-\U0001F1FF# 国旗u\U00002702-\U000027B0u\U000024C2-\U0001F251],flagsre.UNICODE)iftitleincleaned:# 去除表情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. 价格字段去除符号和逗号转成浮点数ifpriceincleaned:price_strre.sub(r[,元],,str(cleaned[price]))try:cleaned[price]float(price_str)except:cleaned[price]0.0# 3. 销量字段把1万转成数字ifsalesincleaned:sales_strstr(cleaned[sales])if万insales_str:numre.findall(r[\d.],sales_str)[0]cleaned[sales]int(float(num)*10000)elifinsales_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:cleanedclean_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_日期.xlsx5.2 去重逻辑必须做采集系统最怕数据重复。去重要在「写入前」做不要在写入后处理。# -*- coding: utf-8 -*-# 功能在写入前对数据进行去重defdeduplicate_by_field(data_list,unique_fieldproduct_id): 根据唯一字段去重 :param data_list: 数据列表 :param unique_field: 唯一字段名如 product_id 或 url :return: 去重后的列表 seenset()result[]foritemindata_list:# 取唯一字段的值keyitem.get(unique_field,)ifnotkey:# 如果没有唯一字段用标题价格组合作为去重依据keyitem.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,encodingutf-8)asf:json.dump({completed:completed_tasks,last_update:time.strftime(%Y-%m-%d %H:%M:%S)},f,ensure_asciiFalse,indent2)defload_progress(progress_file): 加载进度返回已完成的任务列表 :param progress_file: 进度文件路径 :return: 已完成任务列表 ifnotos.path.exists(progress_file):return[]withopen(progress_file,r,encodingutf-8)asf:datajson.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选择器或相对稳定属性定位避免用完整路径XPath2流程跑着跑着变慢内存泄漏浏览器标签页没关每处理50条数据关闭当前标签页或每2小时重启一次浏览器3数据重复写入没有去重逻辑或去重逻辑有bug在写入前用deduplicate_by_field()做去重用唯一字段如商品URL判断4翻页到一半停了页面没加载完就点了下一页在点击下一页前用「等待元素出现」等待页码变化超时则截图重试5Excel写入失败文件被Excel软件打开占用写入前用Python检测文件占用被占用则写入备份文件并发送飞书通知6采集到的价格是空的价格是动态加载的捕获时还没渲染出来在获取价格前加「等待元素出现」等待时间设3秒还不行就用Python直接解析网页源码7定时任务到点没跑电脑休眠了影刀客户端没在运行设置电脑「永不休眠」并勾选影刀「开机自启动」和「客户端关闭后继续运行定时任务」8小红书采集几页后限流IP被识别为爬虫触发验证码降低采集频率每页间隔5-8秒使用代理IP轮换遇到验证码停止当前任务9数据清洗后中文乱码Excel打开CSV时编码不对用utf-8-sig编码写CSVPythonopen(file, w, encodingutf-8-sig)Excel打开就不会乱码10流程报错但不知道哪里错了没有日志记录在关键步骤用「打印日志」指令输出变量值并配置日志写入本地文件11子流程参数传递失败对象变量的字段名写错了用「调试模式」运行在变量面板检查对象变量的实际字段名确保和子流程参数名一致12浏览器启动时很慢加载了太多扩展插件在影刀浏览器配置里勾选「使用干净浏览器环境」不加载用户插件推荐资源影刀官方文档 - XPath元素定位https://academy.yingdao.com/Chrome开发者工具教程https://developer.chrome.com/docs/devtools/Python lxml库文档解析HTMLhttps://lxml.de/小红书反爬虫策略分析知乎搜索关键词「小红书爬虫 2024」内容标签影刀RPA电商数据采集XPath元素定位Python节点数据清洗异常处理定时任务断点续跑反爬虫系统架构转载请注明出处飞行社社区作者林焱