Files
wechat-scan/wechat.py
T
gjm cfb2a36c87 feat: 阶段 2 - /usage/consume、pending 授权惰性激活
- usage.py: POST /usage/consume,事务内校验会话(严格 device_id)、
  激活 pending、按 time/points 分支扣减;失败返回 ok:false + reason
- 积分扣减用条件 UPDATE 原子完成,status 赋值写在自减之前
  (MySQL SET 从左到右求值,否则 IF 读到已减 1 的值,判空差 1)
- 时间授权 usage_logs 按 (user_id, device_id) 60 秒节流;积分每次必写
- auth.py: 新增 activate_pending_authorization(),失效 active 后
  FIFO 激活 pending,时间授权按原时长从当前时刻重新锚定
- get_status/handle_scan 接入惰性激活;get_status 改事务包裹
- _get_active_authorization/_serialize_authorization 改公开供 usage 复用
- config/.env.example: 新增 USAGE_LOG_THROTTLE_SECONDS
2026-09-27 15:00:02 +08:00

167 lines
5.5 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
微信公众号 FastAPI 应用
GET /wechat - 微信服务器验证(签名校验 + 返回 echostr)
POST /wechat - 接收微信推送的消息和事件(subscribe / SCAN 触发扫码授权)
授权接口在 auth.py、使用扣减接口在 usage.py 中定义,通过 include_router 挂载。
"""
import hashlib
import logging
import time
import xml.etree.ElementTree as ET
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.responses import PlainTextResponse
import auth
import db
import usage
from config import WECHAT_TOKEN
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)
logger = logging.getLogger(__name__)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用启动时建立 MySQL 连接池,关闭时释放"""
await db.init_pool()
logger.info("MySQL 连接池已初始化")
yield
await db.close_pool()
logger.info("MySQL 连接池已关闭")
app = FastAPI(title="WeChat API", version="0.3.0", lifespan=lifespan)
app.include_router(auth.router)
app.include_router(usage.router)
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(...),
):
"""POST: 接收微信推送的消息和事件"""
if not verify_signature(signature, timestamp, nonce):
raise HTTPException(status_code=403, detail="Invalid signature")
body = await request.body()
if not body:
return PlainTextResponse(content="success")
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", "")
logger.info(
"收到微信推送 type=%s event=%s from=%s key=%s", msg_type, event, from_user, event_key
)
if msg_type == "event":
if event == "subscribe":
# 未关注用户扫码关注:EventKey 形如 qrscene_<scene_str>
return await _handle_scan_event(from_user, to_user, event_key, is_subscribe=True)
if event == "SCAN":
# 已关注用户扫码:EventKey 直接是 <scene_str>
return await _handle_scan_event(from_user, to_user, event_key, is_subscribe=False)
if event == "unsubscribe":
logger.info("用户取关 openid=%s", from_user)
return PlainTextResponse(content="success")
if 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 PlainTextResponse(content="success")
async def _handle_scan_event(
from_user: str, to_user: str, event_key: str, is_subscribe: bool
) -> PlainTextResponse:
"""解析场景值并完成扫码授权,回复用户处理结果"""
scene_str = _parse_scene_key(event_key, is_subscribe)
if not scene_str:
# 无场景值的普通关注
return _reply_text(from_user, to_user, "欢迎关注!")
result = await auth.handle_scan(scene_str, from_user)
if result == "authorized":
reply = "授权成功,请返回电脑端继续操作。"
elif result == "need_purchase":
reply = "当前无可用授权,请返回电脑端按提示操作。"
else:
reply = "二维码已失效或已被使用,请在电脑端刷新后重新扫码。"
return _reply_text(from_user, to_user, reply)
def _parse_scene_key(event_key: str, is_subscribe: bool) -> str:
"""
从 EventKey 中提取 scene_str。
subscribe 事件带 qrscene_ 前缀,SCAN 事件不带;无场景值时返回空串。
"""
if not event_key:
return ""
if is_subscribe:
prefix = "qrscene_"
return event_key[len(prefix):] if event_key.startswith(prefix) else ""
return event_key
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")