影刀RPA Webhook接收后解析JSON并触发流程

作者：林焱

前面讲了如何向外推送消息，这篇反过来——让影刀接收外部系统发来的Webhook，自动触发流程。比如：GitHub仓库有新issue时自动通知你、客户下单后自动录入Excel、监控系统告警后自动处理。

Webhook接收原理

Webhook本质是一个HTTP POST请求。外部系统在特定事件发生时（比如新订单），向你的URL发一个POST请求，请求体里带着事件数据（通常是JSON）。

问题来了：影刀的本地流程怎么接收HTTP请求？影刀本身没有内置HTTP服务器。

方案一：Python Flask搭建简易Web服务

在Python代码块里启动一个Flask服务器：

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    data = request.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、发通知等
    # ...
    
    return jsonify({"status": "ok"}), 200

# 启动服务（监听5000端口）
app.run(host='0.0.0.0', port=5000)

但这个方案有个问题：app.run()是阻塞的，一旦启动，后续影刀流程代码不会执行。如果你放在单独的线程里，又涉及到线程管理。

方案二（推荐）：用第三方中转服务

更实用的做法是用免费的Webhook中转服务。工作流程：

外部系统 → Webhook中转服务 → 定期查询或转发 → 影刀处理

1. 使用企业微信/钉钉机器人作为中转

外部系统 → 企微群机器人（发消息）→ 影刀定期读取群消息 → 触发处理

2. 使用飞书的Webhook Action

飞书支持把Webhook转成消息推送，影刀定时读取。

3. 创建一个简单的中转方案

# 这个Python脚本独立运行，接收Webhook存到SQLite
# 影刀定期读取SQLite，发现有新数据就处理

import sqlite3
from flask import Flask, request
import json
import datetime

app = Flask(__name__)

def init_db():
    conn = sqlite3.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'])
def receive(source):
    conn = sqlite3.connect('webhooks.db')
    conn.execute(
        'INSERT INTO events (source, data) VALUES (?, ?)',
        (source, json.dumps(request.json))
    )
    conn.commit()
    conn.close()
    return 'OK', 200

init_db()
app.run(host='0.0.0.0', port=5000)

然后在影刀里，每5分钟读取SQLite中未处理的事件：

import sqlite3

conn = sqlite3.connect('webhooks.db')
cursor = conn.execute(
    "SELECT id, source, data FROM events WHERE processed = 0 LIMIT 50"
)
events = cursor.fetchall()

for event in events:
    event_id, source, data_json = event
    data = json.loads(data_json)
    
    # 根据source执行不同的处理逻辑
    if source == 'github':
        print(f"新的GitHub事件: {data}")
        # 影刀：写入Excel、发通知等
    elif source == 'order':
        print(f"新订单: {data}")
    
    # 标记为已处理
    conn.execute("UPDATE events SET processed = 1 WHERE id = ?", (event_id,))

conn.commit()
conn.close()

踩坑实录

坑1：内网穿透

你的影刀跑在本机，外面的系统访问不到。要么把Flask服务部署到有公网IP的服务器上，要么用ngrok/frp做内网穿透（适合测试）。

坑2：Flask服务进程管理

启动Flask服务后，如果不小心关掉窗口，服务就停了。建议用nssm或Windows服务把Flask注册为系统服务。

坑3：并发处理

Flask默认是单线程的，同时来多个Webhook请求会排队。设置threaded=True：

app.run(host='0.0.0.0', port=5000, threaded=True)

坑4：数据安全性

Webhook URL是个公开入口（虽然URL可以设计得很长很难猜），任何人都可以往你的URL发POST请求。在影刀处理数据前，验证请求来源（比如校验签名字段）。

简化方案：定时轮询替代Webhook

如果外部系统不支持Webhook，但有API可以查询"最近变更"，定时轮询也是一种方案：

# 每5分钟查询是否有新数据
import requests
import hashlib

last_check = ""  # 上次检查时最新记录的ID或时间戳

resp = requests.get("https://api.example.com/recent?since=" + last_check)
new_items = resp.json()

for item in new_items:
    # 处理新数据
    
# 更新last_check
if new_items:
    last_check = new_items[0]["id"]

写在最后

Webhook接收是一种进阶能力，让外部系统能主动触发你的影刀流程。对于大多数场景，定时轮询API已经够用。但如果你的业务需要"秒级响应"（比如监控告警、订单提醒），搭建一个Webhook接收服务是必要的投资。