lowmaster-chen
51ee9f738e
- 新增荣耀、vivo、小米、OPPO 服务端推送分发 - 设备注册和心跳支持 push_vendor/push_token - 增加 /api/v1/device/push-test 延迟验收接口 - 补充数据库迁移、配置示例和厂商推送测试 - 保留本分支已有微信登录联调改动 验证：python -m pytest tests/test_device_push.py tests/test_auth.py tests/test_health.py
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""微信开放平台 App 登录集成。
|
|
|
|
客户端只把微信 SDK 回调拿到的临时 code 传给服务端;AppSecret、access_token
|
|
与用户 openid 都留在服务端处理,避免敏感凭证落到 APK 里。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import certifi
|
|
import httpx
|
|
|
|
from app.core.config import settings
|
|
|
|
_API_BASE = "https://api.weixin.qq.com"
|
|
|
|
|
|
class WechatLoginError(Exception):
|
|
"""微信授权换取用户信息失败。"""
|
|
|
|
|
|
def _ensure_configured() -> None:
|
|
if not settings.WECHAT_APP_ID or not settings.WECHAT_APP_SECRET:
|
|
raise WechatLoginError("wechat app not configured")
|
|
|
|
|
|
def _http_client() -> httpx.Client:
|
|
return httpx.Client(verify=certifi.where())
|
|
|
|
|
|
def code_to_userinfo(code: str) -> dict:
|
|
"""用授权 code 换 openid/unionid,尽力补昵称头像。
|
|
|
|
返回字段:
|
|
- openid: 同一开放平台移动应用下的用户唯一标识
|
|
- unionid: 同一开放平台主体下的用户唯一标识,微信可能不返回
|
|
- nickname/avatar_url: sns/userinfo 可用时返回,失败不阻断登录
|
|
- raw: userinfo 原始响应,便于排障
|
|
"""
|
|
_ensure_configured()
|
|
try:
|
|
with _http_client() as client:
|
|
token_resp = client.get(
|
|
f"{_API_BASE}/sns/oauth2/access_token",
|
|
params={
|
|
"appid": settings.WECHAT_APP_ID,
|
|
"secret": settings.WECHAT_APP_SECRET,
|
|
"code": code,
|
|
"grant_type": "authorization_code",
|
|
},
|
|
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
|
|
)
|
|
token_data = token_resp.json()
|
|
except Exception as e: # noqa: BLE001
|
|
raise WechatLoginError("微信授权请求失败,请稍后重试") from e
|
|
if "openid" not in token_data or "access_token" not in token_data:
|
|
raise WechatLoginError(f"微信授权失败: {token_data.get('errmsg', token_data)}")
|
|
|
|
openid = token_data["openid"]
|
|
unionid = token_data.get("unionid")
|
|
nickname = None
|
|
avatar_url = None
|
|
raw: dict = {}
|
|
|
|
try:
|
|
with _http_client() as client:
|
|
info_resp = client.get(
|
|
f"{_API_BASE}/sns/userinfo",
|
|
params={
|
|
"access_token": token_data["access_token"],
|
|
"openid": openid,
|
|
"lang": "zh_CN",
|
|
},
|
|
timeout=settings.WXPAY_REQUEST_TIMEOUT_SEC,
|
|
)
|
|
raw = info_resp.json()
|
|
if "errcode" not in raw:
|
|
nickname = raw.get("nickname") or None
|
|
avatar_url = raw.get("headimgurl") or None
|
|
unionid = unionid or raw.get("unionid")
|
|
except Exception: # noqa: BLE001
|
|
# openid 已拿到即可登录;昵称头像只是展示增强。
|
|
pass
|
|
|
|
return {
|
|
"openid": openid,
|
|
"unionid": unionid,
|
|
"nickname": nickname,
|
|
"avatar_url": avatar_url,
|
|
"raw": raw,
|
|
}
|