Pine
b90539a013
- 登录：/auth/send-code·phone-login·wx-login·wx-phone（复用 RBAC 签发） - 算力：services/compute_client（loopback :3000）+ /admin/compute/ping·provision - /v1 relay 改走 compute-engine（前端不直连引擎） - 模型：User 补 wx_openid/phone；仓储 async create+新方法 - 契约：api/schemas（auth/compute）Pydantic
51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""微信小程序登录服务。
|
|
|
|
``code2session`` 用小程序 ``wx.login`` 的 code 换取 openid。
|
|
未配置 ``WECHAT_APPID`` / ``WECHAT_SECRET`` 时走「测试直通」：直接把 code 当
|
|
openid 返回，便于无真实微信 appid 的环境联调（生产必须配置真实凭据）。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
|
|
from .. import config
|
|
|
|
_WX_SESSION_URL = "https://api.weixin.qq.com/sns/jscode2session"
|
|
|
|
|
|
class WechatError(Exception):
|
|
"""微信登录异常（code 无效 / 网络错误等）。"""
|
|
|
|
|
|
async def code2session(code: str) -> str:
|
|
"""用登录 code 换取 openid。未配置 appid 时直通：``openid = code``。"""
|
|
code = code.strip()
|
|
if not config.WECHAT_APPID or not config.WECHAT_SECRET:
|
|
return code
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
resp = await client.get(
|
|
_WX_SESSION_URL,
|
|
params={
|
|
"appid": config.WECHAT_APPID,
|
|
"secret": config.WECHAT_SECRET,
|
|
"js_code": code,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
)
|
|
data = resp.json()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise WechatError(f"wechat code2session request failed: {exc}") from exc
|
|
|
|
errcode = data.get("errcode", 0)
|
|
if errcode:
|
|
raise WechatError(
|
|
f"wechat code2session error {errcode}: {data.get('errmsg', '')}"
|
|
)
|
|
openid = data.get("openid")
|
|
if not openid:
|
|
raise WechatError("wechat code2session returned no openid")
|
|
return openid
|