webhook.py
2.62 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
import hmac
import hashlib
import base64
from flask import Blueprint, request, jsonify, current_app
from config import Config
from middleware.hmac_validator import verify_webhook_hmac
# 创建Webhook蓝图
webhook_bp = Blueprint('webhook', __name__, url_prefix='/webhook')
@webhook_bp.route('/shoplazza', methods=['POST'])
def shoplazza_webhook():
"""
处理Shoplazza Webhook
"""
# 获取HMAC签名头
hmac_header = request.headers.get('X-Shoplazza-Hmac-Sha256')
if not hmac_header:
return jsonify({'error': 'Missing HMAC header'}), 400
# 获取原始数据
data = request.get_data()
# 验证Webhook签名
if not verify_webhook_hmac(data, hmac_header):
current_app.logger.warning("Webhook HMAC verification failed")
return jsonify({'error': 'Invalid webhook signature'}), 403
try:
# 解析Webhook数据
webhook_data = request.get_json()
current_app.logger.info(f"Received webhook: {webhook_data}")
# 根据Webhook类型处理
webhook_type = webhook_data.get('type')
if webhook_type == 'app/uninstalled':
handle_app_uninstalled(webhook_data)
elif webhook_type == 'orders/create':
handle_order_created(webhook_data)
elif webhook_type == 'orders/updated':
handle_order_updated(webhook_data)
elif webhook_type == 'products/create':
handle_product_created(webhook_data)
else:
current_app.logger.info(f"Unhandled webhook type: {webhook_type}")
return jsonify({'status': 'success'})
except Exception as e:
current_app.logger.error(f"Webhook处理失败: {str(e)}")
return jsonify({'error': 'Webhook processing failed'}), 500
def handle_app_uninstalled(webhook_data):
"""处理应用卸载事件"""
shop = webhook_data.get('shop')
if shop and shop in Config.ACCESS_TOKENS:
del Config.ACCESS_TOKENS[shop]
current_app.logger.info(f"App uninstalled for shop: {shop}")
def handle_order_created(webhook_data):
"""处理订单创建事件"""
order = webhook_data.get('order')
if order:
current_app.logger.info(f"New order created: {order.get('id')}")
def handle_order_updated(webhook_data):
"""处理订单更新事件"""
order = webhook_data.get('order')
if order:
current_app.logger.info(f"Order updated: {order.get('id')}")
def handle_product_created(webhook_data):
"""处理产品创建事件"""
product = webhook_data.get('product')
if product:
current_app.logger.info(f"New product created: {product.get('id')}")