使用Python接受接口回调信息的完整流程

pip install flask

from flask import Flask, request
app = Flask(__name__)

# 完全匹配WCS原回调路径：/wcs/services/rest/cms/agvCallback
@app.route("/wcs/services/rest/cms/agvCallback", methods=["POST","GET"])
def agv_callback():
    print("========== 收到WCS回调数据 ==========")
    print("请求头：", dict(request.headers))
    # 解析JSON报文
    if request.is_json:
        req_data = request.get_json()
        print("回调JSON报文：", req_data)
    else:
        print("表单参数：", request.form)
    # 返回WCS需要的成功应答
    return {"code":0,"msg":"success"}

if __name__ == '__main__':
    # 端口使用23600，和WCS服务同端口
    app.run(host="0.0.0.0", port=23600, debug=False)

python agv_recv.py

curl -X POST "http://127.0.0.1:23600/wcs/services/rest/cms/agvCallback" -H "Content-Type:application/json" -d "{\"agvCode\":\"AGV001\",\"taskNo\":\"T2026060301\"}"

from flask import Flask, request
import datetime
app = Flask(__name__)

@app.route("/wcs/services/rest/cms/agvCallback", methods=["POST","GET"])
def agv_callback():
    print("========== 收到WCS回调数据 ==========")
    headers = dict(request.headers)
    print("请求头：", headers)
    if request.is_json:
        req_data = request.get_json()
        print("回调JSON报文：", req_data)
        # ==========新增：落地文件==========
        with open("agv_data.log","a",encoding="utf-8") as f:
            time_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            f.write(f"【{time_str}】报文：{req_data}\n")
    else:
        print("表单参数：", request.form)
    return {"code":0,"msg":"success"}

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=23600, debug=False)

from flask import Flask, request
import datetime, os
app = Flask(__name__)

# 日志存放文件夹
LOG_DIR = "agv_logs"
# 文件夹不存在自动创建
if not os.path.exists(LOG_DIR):
    os.mkdir(LOG_DIR)

@app.route("/wcs/services/rest/cms/agvCallback", methods=["POST","GET"])
def agv_callback():
    now = datetime.datetime.now()
    time_full = now.strftime("%Y-%m-%d %H:%M:%S")
    date_str = now.strftime("%Y%m%d") # 日期作为日志文件名
    
    print(f"==========【{time_full}】收到WCS回调数据 ==========")
    headers = dict(request.headers)
    print("请求头：", headers)

    # 拼接当日日志完整路径
    log_file = os.path.join(LOG_DIR, f"agv_{date_str}.log")

    if request.is_json:
        req_data = request.get_json()
        print("回调JSON报文：", req_data)
        # 写入当天的日志文件
        with open(log_file,"a",encoding="utf-8") as f:
            f.write(f"【{time_full}】请求头：{headers}\n【报文内容】：{req_data}\n-------------------------\n")
    else:
        form_data = request.form
        print("表单参数：", form_data)
        with open(log_file,"a",encoding="utf-8") as f:
            f.write(f"【{time_full}】请求头：{headers}\n【表单内容】：{form_data}\n-------------------------\n")

    return {"code":0,"msg":"success"}

if __name__ == '__main__':
    app.run(host="0.0.0.0", port=23600, debug=False)

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    # 获取请求体数据（假设为 JSON）
    data = request.get_json()
    if not data:
        return jsonify({'error': 'Invalid data'}), 400

    # 打印或处理数据
    print(f"Received webhook: {data}")

    # 可以在这里触发异步任务，避免阻塞响应
    # process_webhook_async(data)

    # 必须返回 2xx 状态码，否则回调方可能重试
    return jsonify({'status': 'ok'}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

# 兼容多种 Content-Type
if request.is_json:
    data = request.get_json()
else:
    data = request.form.to_dict()   # 或 request.data 原始数据

import hmac
import hashlib

SECRET = "your_webhook_secret"   # 从回调提供方获取

def verify_signature(payload, signature_header):
    # 假设签名头格式：t=timestamp,v1=signature
    # 实际根据文档提取 timestamp 和 signature
    expected = hmac.new(
        SECRET.encode('utf-8'),
        msg=payload,
        digestmod=hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature_header)

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_data()   # 原始字节数据
    sig_header = request.headers.get('X-Signature')
    if not verify_signature(payload, sig_header):
        return '', 401
    # 验证通过后再处理

from threading import Thread

def process_webhook(data):
    # 耗时操作
    pass

@app.route('/webhook', methods=['POST'])
def webhook():
    data = request.get_json()
    Thread(target=process_webhook, args=(data,)).start()
    return '', 200

processed_ids = set()   # 实际应存到 Redis 或数据库
def handle_event(event_id, data):
    if event_id in processed_ids:
        return
    processed_ids.add(event_id)
    # 实际处理

from fastapi import FastAPI, Request, BackgroundTasks

app = FastAPI()

@app.post('/webhook')
async def webhook(request: Request, background_tasks: BackgroundTasks):
    data = await request.json()
    background_tasks.add_task(process_webhook, data)
    return {'status': 'ok'}