71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
import httpx
|
|
|
|
from app.config import get_settings
|
|
|
|
|
|
class WechatMiniClientError(Exception):
|
|
pass
|
|
|
|
|
|
class WechatMiniClient:
|
|
"""微信小程序 code2session / getPhoneNumber 客户端。
|
|
|
|
未配置 appid/secret 时视为 mock 模式，由上层跳过真实调用。
|
|
"""
|
|
|
|
CODE2SESSION_URL = "https://api.weixin.qq.com/sns/jscode2session"
|
|
ACCESS_TOKEN_URL = "https://api.weixin.qq.com/cgi-bin/token"
|
|
GET_PHONE_URL = "https://api.weixin.qq.com/wxa/business/getuserphonenumber"
|
|
|
|
def __init__(self) -> None:
|
|
settings = get_settings()
|
|
self.appid = settings.wx_mini_appid
|
|
self.secret = settings.wx_mini_secret
|
|
self.timeout = 10.0
|
|
|
|
@property
|
|
def configured(self) -> bool:
|
|
return bool(self.appid and self.secret)
|
|
|
|
def code2session(self, js_code: str) -> dict:
|
|
params = {
|
|
"appid": self.appid,
|
|
"secret": self.secret,
|
|
"js_code": js_code,
|
|
"grant_type": "authorization_code",
|
|
}
|
|
with httpx.Client(timeout=self.timeout, trust_env=False) as client:
|
|
resp = client.get(self.CODE2SESSION_URL, params=params)
|
|
data = resp.json()
|
|
if data.get("errcode"):
|
|
raise WechatMiniClientError(f"code2session 失败: {data.get('errmsg')}")
|
|
return data
|
|
|
|
def _access_token(self) -> str:
|
|
params = {
|
|
"grant_type": "client_credential",
|
|
"appid": self.appid,
|
|
"secret": self.secret,
|
|
}
|
|
with httpx.Client(timeout=self.timeout, trust_env=False) as client:
|
|
resp = client.get(self.ACCESS_TOKEN_URL, params=params)
|
|
data = resp.json()
|
|
token = data.get("access_token")
|
|
if not token:
|
|
raise WechatMiniClientError(f"获取 access_token 失败: {data.get('errmsg')}")
|
|
return token
|
|
|
|
def get_phone_number(self, phone_code: str) -> str:
|
|
token = self._access_token()
|
|
url = f"{self.GET_PHONE_URL}?access_token={token}"
|
|
with httpx.Client(timeout=self.timeout, trust_env=False) as client:
|
|
resp = client.post(url, json={"code": phone_code})
|
|
data = resp.json()
|
|
if data.get("errcode"):
|
|
raise WechatMiniClientError(f"getPhoneNumber 失败: {data.get('errmsg')}")
|
|
phone_info = data.get("phone_info") or {}
|
|
phone = phone_info.get("purePhoneNumber") or phone_info.get("phoneNumber")
|
|
if not phone:
|
|
raise WechatMiniClientError("getPhoneNumber 未返回手机号")
|
|
return phone
|