111 lines
3.6 KiB
Python
111 lines
3.6 KiB
Python
|
"""
|
|||
|
|
微信公众号 FastAPI 应用
|
|||
|
|
|
|||
|
|
GET /wechat - 微信服务器验证（签名校验 + 返回 echostr）
|
|||
|
|
POST /wechat - 接收微信推送的消息和事件
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import hashlib
|
|||
|
|
import time
|
|||
|
|
import xml.etree.ElementTree as ET
|
|||
|
|
|
|||
|
|
from fastapi import FastAPI, Request, Query, HTTPException
|
|||
|
|
from fastapi.responses import PlainTextResponse
|
|||
|
|
|
|||
|
|
from config import WECHAT_TOKEN
|
|||
|
|
|
|||
|
|
app = FastAPI(title="WeChat API", version="0.1.0")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def verify_signature(signature: str, timestamp: str, nonce: str) -> bool:
|
|||
|
|
"""
|
|||
|
|
微信签名验证:
|
|||
|
|
1. 将 token、timestamp、nonce 三个参数字典序排序
|
|||
|
|
2. 拼成一个字符串
|
|||
|
|
3. sha1 加密
|
|||
|
|
4. 和 signature 对比
|
|||
|
|
"""
|
|||
|
|
if not WECHAT_TOKEN:
|
|||
|
|
return False
|
|||
|
|
items = sorted([WECHAT_TOKEN, timestamp, nonce])
|
|||
|
|
sha1 = hashlib.sha1("".join(items).encode()).hexdigest()
|
|||
|
|
return sha1 == signature
|
|||
|
|
|
|||
|
|
|
|||
|
|
@app.get("/wechat")
|
|||
|
|
async def wechat_verify(
|
|||
|
|
signature: str = Query(...),
|
|||
|
|
timestamp: str = Query(...),
|
|||
|
|
nonce: str = Query(...),
|
|||
|
|
echostr: str = Query(...),
|
|||
|
|
):
|
|||
|
|
"""GET: 微信服务器验证回调地址有效性"""
|
|||
|
|
if not verify_signature(signature, timestamp, nonce):
|
|||
|
|
raise HTTPException(status_code=403, detail="Invalid signature")
|
|||
|
|
# 验证通过，原样返回 echostr
|
|||
|
|
return PlainTextResponse(content=echostr)
|
|||
|
|
|
|||
|
|
|
|||
|
|
@app.post("/wechat")
|
|||
|
|
async def wechat_message(
|
|||
|
|
request: Request,
|
|||
|
|
signature: str = Query(...),
|
|||
|
|
timestamp: str = Query(...),
|
|||
|
|
nonce: str = Query(...),
|
|||
|
|
openid: str = Query(...),
|
|||
|
|
):
|
|||
|
|
"""POST: 接收微信推送的消息和事件"""
|
|||
|
|
# 验证签名
|
|||
|
|
if not verify_signature(signature, timestamp, nonce):
|
|||
|
|
raise HTTPException(status_code=403, detail="Invalid signature")
|
|||
|
|
|
|||
|
|
# 解析 XML 消息体
|
|||
|
|
body = await request.body()
|
|||
|
|
root = ET.fromstring(body)
|
|||
|
|
|
|||
|
|
msg_type = root.findtext("MsgType", "")
|
|||
|
|
from_user = root.findtext("FromUserName", "") # 发送方（用户 openid）
|
|||
|
|
to_user = root.findtext("ToUserName", "") # 接收方（公众号）
|
|||
|
|
content = root.findtext("Content", "")
|
|||
|
|
event = root.findtext("Event", "")
|
|||
|
|
event_key = root.findtext("EventKey", "")
|
|||
|
|
|
|||
|
|
print(f"[WeChat] type={msg_type} from={from_user} event={event} content={content} key={event_key}")
|
|||
|
|
|
|||
|
|
# 处理事件推送
|
|||
|
|
if msg_type == "event":
|
|||
|
|
if event == "subscribe":
|
|||
|
|
# 用户关注
|
|||
|
|
return _reply_text(from_user, to_user, "欢迎关注！")
|
|||
|
|
elif event == "unsubscribe":
|
|||
|
|
# 用户取关
|
|||
|
|
print(f"[WeChat] 用户取关: {from_user}")
|
|||
|
|
return PlainTextResponse(content="success")
|
|||
|
|
elif event == "SCAN":
|
|||
|
|
# 已关注用户扫码
|
|||
|
|
return _reply_text(from_user, to_user, f"扫码成功，场景值: {event_key}")
|
|||
|
|
elif event == "CLICK":
|
|||
|
|
# 菜单点击
|
|||
|
|
return _reply_text(from_user, to_user, f"点击了: {event_key}")
|
|||
|
|
|
|||
|
|
# 处理文本消息
|
|||
|
|
if msg_type == "text":
|
|||
|
|
# 原样返回（echo 模式，方便测试）
|
|||
|
|
return _reply_text(from_user, to_user, f"你说: {content}")
|
|||
|
|
|
|||
|
|
# 其他类型暂不处理
|
|||
|
|
return _reply_text(from_user, to_user, "收到")
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _reply_text(from_user: str, to_user: str, content: str) -> PlainTextResponse:
|
|||
|
|
"""构造文本回复 XML"""
|
|||
|
|
xml = (
|
|||
|
|
"<xml>"
|
|||
|
|
f"<ToUserName><![CDATA[{from_user}]]></ToUserName>"
|
|||
|
|
f"<FromUserName><![CDATA[{to_user}]]></FromUserName>"
|
|||
|
|
f"<CreateTime>{int(time.time())}</CreateTime>"
|
|||
|
|
"<MsgType><![CDATA[text]]></MsgType>"
|
|||
|
|
f"<Content><![CDATA[{content}]]></Content>"
|
|||
|
|
"</xml>"
|
|||
|
|
)
|
|||
|
|
return PlainTextResponse(content=xml, media_type="application/xml")
|