wechat-scan/wechat.py

"""
微信公众号 FastAPI 应用

GET  /wechat  - 微信服务器验证（签名校验 + 返回 echostr）
POST /wechat  - 接收微信推送的消息和事件（subscribe / SCAN 触发扫码授权）

授权接口在 auth.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
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.2.0", lifespan=lifespan)
app.include_router(auth.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, "欢迎关注！")

    authorized = await auth.handle_scan(scene_str, from_user)
    if authorized:
        return _reply_text(from_user, to_user, "授权成功，请返回电脑端继续操作。")
    return _reply_text(from_user, to_user, "二维码已失效或已被使用，请在电脑端刷新后重新扫码。")


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")