6296392e80
完成首页数据模型的统一重构，具体变更如下： 1. 重构公共首页API接口，移除`playRecommendations`字段，将玩法推荐数据统一放入`experiences`字段 2. 更新前后端类型定义、序列化逻辑与前端页面组件，适配新的API响应结构 3. 为微信登录相关接口添加`trust_env=True`配置，支持企业代理环境并新增`socksio`依赖 4. 新增图片画廊上传组件，重构SingleImageUploader组件支持自定义宽高比 5. 重构Toast与Select组件的实现与样式，统一后台UI设计系统 6. 移除个人中心页面不必要的返回事件与顶部标题组件，优化详情卡片布局 7. 新增微信接口单元测试，更新官方文档与测试用例适配变更 8. 删除过期的文档图片资源
85 lines
3.0 KiB
Python
85 lines
3.0 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,
|
|
trust_env=True,
|
|
)
|
|
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,
|
|
trust_env=True,
|
|
)
|
|
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()
|