影刀RPA Webhook接收后解析JSON并触发流程作者林焱前面讲了如何向外推送消息这篇反过来——让影刀接收外部系统发来的Webhook自动触发流程。比如GitHub仓库有新issue时自动通知你、客户下单后自动录入Excel、监控系统告警后自动处理。Webhook接收原理Webhook本质是一个HTTP POST请求。外部系统在特定事件发生时比如新订单向你的URL发一个POST请求请求体里带着事件数据通常是JSON。问题来了影刀的本地流程怎么接收HTTP请求影刀本身没有内置HTTP服务器。方案一Python Flask搭建简易Web服务在Python代码块里启动一个Flask服务器fromflaskimportFlask,request,jsonify appFlask(__name__)app.route(/webhook,methods[POST])defhandle_webhook():datarequest.json# 拿到外部系统发来的JSON数据print(f收到Webhook:{data})[video(video-G1S0ulzl-1783721822785)(type-csdn)(url-https://live.csdn.net/v/embed/526818)(image-https://v-blog.csdnimg.cn/asset/582d14c3bd0451c5399cd990b56e2a0d/cover/Cover0.jpg)(title-拼多多店群自动化报活动上架)]# 处理数据写入Excel、发通知等# ...returnjsonify({status:ok}),200# 启动服务监听5000端口app.run(host0.0.0.0,port5000)但这个方案有个问题app.run()是阻塞的一旦启动后续影刀流程代码不会执行。如果你放在单独的线程里又涉及到线程管理。方案二推荐用第三方中转服务更实用的做法是用免费的Webhook中转服务。工作流程外部系统 → Webhook中转服务 → 定期查询或转发 → 影刀处理1. 使用企业微信/钉钉机器人作为中转外部系统 → 企微群机器人发消息→ 影刀定期读取群消息 → 触发处理2. 使用飞书的Webhook Action飞书支持把Webhook转成消息推送影刀定时读取。3. 创建一个简单的中转方案# 这个Python脚本独立运行接收Webhook存到SQLite# 影刀定期读取SQLite发现有新数据就处理importsqlite3fromflaskimportFlask,requestimportjsonimportdatetime appFlask(__name__)definit_db():connsqlite3.connect(webhooks.db)conn.execute(CREATE TABLE IF NOT EXISTS events (id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, data TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, processed INTEGER DEFAULT 0))conn.commit()conn.close()app.route(/webhook/source,methods[POST])defreceive(source):connsqlite3.connect(webhooks.db)conn.execute(INSERT INTO events (source, data) VALUES (?, ?),(source,json.dumps(request.json)))conn.commit()conn.close()returnOK,200init_db()app.run(host0.0.0.0,port5000)然后在影刀里每5分钟读取SQLite中未处理的事件importsqlite3 connsqlite3.connect(webhooks.db)cursorconn.execute(SELECT id, source, data FROM events WHERE processed 0 LIMIT 50)eventscursor.fetchall()foreventinevents:event_id,source,data_jsonevent datajson.loads(data_json)# 根据source执行不同的处理逻辑ifsourcegithub:print(f新的GitHub事件:{data})# 影刀写入Excel、发通知等elifsourceorder:print(f新订单:{data})# 标记为已处理conn.execute(UPDATE events SET processed 1 WHERE id ?,(event_id,))conn.commit()conn.close()踩坑实录坑1内网穿透你的影刀跑在本机外面的系统访问不到。要么把Flask服务部署到有公网IP的服务器上要么用ngrok/frp做内网穿透适合测试。坑2Flask服务进程管理启动Flask服务后如果不小心关掉窗口服务就停了。建议用nssm或Windows服务把Flask注册为系统服务。坑3并发处理TEMU店群矩阵自动化运营核价报活动Flask默认是单线程的同时来多个Webhook请求会排队。设置threadedTrueapp.run(host0.0.0.0,port5000,threadedTrue)坑4数据安全性Webhook URL是个公开入口虽然URL可以设计得很长很难猜任何人都可以往你的URL发POST请求。在影刀处理数据前验证请求来源比如校验签名字段。简化方案定时轮询替代Webhook如果外部系统不支持Webhook但有API可以查询最近变更定时轮询也是一种方案# 每5分钟查询是否有新数据importrequestsimporthashlib last_check# 上次检查时最新记录的ID或时间戳resprequests.get(https://api.example.com/recent?sincelast_check)new_itemsresp.json()foriteminnew_items:# 处理新数据# 更新last_checkifnew_items:last_checknew_items[0][id]写在最后Webhook接收是一种进阶能力让外部系统能主动触发你的影刀流程。对于大多数场景定时轮询API已经够用。但如果你的业务需要秒级响应比如监控告警、订单提醒搭建一个Webhook接收服务是必要的投资。
