用OpenClaw做自动化数据采集：定时抓竞品+自动入库+日报推送，解放双手

做产品和运营的朋友应该都有这种体会：每天上班第一件事，就是打开七八个竞品站点，挨个扒价格、上新、销量数据，复制粘贴进表格，再汇总成日报发到群里。整套流程下来四五十分钟就没了，全是机械重复的劳动，还经常因为页面加载慢、数据多抄错数。

之前也试过写Python脚本跑采集，但问题也很明显：站点一改页面结构脚本就崩，要自己维护定时任务、异常重试、告警通知，折腾半天最后还是得每天盯着执行结果，反而多了运维成本。

直到用OpenClaw搭了这套全自动采集链路，从定时抓取、数据清洗入库到日报推送全流程无人值守，跑了两个多月只出过两次小问题，每天实打实省出一个多小时干正事。今天把完整的落地步骤和踩过的坑整理出来，照着做你也能搭出自己的自动化采集流水线。

一、整体方案架构设计

整套方案的核心思路是：让OpenClaw承担调度、执行、异常处理的全部脏活，我们只需要封装最核心的业务逻辑，不用操心调度框架、重试机制、告警通知这些基础设施。

整体分为四层，各司其职：

执行链路很清晰：定时任务到点触发工作流 → 先执行采集拉取原始数据 → 清洗去重后写入数据库 → 取出当日数据生成对比日报 → 推送到业务群。全程自带异常重试和状态追踪，某一步失败会自动重试，重试不成就发告警通知人工介入。

二、前期准备工作

正式动手之前先把环境和依赖备齐，避免中途卡壳：

1. 一套可用的OpenClaw运行环境，本地Windows或Linux服务器都可以，推荐放服务器上7x24小时运行
2. 目标竞品站点的字段梳理：确定要采集哪些核心指标（商品名、价格、销量、上架时间等）
3. MySQL数据库（5.7及以上版本），用来存历史数据做趋势对比
4. 企业微信/钉钉群机器人Webhook，用来推送日报和告警
5. Python基础依赖：requests、pymysql、pandas，用来写Skill的业务逻辑

三、分步落地：从零搭建完整自动化链路

3.1 第一步：封装竞品数据采集Skill

OpenClaw的Skill是最小能力单元，本质就是“给AI看的使用说明 + 可执行的业务脚本”。我们先把采集逻辑封装成独立Skill，方便后续工作流调用。

每个Skill由两部分组成：SKILL.md元信息文件（告诉AI这个技能怎么用）+ 业务执行脚本。

在工作区技能目录下创建competitor-spider文件夹，先写SKILL.md：

---name:competitor-spiderdescription:采集指定竞品站点的商品数据，返回商品名称、价格、销量、上架时间等结构化字段user-invocable:truemetadata:openclaw:emoji:"📊"requires:bins:["python3"]env:[]---## 调用方式输入目标站点标识，返回结构化的竞品商品数据。 参数：-site:目标站点名称，可选值 siteA / siteB / siteC-page:采集页数，默认前3页## 返回格式返回JSON数组，每个元素包含 title、price、sales、publish_time、url 字段。

然后写核心采集脚本spider.py，这里以通用的列表页采集为例，加入随机UA和延迟，规避基础反爬：

importrequestsimportrandomimporttimeimportjsonfrombs4importBeautifulSoup UA_LIST=["Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36","Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"]deffetch_site_data(site:str,max_page:int=3)->list:base_url=SITE_CONFIG[site]["base_url"]result=[]forpageinrange(1,max_page+1):headers={"User-Agent":random.choice(UA_LIST)}resp=requests.get(f"{base_url}?page={page}",headers=headers,timeout=10)soup=BeautifulSoup(resp.text,"html.parser")items=soup.select(".product-item")foriteminitems:result.append({"title":item.select_one(".title").text.strip(),"price":float(item.select_one(".price").text.replace("¥","")),"sales":int(item.select_one(".sales").text.replace("已售","")),"url":item.select_one("a")["href"],"crawl_time":time.strftime("%Y-%m-%d %H:%M:%S")})time.sleep(random.uniform(1,2))returnresultif__name__=="__main__":importsys site=sys.argv[1]iflen(sys.argv)>1else"siteA"page=int(sys.argv[2])iflen(sys.argv)>2else3data=fetch_site_data(site,page)print(json.dumps(data,ensure_ascii=False))

写完后执行openclaw skills reload加载技能，就可以在对话里直接调用了。

3.2 第二步：封装数据清洗与入库Skill

采集到的原始数据不能直接用，需要做去重、格式校验、增量判断，再写入数据库。我们再封装一个入库Skill。

先在MySQL建表，存每日采集快照：

CREATETABLE`competitor_product`(`id`intunsignedNOTNULLAUTO_INCREMENT,`site`varchar(32)NOTNULLCOMMENT'站点标识',`title`varchar(255)NOTNULLCOMMENT'商品名称',`price`decimal(10,2)NOTNULLCOMMENT'当前价格',`sales`intNOTNULLCOMMENT'累计销量',`url`varchar(512)NOTNULLCOMMENT'商品链接',`crawl_date`dateNOTNULLCOMMENT'采集日期',`crawl_time`datetimeNOTNULLCOMMENT'采集时间',PRIMARYKEY(`id`),UNIQUEKEY`uk_site_url_date`(`site`,`url`(255),`crawl_date`))ENGINE=InnoDBDEFAULTCHARSET=utf8mb4;

然后写入库Skill的执行脚本，核心做三件事：数据格式校验、去重判断、批量写入，加上异常捕获和连接重连，避免数据库超时导致任务失败。

importpymysqlimportjsonimportsysimporttimefromdatetimeimportdate DB_CONFIG={"host":"127.0.0.1","port":3306,"user":"root","password":"your_password","database":"data_center","charset":"utf8mb4"}defsave_to_db(site:str,data:list)->dict:conn=pymysql.connect(**DB_CONFIG)cursor=conn.cursor()today=date.today()insert_count=0skip_count=0try:foritemindata:# 增量判断：当日同链接数据已存在则跳过check_sql="SELECT id FROM competitor_product WHERE site=%s AND url=%s AND crawl_date=%s"cursor.execute(check_sql,(site,item["url"],today))ifcursor.fetchone():skip_count+=1continueinsert_sql=""" INSERT INTO competitor_product (site, title, price, sales, url, crawl_date, crawl_time) VALUES (%s, %s, %s, %s, %s, %s, %s) """cursor.execute(insert_sql,(site,item["title"],item["price"],item["sales"],item["url"],today,item["crawl_time"]))insert_count+=1conn.commit()finally:cursor.close()conn.close()return{"insert":insert_count,"skip":skip_count,"total":len(data)}if__name__=="__main__":site=sys.argv[1]raw_data=json.loads(sys.argv[2])result=save_to_db(site,raw_data)print(json.dumps(result,ensure_ascii=False))

3.3 第三步：封装日报生成与推送Skill

数据落库后，需要自动生成对比日报，推送到业务群。日报不用太复杂，核心信息到位就行：当日新增商品数、价格变动Top5、销量涨幅Top5，再附一个数据概览。

推送用企业微信群机器人的Webhook最省事，钉钉逻辑完全一致，替换接口地址即可。

importrequestsimportpymysqlimportjsonimportsysfromdatetimeimportdate,timedelta WEBHOOK_URL="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=your_key"defgenerate_daily_report()->str:conn=pymysql.connect(**DB_CONFIG)cursor=conn.cursor(pymysql.cursors.DictCursor)today=date.today()yesterday=today-timedelta(days=1)# 1. 当日数据概览cursor.execute("SELECT COUNT(*) as total, site FROM competitor_product WHERE crawl_date=%s GROUP BY site",(today,))daily_count=cursor.fetchall()# 2. 价格变动商品cursor.execute(""" SELECT a.title, a.site, b.price as old_price, a.price as new_price, ROUND((a.price - b.price)/b.price*100, 2) as change_rate FROM competitor_product a JOIN competitor_product b ON a.url = b.url AND a.site = b.site WHERE a.crawl_date=%s AND b.crawl_date=%s AND a.price != b.price ORDER BY ABS(change_rate) DESC LIMIT 5 """,(today,yesterday))price_changes=cursor.fetchall()cursor.close()conn.close()# 组装Markdown日报report=f"### 竞品数据日报（{today}）\n\n"report+="**当日采集概览**\n"foritemindaily_count:report+=f"-{item['site']}: 采集{item['total']}条商品数据\n"report+="\n**价格变动Top5**\n"foridx,iteminenumerate(price_changes,1):rate=item['change_rate']tag="📈上涨"ifrate>0else"📉下降"report+=f"{idx}.{item['title'][:20]}...{tag}{abs(rate)}%（¥{item['old_price']}→ ¥{item['new_price']}）\n"report+="\n> 数据由自动化采集生成，详情可查看数据库后台"returnreportdefpush_to_wecom(content:str):payload={"msgtype":"markdown","markdown":{"content":content}}requests.post(WEBHOOK_URL,json=payload,timeout=5)if__name__=="__main__":report=generate_daily_report()push_to_wecom(report)print(json.dumps({"status":"success","length":len(report)},ensure_ascii=False))

3.4 第四步：用TaskFlow编排完整工作流

三个独立的Skill写好后，需要用TaskFlow把它们串成一条完整的流水线，定义好依赖关系和异常处理逻辑。相比单纯的命令拼接，TaskFlow支持状态持久化、断点续跑，中途失败不用从头再来。

在工作区创建daily-collect-flow.js工作流文件：

const{taskflow}=require('@openclaw/sdk');module.exports=taskflow({name:'daily-competitor-collect',description:'每日竞品数据采集全流程：采集→入库→日报推送',steps:[{id:'fetch-data',task:async(ctx)=>{constsites=['siteA','siteB','siteC'];constallData={};for(constsiteofsites){constresult=awaitctx.tools.exec({cmd:`python3 skills/competitor-spider/spider.py${site}3`});allData[site]=JSON.parse(result.stdout);}returnallData;},retry:2},{id:'save-to-db',depends:['fetch-data'],task:async(ctx)=>{constrawData=ctx.state['fetch-data'];constsummary={};for(const[site,data]ofObject.entries(rawData)){constresult=awaitctx.tools.exec({cmd:`python3 skills/data-saver/saver.py${site}'${JSON.stringify(data)}'`});summary[site]=JSON.parse(result.stdout);}returnsummary;}},{id:'push-report',depends:['save-to-db'],task:async(ctx)=>{constresult=awaitctx.tools.exec({cmd:'python3 skills/daily-report/reporter.py'});returnJSON.parse(result.stdout);}}]});

工作流定义了三个步骤，后一步依赖前一步的执行结果，采集步骤配置了2次自动重试，遇到网络波动自动重试，不用人工干预。

3.5 第五步：配置定时任务，实现全自动运行

工作流调试通过后，最后一步就是配置定时任务，让它每天自动跑。OpenClaw内置了Cron调度器，支持标准Cron表达式，不用额外装crontab或定时任务服务。

执行命令添加每日定时任务，设置每天早上9点执行，时区设为上海，独立会话运行：

openclawcronadd\--name"每日竞品数据采集"\--cron"0 9 * * *"\--tz"Asia/Shanghai"\--sessionisolated\--message"执行 daily-competitor-collect 工作流，完成后返回执行结果"\--announce\--channelwecom\--to"你的群机器人标识"

添加完成后可以用openclaw cron list查看所有定时任务，openclaw cron disable <任务ID>可以临时暂停。到点后系统会自动唤醒Agent执行工作流，执行完成后把结果推送到企业微信群，连执行状态都不用手动查。

四、踩坑实录：这些问题我都替你踩过了

这套流程跑了两个多月，遇到过不少细节问题，都是很容易踩的坑，提前避开能省很多事。

坑1：定时任务执行时间差8小时

刚上线的时候发现日报总是下午5点才发，排查了半天发现是默认时区是UTC，和北京时间差8小时。
解决方法：添加定时任务时必须加上--tz "Asia/Shanghai"参数，同时确认服务器时区也是东八区，两边保持一致。

坑2：数据库长连接超时导致入库失败

跑了一周后出现过一次入库失败，原因是数据库连接闲置超时，断开后没有自动重连。
解决方法：不要复用全局数据库连接，每次执行入库操作都新建连接，用完就释放；或者在执行SQL前先做一次连接心跳检测，断开了自动重连。

坑3：企业微信Markdown渲染错乱

日报里的列表、加粗在群里显示异常，有的格式不生效。
解决方法：企业微信只支持子集Markdown语法，不要用复杂的嵌套列表、表格、链接标题，尽量用简单的加粗、无序列表，写完先手动调用测试一次格式。

坑4：采集频率太高触发站点反爬

最开始设置了每页间隔0.5秒，跑了三天就被站点限流了，返回403。
解决方法：把随机延迟调到1-2秒，加上随机UA，单站点每天采集不超过3次；如果需要高频采集，最好搭配代理池使用，不要用固定IP硬冲。

五、写在最后

整套流程搭完跑通其实只花了不到一天时间，但带来的效率提升是持续的。以前每天手动整理数据的时间，现在可以用来分析数据、做决策，这才是自动化真正的价值。

很多人觉得Agent是花架子，干不了实事。但落地之后你会发现，对于这种有固定流程、重复执行的事务性工作，OpenClaw这类执行型Agent的价值非常明确——它不用你从零搭调度框架、写异常处理、做告警通知，只需要把核心业务逻辑封装成技能，剩下的脏活累活全交给框架。

工具是用来解决问题的，能稳定帮你省时间的，就是好工具。