bac68842ef
Implement Wechat Mini Program phone number login flow for end customers: - add PhoneLoginIn Pydantic request schema - create wechat.py module for Wechat API interactions and phone code exchange - add customer JWT utilities and require_customer authentication dependency - add new public API endpoints: /api/public/auth/phone-login and /api/public/auth/me - add required environment config variables and update example .env - add comprehensive test cases for the new auth flow and endpoints
83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
import httpx
|
|
|
|
from .config import Settings, get_settings
|
|
|
|
|
|
class WechatConfigError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class WechatApiError(RuntimeError):
|
|
pass
|
|
|
|
|
|
_access_token_cache: dict[str, object] = {"appid": None, "token": None, "expires_at": 0.0}
|
|
|
|
|
|
def _require_miniapp_settings(settings: Settings) -> tuple[str, str]:
|
|
if not settings.wechat_miniapp_appid or not settings.wechat_miniapp_secret:
|
|
raise WechatConfigError("微信小程序登录未配置")
|
|
return settings.wechat_miniapp_appid, settings.wechat_miniapp_secret
|
|
|
|
|
|
def _wechat_json(response: httpx.Response) -> dict:
|
|
try:
|
|
return response.json()
|
|
except ValueError as exc:
|
|
raise WechatApiError("微信登录服务返回异常") from exc
|
|
|
|
|
|
def get_access_token(settings: Settings | None = None) -> str:
|
|
settings = settings or get_settings()
|
|
appid, secret = _require_miniapp_settings(settings)
|
|
now = time.time()
|
|
cached_token = _access_token_cache.get("token")
|
|
if _access_token_cache.get("appid") == appid and isinstance(cached_token, str) and now < float(_access_token_cache.get("expires_at", 0)):
|
|
return cached_token
|
|
|
|
try:
|
|
response = httpx.get(
|
|
"https://api.weixin.qq.com/cgi-bin/token",
|
|
params={"grant_type": "client_credential", "appid": appid, "secret": secret},
|
|
timeout=8,
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
raise WechatApiError("微信登录服务暂时不可用") from exc
|
|
|
|
data = _wechat_json(response)
|
|
token = data.get("access_token")
|
|
if not isinstance(token, str) or not token:
|
|
raise WechatApiError("微信登录服务暂时不可用")
|
|
expires_in = data.get("expires_in")
|
|
ttl = int(expires_in) if isinstance(expires_in, int | str) and str(expires_in).isdigit() else 7200
|
|
_access_token_cache.update({"appid": appid, "token": token, "expires_at": now + max(ttl - 300, 60)})
|
|
return token
|
|
|
|
|
|
def exchange_phone_code(code: str, settings: Settings | None = None) -> str:
|
|
access_token = get_access_token(settings)
|
|
try:
|
|
response = httpx.post(
|
|
"https://api.weixin.qq.com/wxa/business/getuserphonenumber",
|
|
params={"access_token": access_token},
|
|
json={"code": code},
|
|
timeout=8,
|
|
)
|
|
except httpx.HTTPError as exc:
|
|
raise WechatApiError("手机号授权服务暂时不可用") from exc
|
|
|
|
data = _wechat_json(response)
|
|
if data.get("errcode", 0) != 0:
|
|
raise WechatApiError("手机号授权失败，请重试")
|
|
phone_info = data.get("phone_info")
|
|
if not isinstance(phone_info, dict):
|
|
raise WechatApiError("手机号授权服务返回异常")
|
|
phone_number = phone_info.get("phoneNumber") or phone_info.get("purePhoneNumber")
|
|
if not isinstance(phone_number, str) or not phone_number.strip():
|
|
raise WechatApiError("手机号授权服务返回异常")
|
|
return phone_number.strip()
|