123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
"""微信小程序登录：code2session、手机号快速验证。"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from typing import Any, Dict, Optional
|
||
|
||
import httpx
|
||
|
||
from app.core.config import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_WX_CODE2SESSION = "https://api.weixin.qq.com/sns/jscode2session"
|
||
_WX_TOKEN = "https://api.weixin.qq.com/cgi-bin/token"
|
||
_WX_PHONE = "https://api.weixin.qq.com/wxa/business/getuserphonenumber"
|
||
|
||
|
||
class WechatMiniAuthError(Exception):
|
||
def __init__(self, message: str, *, wx_errcode: Optional[int] = None):
|
||
super().__init__(message)
|
||
self.wx_errcode = wx_errcode
|
||
|
||
|
||
def normalize_cn_mobile(raw: Optional[str]) -> str:
|
||
if raw is None:
|
||
return ""
|
||
s = re.sub(r"\s+", "", str(raw).strip())
|
||
if s.startswith("+86"):
|
||
s = s[3:]
|
||
elif s.startswith("86") and len(s) == 13:
|
||
s = s[2:]
|
||
return s
|
||
|
||
|
||
def _require_mini_credentials() -> tuple:
|
||
appid = (settings.WXPAY_MINI_APPID or "").strip()
|
||
secret = (settings.WXPAY_SECRET or "").strip()
|
||
if not appid or not secret:
|
||
raise WechatMiniAuthError("未配置小程序凭证（WXPAY_MINI_APPID / WXPAY_SECRET）")
|
||
return appid, secret
|
||
|
||
|
||
def code2session(login_code: str) -> Dict[str, Any]:
|
||
code = (login_code or "").strip()
|
||
if not code:
|
||
raise WechatMiniAuthError("loginCode 不能为空")
|
||
appid, secret = _require_mini_credentials()
|
||
with httpx.Client(timeout=15.0) as client:
|
||
r = client.get(
|
||
_WX_CODE2SESSION,
|
||
params={
|
||
"appid": appid,
|
||
"secret": secret,
|
||
"js_code": code,
|
||
"grant_type": "authorization_code",
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if data.get("errcode"):
|
||
logger.warning("code2session failed: %s", data)
|
||
raise WechatMiniAuthError(
|
||
"微信登录凭证无效或已过期", wx_errcode=data.get("errcode")
|
||
)
|
||
openid = data.get("openid")
|
||
if not openid:
|
||
raise WechatMiniAuthError("微信未返回 openid")
|
||
return {
|
||
"openid": openid,
|
||
"session_key": data.get("session_key"),
|
||
"unionid": data.get("unionid"),
|
||
}
|
||
|
||
|
||
def _get_access_token() -> str:
|
||
appid, secret = _require_mini_credentials()
|
||
with httpx.Client(timeout=15.0) as client:
|
||
r = client.get(
|
||
_WX_TOKEN,
|
||
params={
|
||
"grant_type": "client_credential",
|
||
"appid": appid,
|
||
"secret": secret,
|
||
},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
token = data.get("access_token")
|
||
if not token:
|
||
logger.warning("get access_token failed: %s", data)
|
||
raise WechatMiniAuthError(
|
||
"获取微信 access_token 失败", wx_errcode=data.get("errcode")
|
||
)
|
||
return token
|
||
|
||
|
||
def get_phone_number(phone_code: str) -> str:
|
||
code = (phone_code or "").strip()
|
||
if not code:
|
||
raise WechatMiniAuthError("phoneCode 不能为空")
|
||
token = _get_access_token()
|
||
with httpx.Client(timeout=15.0) as client:
|
||
r = client.post(
|
||
f"{_WX_PHONE}?access_token={token}",
|
||
json={"code": code},
|
||
)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
if data.get("errcode") not in (0, None):
|
||
logger.warning("getuserphonenumber failed: %s", data)
|
||
raise WechatMiniAuthError(
|
||
"获取手机号失败，请改用账号密码登录",
|
||
wx_errcode=data.get("errcode"),
|
||
)
|
||
info = data.get("phone_info") or {}
|
||
phone = normalize_cn_mobile(
|
||
info.get("purePhoneNumber") or info.get("phoneNumber")
|
||
)
|
||
if not phone:
|
||
raise WechatMiniAuthError("微信未返回手机号")
|
||
return phone
|