做产品和运营的朋友应该都有这种体会每天上班第一件事就是打开七八个竞品站点挨个扒价格、上新、销量数据复制粘贴进表格再汇总成日报发到群里。整套流程下来四五十分钟就没了全是机械重复的劳动还经常因为页面加载慢、数据多抄错数。之前也试过写Python脚本跑采集但问题也很明显站点一改页面结构脚本就崩要自己维护定时任务、异常重试、告警通知折腾半天最后还是得每天盯着执行结果反而多了运维成本。直到用OpenClaw搭了这套全自动采集链路从定时抓取、数据清洗入库到日报推送全流程无人值守跑了两个多月只出过两次小问题每天实打实省出一个多小时干正事。今天把完整的落地步骤和踩过的坑整理出来照着做你也能搭出自己的自动化采集流水线。一、整体方案架构设计整套方案的核心思路是让OpenClaw承担调度、执行、异常处理的全部脏活我们只需要封装最核心的业务逻辑不用操心调度框架、重试机制、告警通知这些基础设施。整体分为四层各司其职存储输出层能力层Agent执行层调度层Cron定时触发器TaskFlow工作流编排OpenClaw 执行引擎竞品采集Skill数据清洗入库Skill日报生成推送SkillMySQL业务数据库企业微信/钉钉群执行链路很清晰定时任务到点触发工作流 → 先执行采集拉取原始数据 → 清洗去重后写入数据库 → 取出当日数据生成对比日报 → 推送到业务群。全程自带异常重试和状态追踪某一步失败会自动重试重试不成就发告警通知人工介入。二、前期准备工作正式动手之前先把环境和依赖备齐避免中途卡壳一套可用的OpenClaw运行环境本地Windows或Linux服务器都可以推荐放服务器上7x24小时运行目标竞品站点的字段梳理确定要采集哪些核心指标商品名、价格、销量、上架时间等MySQL数据库5.7及以上版本用来存历史数据做趋势对比企业微信/钉钉群机器人Webhook用来推送日报和告警Python基础依赖requests、pymysql、pandas用来写Skill的业务逻辑三、分步落地从零搭建完整自动化链路3.1 第一步封装竞品数据采集SkillOpenClaw的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:int3)-list:base_urlSITE_CONFIG[site][base_url]result[]forpageinrange(1,max_page1):headers{User-Agent:random.choice(UA_LIST)}resprequests.get(f{base_url}?page{page},headersheaders,timeout10)soupBeautifulSoup(resp.text,html.parser)itemssoup.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 sitesys.argv[1]iflen(sys.argv)1elsesiteApageint(sys.argv[2])iflen(sys.argv)2else3datafetch_site_data(site,page)print(json.dumps(data,ensure_asciiFalse))写完后执行openclaw skills reload加载技能就可以在对话里直接调用了。3.2 第二步封装数据清洗与入库Skill采集到的原始数据不能直接用需要做去重、格式校验、增量判断再写入数据库。我们再封装一个入库Skill。先在MySQL建表存每日采集快照CREATETABLEcompetitor_product(idintunsignedNOTNULLAUTO_INCREMENT,sitevarchar(32)NOTNULLCOMMENT站点标识,titlevarchar(255)NOTNULLCOMMENT商品名称,pricedecimal(10,2)NOTNULLCOMMENT当前价格,salesintNOTNULLCOMMENT累计销量,urlvarchar(512)NOTNULLCOMMENT商品链接,crawl_datedateNOTNULLCOMMENT采集日期,crawl_timedatetimeNOTNULLCOMMENT采集时间,PRIMARYKEY(id),UNIQUEKEYuk_site_url_date(site,url(255),crawl_date))ENGINEInnoDBDEFAULTCHARSETutf8mb4;然后写入库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:connpymysql.connect(**DB_CONFIG)cursorconn.cursor()todaydate.today()insert_count0skip_count0try:foritemindata:# 增量判断当日同链接数据已存在则跳过check_sqlSELECT id FROM competitor_product WHERE site%s AND url%s AND crawl_date%scursor.execute(check_sql,(site,item[url],today))ifcursor.fetchone():skip_count1continueinsert_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_count1conn.commit()finally:cursor.close()conn.close()return{insert:insert_count,skip:skip_count,total:len(data)}if__name____main__:sitesys.argv[1]raw_datajson.loads(sys.argv[2])resultsave_to_db(site,raw_data)print(json.dumps(result,ensure_asciiFalse))3.3 第三步封装日报生成与推送Skill数据落库后需要自动生成对比日报推送到业务群。日报不用太复杂核心信息到位就行当日新增商品数、价格变动Top5、销量涨幅Top5再附一个数据概览。推送用企业微信群机器人的Webhook最省事钉钉逻辑完全一致替换接口地址即可。importrequestsimportpymysqlimportjsonimportsysfromdatetimeimportdate,timedelta WEBHOOK_URLhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_keydefgenerate_daily_report()-str:connpymysql.connect(**DB_CONFIG)cursorconn.cursor(pymysql.cursors.DictCursor)todaydate.today()yesterdaytoday-timedelta(days1)# 1. 当日数据概览cursor.execute(SELECT COUNT(*) as total, site FROM competitor_product WHERE crawl_date%s GROUP BY site,(today,))daily_countcursor.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_changescursor.fetchall()cursor.close()conn.close()# 组装Markdown日报reportf### 竞品数据日报{today}\n\nreport**当日采集概览**\nforitemindaily_count:reportf-{item[site]}: 采集{item[total]}条商品数据\nreport\n**价格变动Top5**\nforidx,iteminenumerate(price_changes,1):rateitem[change_rate]tag上涨ifrate0else下降reportf{idx}.{item[title][:20]}...{tag}{abs(rate)}%¥{item[old_price]}→ ¥{item[new_price]}\nreport\n 数据由自动化采集生成详情可查看数据库后台returnreportdefpush_to_wecom(content:str):payload{msgtype:markdown,markdown:{content:content}}requests.post(WEBHOOK_URL,jsonpayload,timeout5)if__name____main__:reportgenerate_daily_report()push_to_wecom(report)print(json.dumps({status:success,length:len(report)},ensure_asciiFalse))3.4 第四步用TaskFlow编排完整工作流三个独立的Skill写好后需要用TaskFlow把它们串成一条完整的流水线定义好依赖关系和异常处理逻辑。相比单纯的命令拼接TaskFlow支持状态持久化、断点续跑中途失败不用从头再来。在工作区创建daily-collect-flow.js工作流文件const{taskflow}require(openclaw/sdk);module.exportstaskflow({name:daily-competitor-collect,description:每日竞品数据采集全流程采集→入库→日报推送,steps:[{id:fetch-data,task:async(ctx){constsites[siteA,siteB,siteC];constallData{};for(constsiteofsites){constresultawaitctx.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){constrawDatactx.state[fetch-data];constsummary{};for(const[site,data]ofObject.entries(rawData)){constresultawaitctx.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){constresultawaitctx.tools.exec({cmd:python3 skills/daily-report/reporter.py});returnJSON.parse(result.stdout);}}]});工作流定义了三个步骤后一步依赖前一步的执行结果采集步骤配置了2次自动重试遇到网络波动自动重试不用人工干预。3.5 第五步配置定时任务实现全自动运行工作流调试通过后最后一步就是配置定时任务让它每天自动跑。OpenClaw内置了Cron调度器支持标准Cron表达式不用额外装crontab或定时任务服务。执行命令添加每日定时任务设置每天早上9点执行时区设为上海独立会话运行openclawcronadd\--name每日竞品数据采集\--cron0 9 * * *\--tzAsia/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的价值非常明确——它不用你从零搭调度框架、写异常处理、做告警通知只需要把核心业务逻辑封装成技能剩下的脏活累活全交给框架。工具是用来解决问题的能稳定帮你省时间的就是好工具。