lowmaster-chen
1913dbcc6e
改动内容：新增微信 code 换 openid 集成、/api/v1/auth/wechat-login 登录接口、微信用户创建/复用逻辑和测试覆盖。 验证方式：python -m pytest tests\\test_auth.py tests\\test_health.py -q 通过。
88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
"""微信开放平台 App 登录 OAuth 封装。
|
|
|
|
客户端只传微信 SDK 返回的临时 code;AppSecret、access_token、refresh_token 都只留在服务端。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
WECHAT_TOKEN_URL = "https://api.weixin.qq.com/sns/oauth2/access_token"
|
|
WECHAT_USERINFO_URL = "https://api.weixin.qq.com/sns/userinfo"
|
|
|
|
|
|
class WechatLoginError(Exception):
|
|
"""微信登录链路失败,由 auth endpoint 转成 502/503。"""
|
|
|
|
|
|
def _ensure_configured() -> None:
|
|
if not settings.WECHAT_APP_ID or not settings.WECHAT_APP_SECRET:
|
|
raise WechatLoginError("wechat app not configured")
|
|
|
|
|
|
def _json(resp: httpx.Response) -> dict[str, Any]:
|
|
try:
|
|
data = resp.json()
|
|
except ValueError as e:
|
|
raise WechatLoginError("微信授权响应格式异常") from e
|
|
if not isinstance(data, dict):
|
|
raise WechatLoginError("微信授权响应格式异常")
|
|
return data
|
|
|
|
|
|
def code_to_userinfo(code: str) -> dict[str, Any]:
|
|
"""用客户端授权 code 换 openid/unionid,并尽力拉取微信昵称头像。"""
|
|
_ensure_configured()
|
|
try:
|
|
with httpx.Client(timeout=8.0) as client:
|
|
token_resp = client.get(
|
|
WECHAT_TOKEN_URL,
|
|
params={
|
|
"appid": settings.WECHAT_APP_ID,
|
|
"secret": settings.WECHAT_APP_SECRET,
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
)
|
|
token_data = _json(token_resp)
|
|
if token_data.get("errcode"):
|
|
raise WechatLoginError(f"微信授权失败: {token_data.get('errmsg', token_data)}")
|
|
|
|
access_token = token_data.get("access_token")
|
|
openid = token_data.get("openid")
|
|
if not access_token or not openid:
|
|
raise WechatLoginError("微信授权响应缺少 openid")
|
|
|
|
info: dict[str, Any] = {
|
|
"openid": openid,
|
|
"unionid": token_data.get("unionid"),
|
|
"nickname": None,
|
|
"avatar_url": None,
|
|
"raw": {"token": token_data},
|
|
}
|
|
|
|
# userinfo 失败不阻断登录:登录主身份是 openid,昵称头像只是展示增强。
|
|
userinfo_resp = client.get(
|
|
WECHAT_USERINFO_URL,
|
|
params={
|
|
"access_token": access_token,
|
|
"openid": openid,
|
|
"lang": "zh_CN",
|
|
},
|
|
)
|
|
userinfo = _json(userinfo_resp)
|
|
if not userinfo.get("errcode"):
|
|
info["unionid"] = userinfo.get("unionid") or info["unionid"]
|
|
info["nickname"] = userinfo.get("nickname")
|
|
info["avatar_url"] = userinfo.get("headimgurl")
|
|
info["raw"]["userinfo"] = userinfo
|
|
|
|
return info
|
|
except WechatLoginError:
|
|
raise
|
|
except httpx.HTTPError as e:
|
|
raise WechatLoginError("微信授权请求失败,请稍后重试") from e
|