diff --git a/app/api/routers/auth.py b/app/api/routers/auth.py index fd2fcdf..1bc98a5 100644 --- a/app/api/routers/auth.py +++ b/app/api/routers/auth.py @@ -7,6 +7,7 @@ """ from __future__ import annotations +import base64 import json import re import secrets @@ -670,17 +671,30 @@ async def wx_qr_poll(scene: str): @router.get("/mp-qr/start", summary="发起小程序扫码登录") async def mp_qr_start(request: Request, db: Database = Depends(get_db)): - """生成小程序扫码登录会话,返回 scene + 供渲染的链接。""" - if not config.AUTH_WECHAT_LOGIN: - raise HTTPException(status_code=503, detail="小程序登录未开启") + """生成小程序扫码登录会话,返回 scene + 小程序码图片。 + + 用于「小程序扫码登录其它端口」:桌面/admin/网页登录页展示该码, + 用户用 **微信扫一扫** → 自动打开小程序「扫码登录」页(复用其登录身份)确认, + 目标端口轮询取得登录令牌。scene 用短 token(wxacode 限 32 字符)。 + + 未配置 WECHAT_APPID/SECRET 时退化为普通二维码(qr_url),仅支持「小程序内 + scanCode 路径」,无法被微信自动拉起小程序。 + """ + if not config.AUTH_WECHAT_QR: + raise HTTPException(status_code=503, detail="小程序扫码登录未开启") if not config.AUTH_ENABLED: raise HTTPException(status_code=403, detail="认证未开启") - scene = wechat.new_scene() + scene = f"mp_{secrets.token_hex(10)}" # 3+20 = 23 字符 wx_qr_store.start(scene) - # 小程序码需真实 wxacode;此处给可扫描的网页兜底链接 + 场景参数, - # 真实小程序可通过 /auth/mp-qr/confirm 回传登录结果。前端据 qr_url 渲染二维码。 - qr_url = f"https://opc.pinesound.cn/mp/login?scene={scene}" - return {"scene": scene, "qr_url": qr_url, "mp_enabled": True} + # 生成小程序码(微信扫码自动打开小程序 pages/scan-login 并携带 scene) + try: + png = await wechat.get_wxacode(scene, page="pages/scan-login/index") + b64 = base64.b64encode(png).decode("ascii") + return {"scene": scene, "qr_image": f"data:image/png;base64,{b64}", "mp_enabled": True} + except wechat.WechatError: + # 无凭据/失败兜底:返回可扫描的 URL,前端按 qr_url 渲染普通二维码 + qr_url = f"https://opc.pinesound.cn/mp/login?scene={scene}" + return {"scene": scene, "qr_url": qr_url, "mp_enabled": False} @router.get("/mp-qr/poll", summary="轮询小程序扫码登录状态") diff --git a/app/services/wechat.py b/app/services/wechat.py index f931bf9..4302cad 100644 --- a/app/services/wechat.py +++ b/app/services/wechat.py @@ -110,6 +110,69 @@ async def oauth_code2openid(code: str) -> str: return openid +# --------------------------------------------------------------------------- +# 小程序码(wxacode)—— 微信「扫一扫」自动打开小程序到指定页面 + 携带 scene +# --------------------------------------------------------------------------- + +_WX_ACODE_URL = "https://api.weixin.qq.com/wxa/getwxacodeunlimit?access_token={token}" +_WX_TOKEN_URL = "https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid={appid}&secret={secret}" + +# 缓存 access_token(模块级,线程安全;微信 token 有效期约 7200s) +_token_cache: dict = {"token": "", "expires_at": 0.0} +_token_lock = threading.Lock() + + +async def _wx_access_token() -> str: + """获取小程序 access_token(缓存)。未配置 WECHAT_APPID/SECRET 时抛 WechatError。""" + if not config.WECHAT_APPID or not config.WECHAT_SECRET: + raise WechatError("WECHAT_APPID/WECHAT_SECRET 未配置,无法生成小程序码") + now = time.monotonic() + with _token_lock: + if _token_cache["token"] and now < _token_cache["expires_at"]: + return _token_cache["token"] + try: + async with httpx.AsyncClient(timeout=10) as client: + resp = await client.get( + _WX_TOKEN_URL.format(appid=config.WECHAT_APPID, secret=config.WECHAT_SECRET), + ) + data = resp.json() + except Exception as exc: # noqa: BLE001 + raise WechatError(f"wechat access_token request failed: {exc}") from exc + token = data.get("access_token", "") + if not token: + raise WechatError(f"wechat access_token error: {data.get('errmsg', data)}") + with _token_lock: + _token_cache["token"] = token + _token_cache["expires_at"] = now + int(data.get("expires_in", 7200)) - 300 + return token + + +async def get_wxacode(scene: str, page: str = "pages/scan-login/index") -> bytes: + """生成小程序码(wxacode)图片,扫码自动打开小程序指定页面并携带 scene。 + + scene 限 32 字符(本服务用短 token)。返回 PNG 字节。 + """ + token = await _wx_access_token() + try: + async with httpx.AsyncClient(timeout=15) as client: + resp = await client.post( + _WX_ACODE_URL.format(token=token), + json={"scene": scene, "page": page, "check_path": False, "env_version": "release"}, + ) + except Exception as exc: # noqa: BLE001 + raise WechatError(f"wechat getwxacodeunlimit failed: {exc}") from exc + if resp.status_code != 200 or resp.headers.get("content-type", "").startswith("application/json"): + # 出错时返回 JSON 错误体 + try: + err = resp.json() + raise WechatError(f"wechat getwxacodeunlimit error: {err.get('errcode')} {err.get('errmsg','')}") + except WechatError: + raise + except Exception: + raise WechatError("wechat getwxacodeunlimit returned non-image") + return resp.content + + class WxQrSessionStore: """扫码登录会话:scene → {status, token, profile, expires_at}(内存 + TTL)。