使用Python接受接口回調(diào)信息的完整流程
步驟一
電腦安裝Python，下載鏈接：下載Python |Python.org
步驟二
安裝依賴(lài)環(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("請(qǐng)求頭：", dict(request.headers))
# 解析JSON報(bào)文
if request.is_json:
req_data = request.get_json()
print("回調(diào)JSON報(bào)文：", 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í)行啟動(dòng)命令：
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í)行后回到運(yùn)行服務(wù)的 CMD，即可打印出收到的回調(diào) JSON 數(shù)據(jù)。
補(bǔ)充說(shuō)明
將打印數(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("請(qǐng)求頭：", headers)
if request.is_json:
req_data = request.get_json()
print("回調(diào)JSON報(bào)文：", 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}】報(bào)文：{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)按日期自動(dòng)生成日志文件
from flask import Flask, request
import datetime, os
app = Flask(__name__)
# 日志存放文件夾
LOG_DIR = "agv_logs"
# 文件夾不存在自動(dòng)創(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("請(qǐng)求頭：", 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報(bào)文：", req_data)
# 寫(xiě)入當(dāng)天的日志文件
with open(log_file,"a",encoding="utf-8") as f:
f.write(f"【{time_full}】請(qǐng)求頭：{headers}\n【報(bào)文內(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}】請(qǐng)求頭：{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)知識(shí)擴(kuò)展
在開(kāi)發(fā)中，“接口回調(diào)”通常指 Webhook：一方（回調(diào)提供方）在某個(gè)事件發(fā)生時(shí)，主動(dòng)向一個(gè)預(yù)先配置的 URL（你的服務(wù)器端點(diǎn)）發(fā)送 HTTP 請(qǐng)求，攜帶事件數(shù)據(jù)。你需要做的就是搭建一個(gè) HTTP 服務(wù)器，監(jiān)聽(tīng)這些請(qǐng)求并處理。
下面以 Flask 為例，演示如何安全、穩(wěn)定地接收和處理回調(diào)。
1. 最簡(jiǎn)單的 Flask 回調(diào)接收端
from flask import Flask, request, jsonify
app = Flask(__name__)
@app.route('/webhook', methods=['POST'])
def webhook():
# 獲取請(qǐng)求體數(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)- 運(yùn)行：
python app.py - 測(cè)試：用
curl -X POST -H "Content-Type: application/json" -d '{"event":"test"}' http://localhost:5000/webhook
2. 處理不同類(lèi)型的數(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. 驗(yàn)證簽名（安全性）—— 防止偽造回調(diào)
很多服務(wù)（如 Stripe、GitHub、支付寶）會(huì)在 HTTP 頭中附帶簽名，你需要驗(yàn)證該簽名以確保請(qǐng)求確實(shí)來(lái)自對(duì)方。
示例：Stripe 風(fēng)格簽名驗(yàn)證（HMAC-SHA256）
import hmac
import hashlib
SECRET = "your_webhook_secret" # 從回調(diào)提供方獲取
def verify_signature(payload, signature_header):
# 假設(shè)簽名頭格式：t=timestamp,v1=signature
# 實(shí)際根據(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
# 驗(yàn)證通過(guò)后再處理4. 異步處理任務(wù) —— 防止超時(shí)
回調(diào)通常要求快速響應(yīng)（比如 10 秒內(nèi)），如果處理邏輯耗時(shí)（寫(xiě)數(shù)據(jù)庫(kù)、發(fā)郵件等），應(yīng)該立即返回 200，然后放入后臺(tái)隊(duì)列處理。
使用 Celery 或 Redis Queue，最簡(jiǎn)單的做法是啟動(dòng)一個(gè)線程：
from threading import Thread
def process_webhook(data):
# 耗時(shí)操作
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 來(lái)管理任務(wù)。
5. 斷連與重試機(jī)制
回調(diào)方通常有重試策略（如指數(shù)退避）。你的服務(wù)應(yīng)保證冪等性：即使收到兩次相同的事件，處理結(jié)果也應(yīng)一致（例如通過(guò) event_id 去重）。
processed_ids = set() # 實(shí)際應(yīng)存到 Redis 或數(shù)據(jù)庫(kù)
def handle_event(event_id, data):
if event_id in processed_ids:
return
processed_ids.add(event_id)
# 實(shí)際處理6. 使用 FastAPI 編寫(xiě)異步回調(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)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要給大家介紹了關(guān)于如何利用Python畫(huà)花樣圖的相關(guān)資料,文中通過(guò)示例代碼以及圖文介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2021-10-10
YAML是一種可讀的數(shù)據(jù)序列化格式,常用于配置文件、數(shù)據(jù)交換等場(chǎng)景,本文主要介紹了PyYAML高級(jí)用法,具有一定的參考價(jià)值,感興趣的可以了解一下2025-04-04
這篇文章主要介紹了python函數(shù)裝飾器用法,以實(shí)例形式較為詳細(xì)的分析了Python函數(shù)裝飾器的常見(jiàn)使用技巧,需要的朋友可以參考下2015-06-06
Selenium基于PIL實(shí)現(xiàn)拼接滾動(dòng)截圖
這篇文章主要介紹了Selenium基于PIL實(shí)現(xiàn)拼接滾動(dòng)截圖,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-04-04
今天給大家?guī)?lái)的是用Python語(yǔ)言寫(xiě)一個(gè)關(guān)于手游王者榮耀刷金幣的腳本，有興趣的朋友參考學(xué)習(xí)下。2017-12-12
解決Pycharm后臺(tái)indexing導(dǎo)致不能run的問(wèn)題
今天小編就為大家分享一篇解決Pycharm后臺(tái)indexing導(dǎo)致不能run的問(wèn)題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2019-06-06
今天給大家介紹的是Python列表和字典的相關(guān)知識(shí),文中對(duì)Python字典和列表的性能作了充分的比較,好奇的小伙伴們一起來(lái)看看吧,需要的朋友可以參考下2021-06-06
Python使用Tesseract?OCR實(shí)現(xiàn)識(shí)別圖片中的文字
光學(xué)字符識(shí)別OCR是一項(xiàng)將圖像中的文字轉(zhuǎn)換為可編輯文本的技術(shù),本文將詳細(xì)介紹如何使用Python結(jié)合Tesseract?OCR來(lái)實(shí)現(xiàn)圖片中文字的識(shí)別,感興趣的小伙伴可以了解下2025-11-11
對(duì)python中的乘法dot和對(duì)應(yīng)分量相乘multiply詳解
今天小編就為大家分享一篇對(duì)python中的乘法dot和對(duì)應(yīng)分量相乘multiply詳解，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-11-11
這篇文章主要為大家詳細(xì)介紹了python學(xué)生信息管理系統(tǒng)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2018-03-03