70 lines
2.0 KiB
Python
70 lines
2.0 KiB
Python
from fastapi import FastAPI, Request, HTTPException, Query, APIRouter
|
|
from fastapi.responses import JSONResponse, PlainTextResponse
|
|
import hashlib
|
|
import xml.etree.ElementTree as ET
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
APP_ID = "wx33726a2d409faa94"
|
|
APP_SECRET = "59b7919d45d4631ab55bf418abb3e8bf"
|
|
|
|
# TODO: 接收消息推送使用安全模式
|
|
TOKEN = "sleepwithoutbz"
|
|
ENCODING_AES_KEY = "njMce1oYKW7V11yug9Qjm689DYzXQceFedaeRDHSOsL"
|
|
|
|
|
|
@router.get("/callback", response_class=PlainTextResponse)
|
|
async def wechat_callback(
|
|
signature: str = Query(..., description="微信加密签名"),
|
|
timestamp: str = Query(..., description="时间戳"),
|
|
nonce: str = Query(..., description="随机数"),
|
|
echostr: str = Query(..., description="验证字符串"),
|
|
):
|
|
"""
|
|
微信验证回调接口
|
|
1. 将token/timestamp/nonce按字典序排序
|
|
2. 拼接后sha1加密
|
|
3. 与signature比对验证
|
|
"""
|
|
# 1. 参数排序
|
|
params = sorted([TOKEN, timestamp, nonce])
|
|
|
|
# 2. 拼接字符串并加密
|
|
sign_str = "".join(params).encode("utf-8")
|
|
calculated_signature = hashlib.sha1(sign_str).hexdigest()
|
|
|
|
# 3. 验证签名
|
|
if calculated_signature == signature:
|
|
# 验证成功返回原始echostr
|
|
return echostr
|
|
else:
|
|
# 验证失败返回错误
|
|
raise HTTPException(status_code=403, detail="Invalid signature")
|
|
|
|
|
|
@router.post("/callback")
|
|
async def wechat_callback_post(request: Request):
|
|
"""
|
|
微信消息推送接口
|
|
"""
|
|
# 1. 获取请求数据
|
|
xml_data = await request.body()
|
|
|
|
# 2. 解析xml数据
|
|
root = ET.fromstring(xml_data)
|
|
|
|
# 3. 获取消息类型
|
|
msg_type = root.find("MsgType").text
|
|
from_user_name = root.find("FromUserName").text
|
|
|
|
# 4. 处理消息
|
|
if msg_type == "text":
|
|
# 4.1. 处理文本消息
|
|
content = root.find("Content").text
|
|
return content
|
|
# return JSONResponse(content={"Content": content})
|
|
else:
|
|
# 4.2. 处理其他类型消息
|
|
return PlainTextResponse("Some content")
|