128 lines
4.2 KiB
Python
128 lines
4.2 KiB
Python
"""
|
||
微信小程序服务模块
|
||
|
||
提供微信小程序登录功能。
|
||
"""
|
||
import httpx
|
||
from typing import Optional
|
||
|
||
from core.config import settings
|
||
from logger.logging import get_logger
|
||
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
class WechatService:
|
||
"""微信小程序服务类"""
|
||
|
||
@staticmethod
|
||
async def code2session(code: str) -> Optional[dict]:
|
||
"""
|
||
通过微信登录凭证获取 session 信息
|
||
|
||
Args:
|
||
code: 微信登录凭证
|
||
|
||
Returns:
|
||
dict: {"openid": str, "session_key": str, "unionid": str (可选)}
|
||
"""
|
||
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
||
logger.error("微信小程序配置缺失")
|
||
return None
|
||
|
||
url = "https://api.weixin.qq.com/sns/jscode2session"
|
||
params = {
|
||
"appid": settings.wechat_app_id,
|
||
"secret": settings.wechat_app_secret,
|
||
"js_code": code,
|
||
"grant_type": "authorization_code"
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.get(url, params=params)
|
||
data = response.json()
|
||
|
||
if "errcode" in data and data["errcode"] != 0:
|
||
logger.error(f"微信登录失败: {data.get('errmsg')}")
|
||
return None
|
||
|
||
return {
|
||
"openid": data.get("openid"),
|
||
"session_key": data.get("session_key"),
|
||
"unionid": data.get("unionid")
|
||
}
|
||
except Exception as e:
|
||
logger.exception(f"微信登录异常: {e}")
|
||
return None
|
||
|
||
@staticmethod
|
||
async def get_phone_number(phone_code: str) -> Optional[str]:
|
||
"""
|
||
通过手机号授权码获取用户手机号
|
||
|
||
微信新版 API，使用 getPhoneNumber 返回的 code 获取手机号
|
||
|
||
Args:
|
||
phone_code: 手机号授权码 (getPhoneNumber 返回的 code)
|
||
|
||
Returns:
|
||
str: 用户手机号，失败返回 None
|
||
"""
|
||
if not settings.wechat_app_id or not settings.wechat_app_secret:
|
||
logger.error("微信小程序配置缺失")
|
||
return None
|
||
|
||
access_token = await WechatService._get_access_token()
|
||
if not access_token:
|
||
return None
|
||
|
||
url = "https://api.weixin.qq.com/wxa/business/getuserphonenumber"
|
||
|
||
try:
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.post(
|
||
url,
|
||
params={"access_token": access_token},
|
||
json={"code": phone_code}
|
||
)
|
||
data = response.json()
|
||
|
||
if data.get("errcode", 0) != 0:
|
||
logger.error(f"获取手机号失败: {data.get('errmsg')}")
|
||
return None
|
||
|
||
phone_info = data.get("phone_info", {})
|
||
return phone_info.get("purePhoneNumber") or phone_info.get("phoneNumber")
|
||
except Exception as e:
|
||
logger.exception(f"获取手机号异常: {e}")
|
||
return None
|
||
|
||
@staticmethod
|
||
async def _get_access_token() -> Optional[str]:
|
||
"""
|
||
获取微信小程序 access_token
|
||
|
||
注意：生产环境应缓存 access_token，有效期 2 小时
|
||
"""
|
||
url = "https://api.weixin.qq.com/cgi-bin/token"
|
||
params = {
|
||
"grant_type": "client_credential",
|
||
"appid": settings.wechat_app_id,
|
||
"secret": settings.wechat_app_secret
|
||
}
|
||
|
||
try:
|
||
async with httpx.AsyncClient() as client:
|
||
response = await client.get(url, params=params)
|
||
data = response.json()
|
||
|
||
if "access_token" in data:
|
||
return data["access_token"]
|
||
|
||
logger.error(f"获取 access_token 失败: {data.get('errmsg')}")
|
||
return None
|
||
except Exception as e:
|
||
logger.exception(f"获取 access_token 异常: {e}")
|
||
return None
|