影刀RPA 自动化竞品监控系统：价格与产品变动实时追踪

影刀RPA 自动化竞品监控系统价格与产品变动实时追踪作者林焱写在前面竞品监控是运营和产品团队的日常工作但靠人工每天盯竞品官网、电商页面极其低效。本文搭建一套完整的竞品监控系统定时采集竞品数据 → 与历史数据对比 → 发现变化 → 推送告警全程自动化。一、监控哪些内容拼多多店群自动化上架方案# 竞品监控指标清单MONITOR_ITEMS{价格监控:[单品价格,套餐价格,活动价格,折扣力度],产品变动:[新品上市,产品下架,产品规格变化,功能更新],营销活动:[促销活动,满减规则,优惠券,限时折扣],流量数据:[搜索排名,评价数量,销量变化],内容变化:[官网首页,产品介绍页,招聘页面推断业务方向],}二、电商竞品价格监控importrequestsfrombs4importBeautifulSoupimportjsonimportsqlite3importdatetimeimporthashlibclassCompetitorMonitor:def__init__(self,db_pathcompetitor_data.db):self.db_pathdb_path self.connsqlite3.connect(db_path)self._init_db()def_init_db(self):初始化数据库cursorself.conn.cursor()# 产品监控表cursor.execute( CREATE TABLE IF NOT EXISTS product_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, competitor TEXT NOT NULL, product_id TEXT NOT NULL, product_name TEXT, price REAL, original_price REAL, stock_status TEXT, rating REAL, review_count INTEGER, snapshot_hash TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ))# 变化记录表cursor.execute( CREATE TABLE IF NOT EXISTS changes ( id INTEGER PRIMARY KEY AUTOINCREMENT, competitor TEXT, product_id TEXT, change_type TEXT, old_value TEXT, new_value TEXT, detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ))self.conn.commit()defget_last_snapshot(self,competitor,product_id):获取最近一次的快照cursorself.conn.cursor()cursor.execute( SELECT * FROM product_snapshots WHERE competitor? AND product_id? ORDER BY created_at DESC LIMIT 1 ,(competitor,product_id))rowcursor.fetchone()ifrow:columns[d[0]fordincursor.description]returndict(zip(columns,row))returnNonedefsave_snapshot(self,competitor,product_id,data):保存当前快照snapshot_contentjson.dumps(data,sort_keysTrue)snapshot_hashhashlib.md5(snapshot_content.encode()).hexdigest()cursorself.conn.cursor()cursor.execute( INSERT INTO product_snapshots (competitor, product_id, product_name, price, original_price, stock_status, rating, review_count, snapshot_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ,(competitor,product_id,data.get(product_name),data.get(price),data.get(original_price),data.get(stock_status),data.get(rating),data.get(review_count),snapshot_hash))self.conn.commit()returnsnapshot_hashdefdetect_changes(self,competitor,product_id,old_data,new_data):检测数据变化并记录changes[]# 价格变化检测old_priceold_data.get(price)new_pricenew_data.get(price)ifold_priceandnew_priceandabs(new_price-old_price)/old_price0.01:# 1%以上变化change_pct(new_price-old_price)/old_price*100change{type:price_change,description:f价格变化{old_price}→{new_price}{change_pct:.1f}%,severity:highifabs(change_pct)10elsemedium}changes.append(change)cursorself.conn.cursor()cursor.execute( INSERT INTO changes (competitor, product_id, change_type, old_value, new_value) VALUES (?, ?, ?, ?, ?) ,(competitor,product_id,price,str(old_price),str(new_price)))self.conn.commit()# 缺货检测old_stockold_data.get(stock_status)new_stocknew_data.get(stock_status)ifold_stock!new_stock:changes.append({type:stock_change,description:f库存状态变化{old_stock}→{new_stock},severity:medium})returnchanges三、实战京东商品价格监控defscrape_jd_product(product_id,sessionNone): 采集京东商品信息 product_id: 商品IDURL中的数字 ifsessionisNone:sessionrequests.Session()session.headers.update({User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36,Referer:https://www.jd.com/})urlfhttps://item.jd.com/{product_id}.htmltry:respsession.get(url,timeout10)soupBeautifulSoup(resp.text,html.parser)# 提取商品名name_tagsoup.select_one(.sku-name)product_namename_tag.text.strip()ifname_tagelse# 价格需要通过API获取京东实时价格是异步加载的price_urlfhttps://p.3.cn/prices/mgets?skuidJ_{product_id}price_respsession.get(price_url,timeout5)price_datajson.loads(price_resp.text.strip(/**/))current_pricefloat(price_data[0].get(p,0))ifprice_dataelse0original_pricefloat(price_data[0].get(op,current_price))ifprice_dataelsecurrent_price# 评价数需要额外接口comment_urlfhttps://club.jd.com/comment/productCommentSummaries.action?referenceIds{product_id}comment_respsession.get(comment_url,timeout5)comment_datacomment_resp.json()review_countcomment_data.get(CommentsCount,[{}])[0].get(ShowCount,0)return{product_name:product_name,price:current_price,original_price:original_price,review_count:review_count,url:url}exceptExceptionase:print(f京东采集失败{product_id}{e})returnNonedefmonitor_competitor_products():监控竞品商品列表# 竞品商品配置COMPETITOR_PRODUCTS[{competitor:竞品A,platform:jd,product_id:100012345678},{competitor:竞品A,platform:jd,product_id:100023456789},{competitor:竞品B,platform:jd,product_id:100034567890},]monitorCompetitorMonitor()sessionrequests.Session()session.headers.update({User-Agent:Mozilla/5.0 ...})all_changes[]forproduct_configinCOMPETITOR_PRODUCTS:competitorproduct_config[competitor]product_idproduct_config[product_id]# 采集当前数据current_datascrape_jd_product(product_id,session)ifnotcurrent_data:continueprint(f{competitor}-{current_data[product_name][:20]}: ¥{current_data[price]})# 获取历史数据检测变化last_snapshotmonitor.get_last_snapshot(competitor,product_id)iflast_snapshot:changesmonitor.detect_changes(competitor,product_id,last_snapshot,current_data)ifchanges:forchangeinchanges:change[competitor]competitor change[product_name]current_data.get(product_name,)[:30]change[current_price]current_data.get(price)all_changes.append(change)print(f ⚡ 变化检测{change[description]})# 保存新快照monitor.save_snapshot(competitor,product_id,current_data)importtime time.sleep(1)# 采集间隔returnall_changes四、官网内容变化监控defmonitor_website_content(urls,keyword_filtersNone): 监控竞品官网内容变化 通过对比页面内容hash来检测变化 sessionrequests.Session()session.headers.update({User-Agent:Mozilla/5.0 ...})changes[]snapshots_filewebsite_snapshots.json# 读取历史快照try:withopen(snapshots_file,r,encodingutf-8)asf:old_snapshotsjson.load(f)except:old_snapshots{}new_snapshots{}forurl_configinurls:urlurl_config[url]nameurl_config.get(name,url)try:respsession.get(url,timeout15)soupBeautifulSoup(resp.text,html.parser)# 提取正文文字排除导航、页脚等噪声fortaginsoup.find_all([nav,footer,script,style,header]):tag.decompose()page_textsoup.get_text(separator\n,stripTrue)# 过滤关键词只关注包含特定词汇的段落ifkeyword_filters:relevant_lines[lineforlineinpage_text.split(\n)ifany(kwinlineforkwinkeyword_filters)]page_text\n.join(relevant_lines)content_hashhashlib.md5(page_text.encode()).hexdigest()new_snapshots[url]{hash:content_hash,text_preview:page_text[:500],updated_at:datetime.datetime.now().isoformat()}# 与历史对比ifurlinold_snapshots:ifold_snapshots[url][hash]!content_hash:changes.append({type:content_change,name:name,url:url,description:f官网内容发生变化{name},severity:medium})print(f⚡ 内容变化{name})exceptExceptionase:print(f采集失败{url}-{e})# 保存新快照withopen(snapshots_file,w,encodingutf-8)asf:json.dump(new_snapshots,f,ensure_asciiFalse,indent2)returnchanges五、变化告警推送defsend_change_alert(all_changes,webhook_url):汇总并推送变化告警ifnotall_changes:print(本次监控无变化)returnhigh_changes[cforcinall_changesifc.get(severity)high]medium_changes[cforcinall_changesifc.get(severity)medium]contentf## 竞品监控告警\n\ncontentf**监控时间**{datetime.datetime.now().strftime(%Y-%m-%d %H:%M)}\ncontentf**变化总数**{len(all_changes)}项高优先{len(high_changes)}项\n\nifhigh_changes:content### ⚠️ 高优先级变化\nforchangeinhigh_changes:contentf- **{change.get(competitor,)}** |{change.get(description,)}\ncontent\nifmedium_changes:content### ℹ️ 一般变化\nforchangeinmedium_changes[:5]:contentf-{change.get(description,)}\niflen(medium_changes)5:contentf- 等{len(medium_changes)-5}项更多...\n[video(video-spylf4IR-1784822642246)(type-csdn)(url-https://live.csdn.net/v/embed/524993)(image-https://v-blog.csdnimg.cn/asset/a547123d88ad712dccba346c9217e237/cover/Cover0.jpg)(title-TEMU店群如何管理运营)]importrequests payload{msgtype:markdown,markdown:{content:content}}requests.post(webhook_url,jsonpayload)defrun_full_monitoring():完整的竞品监控流程print(f开始竞品监控{datetime.datetime.now().strftime(%Y-%m-%d %H:%M)})# 1. 电商价格监控price_changesmonitor_competitor_products()# 2. 官网内容监控competitor_urls[{name:竞品A官网,url:https://competitor-a.com/products},{name:竞品B定价页,url:https://competitor-b.com/pricing},]content_changesmonitor_website_content(competitor_urls,keyword_filters[价格,新品,发布])# 3. 汇总告警all_changesprice_changescontent_changes WEBHOOKhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的keysend_change_alert(all_changes,WEBHOOK)run_full_monitoring()六、踩坑记录坑1电商价格异步加载京东、天猫的价格不在页面HTML里是JS动态加载的。不能直接抓HTML要找价格API接口通过抓包或查看网络请求单独请求。坑2官网内容频繁细微变化广告位、时间戳、随机内容会导致每次都检测到变化。解决方案在提取文字时过滤掉这些动态元素只关注产品相关内容区域。坑3IP被封频繁采集竞品网站会触发封IP。关键对策合理设置采集间隔每个URL至少间隔1分钟、用代理IP轮换、User-Agent随机化。坑4需要登录才能看到价格会员价、登录后价格需要维护一个已登录的Sessioncookies。用Playwright登录一次后导出cookies在requests.Session里设置cookies使用。总结竞品监控系统的核心定时采集历史对比即时告警。价格监控优先用电商平台API接口稳定内容监控用hash对比简单有效。告警要有优先级不然信息轰炸会被忽略。署名林焱