影刀RPA深度实战多平台商品价格监控系统搭建与自动告警方案作者林焱做电商的都知道价格监控是个苦力活——每天手动去淘宝、拼多多、京东搜同一个商品看价格有没有变化有的话记录下来价格变动大的还要通知运营。我之前帮一个电商运营团队搭过这套价格监控系统把每天1小时的手工活变成了全自动。这篇文章把整套系统的搭建方法写出来。一、系统整体设计1.1 监控流程配置监控商品列表 → 定时采集各平台价格 → 和上次价格对比 ↓ 价格有变化 → 记录历史 → 判断是否告警 → 发送飞书通知 ↓ 价格无变化 → 继续监控不记录1.2 数据表设计在飞书多维表格里创建「价格监控」数据表字段设计字段名类型说明商品名称文本监控的商品名称如iPhone 15 Pro Max 256G商品ID文本唯一标识用于去重淘宝价格数字最新淘宝价格拼多多价格数字最新拼多多价格京东价格数字最新京东价格最低价平台单选淘宝/拼多多/京东最后更新时间日期最后一次价格更新时间价格历史文本长JSON格式的历史价格记录告警状态单选正常/已告警/已处理二、多平台价格采集2.1 淘宝价格采集# -*- coding: utf-8 -*-# 功能采集淘宝商品价格importrequestsfromlxmlimportetreeimporttimeimportrandomdeffetch_taobao_price(keyword,cookie): 采集淘宝商品价格搜索结果第一个商品的价格 :param keyword: 搜索关键词 :param cookie: 淘宝Cookie :return: 价格浮点数和商品URL headers{User-Agent:get_random_user_agent(),# 复用反爬虫文章里的函数Cookie:cookie,Referer:https://s.taobao.com/,}# 淘宝搜索URLsearch_urlfhttps://s.taobao.com/search?q{keyword}sortsale-desctry:resprequests.get(search_url,headersheaders,timeout15)htmletree.HTML(resp.text)# 提取第一个商品的价格# 淘宝的价格数据在页面的JavaScript里需要从JSON里解析# 简化方案用正则提取价格数据importre price_matchre.search(rprice:([\d.]),resp.text)ifprice_match:pricefloat(price_match.group(1))# 提取商品URLurl_matchre.search(rnid:(\d),resp.text)ifurl_match:item_idurl_match.group(1)item_urlfhttps://item.taobao.com/item.htm?id{item_id}else:item_urlprint(f淘宝价格采集成功¥{price})returnprice,item_urlelse:print(淘宝价格提取失败)returnNone,NoneexceptExceptionase:print(f淘宝价格采集异常{str(e)})returnNone,None2.2 拼多多价格采集# -*- coding: utf-8 -*-# 功能采集拼多多商品价格deffetch_pinduoduo_price(keyword,cookie): 采集拼多多商品价格 :param keyword: 搜索关键词 :param cookie: 拼多多Cookie :return: 价格浮点数 headers{User-Agent:get_random_user_agent(),Cookie:cookie,Referer:https://mobile.yangkeduo.com/,}# 拼多多H5搜索URLsearch_urlfhttps://mobile.yangkeduo.com/search_result.html?search_key{keyword}try:resprequests.get(search_url,headersheaders,timeout15)htmletree.HTML(resp.text)# 拼多多的价格数据在页面JSON里# 提取商品价格第一个商品price_elemhtml.xpath(//div[classprice]/text())ifprice_elem:price_strprice_elem[0].strip().replace(¥,)pricefloat(price_str)print(f拼多多价格采集成功¥{price})returnpriceelse:print(拼多多价格提取失败)returnNoneexceptExceptionase:print(f拼多多价格采集异常{str(e)})returnNone2.3 京东价格采集# -*- coding: utf-8 -*-# 功能采集京东商品价格deffetch_jd_price(keyword,cookie): 采集京东商品价格 :param keyword: 搜索关键词 :param cookie: 京东Cookie :return: 价格浮点数 headers{User-Agent:get_random_user_agent(),Cookie:cookie,Referer:https://search.jd.com/,}# 京东搜索URLsearch_urlfhttps://search.jd.com/Search?keyword{keyword}psort3try:resprequests.get(search_url,headersheaders,timeout15)htmletree.HTML(resp.text)# 提取第一个商品的价格# 京东价格元素的class是动态的用英文数字需要用containsprice_elemhtml.xpath(//i[contains(class, price)]/text())ifprice_elem:price_strprice_elem[0].strip().replace(¥,)# 京东价格可能有小数点pricefloat(price_str)print(f京东价格采集成功¥{price})returnpriceelse:print(京东价格提取失败)returnNoneexceptExceptionase:print(f京东价格采集异常{str(e)})returnNone三、价格对比与历史记录3.1 价格对比逻辑# -*- coding: utf-8 -*-# 功能对比价格变化判断是否告警importjsonimporttimedefcompare_prices(product_id,new_prices): 对比新旧价格记录变化 :param product_id: 商品ID :param new_prices: 新价格字典 {taobao: 99.9, pinduoduo: 89.9, jd: 109.9} :return: 价格变化信息和是否需要告警 # 加载历史价格history_filefC:/rpa/price_history/{product_id}.jsontry:withopen(history_file,r,encodingutf-8)asf:historyjson.load(f)except:history{product_id:product_id,records:[]# 历史记录列表}# 获取上一次的价格last_recordhistory[records][-1]ifhistory[records]elseNone# 对比每个平台的价格changes[]need_alertFalseforplatform,new_priceinnew_prices.items():ifnew_priceisNone:continueiflast_recordisNone:# 第一次采集没有对比数据changes.append({platform:platform,old_price:None,new_price:new_price,change:None,change_pct:None})else:old_pricelast_record[prices].get(platform)ifold_priceisNone:changes.append({platform:platform,old_price:None,new_price:new_price,change:None,change_pct:None})else:price_changenew_price-old_price change_pct(price_change/old_price*100)ifold_price0else0changes.append({platform:platform,old_price:old_price,new_price:new_price,change:price_change,change_pct:change_pct})# 判断是否需要告警价格变动超过10%ifabs(change_pct)10:need_alertTrue# 保存新价格到历史记录new_record{timestamp:int(time.time()),date:time.strftime(%Y-%m-%d %H:%M:%S),prices:new_prices,changes:changes}history[records].append(new_record)# 只保留最近30天的历史防止文件太大iflen(history[records])30*24:# 假设每小时采集一次history[records]history[records][-30*24:]# 保存历史记录importos os.makedirs(os.path.dirname(history_file),exist_okTrue)withopen(history_file,w,encodingutf-8)asf:json.dump(history,f,ensure_asciiFalse,indent2)return{changes:changes,need_alert:need_alert,history:history}3.2 找出最低价平台# -*- coding: utf-8 -*-# 功能找出最低价平台deffind_lowest_price_platform(prices): 找出最低价的平台 :param prices: 价格字典 {taobao: 99.9, pinduoduo: 89.9, jd: None} :return: 最低价平台名称和价格 # 过滤掉 None 值valid_prices{k:vfork,vinprices.items()ifvisnotNone}ifnotvalid_prices:returnNone,Nonelowest_platformmin(valid_prices,keyvalid_prices.get)lowest_pricevalid_prices[lowest_platform]returnlowest_platform,lowest_price四、告警规则配置4.1 告警规则规则触发条件告警级别价格大幅下跌某个平台价格下跌超过10%高级价格大幅上涨某个平台价格上涨超过10%中级出现更低价格某个平台出现了比之前更低的 Price低级价格恢复到原价之前降价了现在恢复原价低级4.2 告警判断代码# -*- coding: utf-8 -*-# 功能判断是否需要告警并生成告警消息defcheck_alert_rules(changes,product_name): 检查告警规则 :param changes: 价格变化列表 :param product_name: 商品名称 :return: 告警消息列表 alerts[]forchangeinchanges:platformchange[platform]old_pricechange[old_price]new_pricechange[new_price]change_pctchange.get(change_pct)ifold_priceisNone:# 第一次采集不需要告警continue# 规则1价格大幅变动超过10%ifchange_pctisnotNoneandabs(change_pct)10:direction下跌ifchange_pct0else上涨alert{level:highifabs(change_pct)20elsemedium,message:f【{platform}】{product_name}价格{direction}{abs(change_pct):.1f}%\nf原价¥{old_price:.2f}\nf现价¥{new_price:.2f}\nf变动¥{change[change]:.2f}}alerts.append(alert)# 规则2价格恢复到原价之前有降价现在涨回去了# 需要检查历史记录里是否有更低的价格# 这里简化只检查上一次价格returnalerts五、飞书消息推送5.1 发送价格变动通知# -*- coding: utf-8 -*-# 功能发送价格监控告警通知到飞书defsend_price_alert(product_name,alerts,webhook_url): 发送价格告警通知 :param product_name: 商品名称 :param alerts: 告警列表 :param webhook_url: 飞书机器人 Webhook ifnotalerts:return# 生成消息内容contentf⚠️ 价格监控告警\n\n商品{product_name}\n\nforalertinalerts:level_emojiifalert[level]highelsecontentf{level_emoji}{alert[message]}\n\ncontentf---\n监控时间{time.strftime(%Y-%m-%d %H:%M:%S)}# 发送飞书消息send_feishu_notification(webhook_url,content)defsend_daily_price_report(products,webhook_url): 发送每日价格汇总报告 contentf 每日价格监控报告{time.strftime(%Y-%m-%d)}\n\nforproductinproducts:nameproduct[name]pricesproduct[prices]lowest_platform,lowest_pricefind_lowest_price_platform(prices)contentf【{name}】\ncontentf 淘宝¥{prices.get(taobao,N/A)}\ncontentf 拼多多¥{prices.get(pinduoduo,N/A)}\ncontentf 京东¥{prices.get(jd,N/A)}\ncontentf 最低价{lowest_platform}¥{lowest_price:.2f}\n\nsend_feishu_notification(webhook_url,content)六、完整流程编排【主流程价格监控】 1. 【Python节点】- 加载监控商品列表 输出product_list列表 2. 【循环】- 遍历 product_list 中的每个商品 3. 【Python节点】- 采集淘宝价格 调用fetch_taobao_price(keyword, cookie) 输出taobao_price 4. 【Python节点】- 采集拼多多价格 调用fetch_pinduoduo_price(keyword, cookie) 输出pinduoduo_price 5. 【Python节点】- 采集京东价格 调用fetch_jd_price(keyword, cookie) 输出jd_price 6. 【Python节点】- 对比价格变化 调用compare_prices(product_id, new_prices) 输出comparison_result 7. 【条件判断】- 如果 comparison_result[need_alert] True a. 【Python节点】- 生成告警消息 b. 【飞书通知】- 发送告警 8. 【Python节点】- 更新飞书多维表格 调用update_feishu_price_table() 9. 【等待】- 随机延迟 30-60 秒 10.【循环结束】 11.【条件判断】- 如果当前时间是每天早上 9 点 【飞书通知】- 发送每日价格汇总报告常见问题速查#现象原因解决方式1价格采集不准确页面数据是动态加载的捕获时还没渲染在采集前加等待时间或用API直接获取数据2拼多多价格采集失败率高拼多多反爬虫比较严格降低采集频率或用多个账号轮换采集3历史价格文件太大采集频率高历史记录多只保留最近30天的数据 older数据归档到备份文件4飞书告警消息太频繁价格波动频繁每次都告警增加告警冷却时间如同一个商品24小时内只告警一次5多平台价格对比不准确不同平台的商品可能不是同一个规格在监控商品列表里明确指定商品规格如256G 黑色6定时任务运行失败电脑休眠或网络断开设置电脑永不休眠并在流程里加网络检测逻辑7Cookie频繁失效平台登出或Cookie过期定期手动更新Cookie或在流程里加Cookie有效性检测推荐资源淘宝开放平台商品APIhttps://open.taobao.com/京东宙斯开放平台https://jos.jd.com/飞书自定义机器人配置https://open.feishu.cn/内容标签影刀RPA价格监控多平台对比淘宝采集拼多多采集京东采集价格告警飞书通知历史记录定时任务转载请注明出处飞行社社区作者林焱