使用Python接受接口回調(diào)信息的完整流程

更新時間：2026年06月04日 08:31:02 作者：snow_to

本文主要為大家詳細介紹了搭建Python?Flask服務(wù)接收WCS回調(diào)數(shù)據(jù)的完整流程,文內(nèi)涵蓋從安裝Python到執(zhí)行回調(diào)腳本的全過程,適合初學(xué)者快速上手

步驟一

電腦安裝Python，下載鏈接：下載Python |Python.org

步驟二

安裝依賴環(huán)境

cmd下，執(zhí)行以下命令

pip install flask

出現(xiàn)以下結(jié)果，代表安裝成功

步驟三

創(chuàng)建回調(diào)腳本 agv_recv.py

CMD 輸入：notepad agv_recv.py，彈出記事本后粘貼下面代碼

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

# 完全匹配WCS原回調(diào)路徑：/wcs/services/rest/cms/agvCallback
@app.route("/wcs/services/rest/cms/agvCallback", methods=["POST","GET"])
def agv_callback():
    print("========== 收到WCS回調(diào)數(shù)據(jù) ==========")
    print("請求頭：", dict(request.headers))
    # 解析JSON報文
    if request.is_json:
        req_data = request.get_json()
        print("回調(diào)JSON報文：", req_data)
    else:
        print("表單參數(shù)：", request.form)
    # 返回WCS需要的成功應(yīng)答
    return {"code":0,"msg":"success"}

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

Ctrl+S 保存，關(guān)閉記事本。

步驟四

CMD 直接執(zhí)行啟動命令：

python agv_recv.py

步驟五

執(zhí)行 curl 命令模擬推送消息：

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

執(zhí)行后回到運行服務(wù)的 CMD，即可打印出收到的回調(diào) JSON 數(shù)據(jù)。

補充說明

將打印數(shù)據(jù)保存至文件中

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回調(diào)數(shù)據(jù) ==========")
    headers = dict(request.headers)
    print("請求頭：", headers)
    if request.is_json:
        req_data = request.get_json()
        print("回調(diào)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("表單參數(shù)：", 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"
# 文件夾不存在自動創(chuàng)建
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回調(diào)數(shù)據(jù) ==========")
    headers = dict(request.headers)
    print("請求頭：", headers)

    # 拼接當(dāng)日日志完整路徑
    log_file = os.path.join(LOG_DIR, f"agv_{date_str}.log")

    if request.is_json:
        req_data = request.get_json()
        print("回調(diào)JSON報文：", req_data)
        # 寫入當(dāng)天的日志文件
        with open(log_file,"a",encoding="utf-8") as f:
            f.write(f"【{time_full}】請求頭：{headers}\n【報文內(nèi)容】：{req_data}\n-------------------------\n")
    else:
        form_data = request.form
        print("表單參數(shù)：", form_data)
        with open(log_file,"a",encoding="utf-8") as f:
            f.write(f"【{time_full}】請求頭：{headers}\n【表單內(nèi)容】：{form_data}\n-------------------------\n")

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

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

知識擴展

在開發(fā)中，“接口回調(diào)”通常指 Webhook：一方（回調(diào)提供方）在某個事件發(fā)生時，主動向一個預(yù)先配置的 URL（你的服務(wù)器端點）發(fā)送 HTTP 請求，攜帶事件數(shù)據(jù)。你需要做的就是搭建一個 HTTP 服務(wù)器，監(jiān)聽這些請求并處理。

下面以 Flask 為例，演示如何安全、穩(wěn)定地接收和處理回調(diào)。

1. 最簡單的 Flask 回調(diào)接收端

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    # 獲取請求體數(shù)據(jù)（假設(shè)為 JSON）
    data = request.get_json()
    if not data:
        return jsonify({'error': 'Invalid data'}), 400

    # 打印或處理數(shù)據(jù)
    print(f"Received webhook: {data}")

    # 可以在這里觸發(fā)異步任務(wù)，避免阻塞響應(yīng)
    # process_webhook_async(data)

    # 必須返回 2xx 狀態(tài)碼，否則回調(diào)方可能重試
    return jsonify({'status': 'ok'}), 200

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

運行：python app.py
測試：用 curl -X POST -H "Content-Type: application/json" -d '{"event":"test"}' http://localhost:5000/webhook

2. 處理不同類型的數(shù)據(jù)

回調(diào)可以是 application/json、application/x-www-form-urlencoded 或 multipart/form-data。

# 兼容多種 Content-Type
if request.is_json:
    data = request.get_json()
else:
    data = request.form.to_dict()   # 或 request.data 原始數(shù)據(jù)

3. 驗證簽名（安全性）—— 防止偽造回調(diào)

很多服務(wù)（如 Stripe、GitHub、支付寶）會在 HTTP 頭中附帶簽名，你需要驗證該簽名以確保請求確實來自對方。

示例：Stripe 風(fēng)格簽名驗證（HMAC-SHA256）

import hmac
import hashlib

SECRET = "your_webhook_secret"   # 從回調(diào)提供方獲取

def verify_signature(payload, signature_header):
    # 假設(shè)簽名頭格式：t=timestamp,v1=signature
    # 實際根據(jù)文檔提取 timestamp 和 signature
    expected = hmac.new(
        SECRET.encode('utf-8'),
        msg=payload,
        digestmod=hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature_header)

在 Flask 中：

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_data()   # 原始字節(jié)數(shù)據(jù)
    sig_header = request.headers.get('X-Signature')
    if not verify_signature(payload, sig_header):
        return '', 401
    # 驗證通過后再處理

4. 異步處理任務(wù) —— 防止超時

回調(diào)通常要求快速響應(yīng)（比如 10 秒內(nèi)），如果處理邏輯耗時（寫數(shù)據(jù)庫、發(fā)郵件等），應(yīng)該立即返回 200，然后放入后臺隊列處理。

使用 Celery 或 Redis Queue，最簡單的做法是啟動一個線程：

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

注意：生產(chǎn)環(huán)境建議使用 Celery + Redis/RabbitMQ 來管理任務(wù)。

5. 斷連與重試機制

回調(diào)方通常有重試策略（如指數(shù)退避）。你的服務(wù)應(yīng)保證冪等性：即使收到兩次相同的事件，處理結(jié)果也應(yīng)一致（例如通過 event_id 去重）。

processed_ids = set()   # 實際應(yīng)存到 Redis 或數(shù)據(jù)庫
def handle_event(event_id, data):
    if event_id in processed_ids:
        return
    processed_ids.add(event_id)
    # 實際處理

6. 使用 FastAPI 編寫異步回調(diào)接收

FastAPI 天然支持異步，性能更好：

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'}

到此這篇關(guān)于使用Python接受接口回調(diào)信息的完整流程的文章就介紹到這了,更多相關(guān)Python接受接口回調(diào)信息內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: