83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
"""
|
||
阿里云号码认证服务工具类
|
||
用于一键登录：App 端获取 token → 后端用 token 换取真实手机号
|
||
"""
|
||
import logging
|
||
from django.conf import settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
try:
|
||
from alibabacloud_dypnsapi20170525.client import Client
|
||
from alibabacloud_tea_openapi.models import Config
|
||
from alibabacloud_dypnsapi20170525.models import GetMobileRequest
|
||
PHONE_AUTH_SDK_AVAILABLE = True
|
||
except ImportError:
|
||
PHONE_AUTH_SDK_AVAILABLE = False
|
||
|
||
|
||
class PhoneAuthClient:
|
||
"""阿里云号码认证客户端（单例）"""
|
||
|
||
_instance = None
|
||
|
||
def __new__(cls):
|
||
if cls._instance is None:
|
||
cls._instance = super().__new__(cls)
|
||
cls._instance._initialized = False
|
||
return cls._instance
|
||
|
||
def __init__(self):
|
||
if self._initialized:
|
||
return
|
||
|
||
if not PHONE_AUTH_SDK_AVAILABLE:
|
||
self.client = None
|
||
self._initialized = True
|
||
return
|
||
|
||
auth_config = settings.ALIYUN_PHONE_AUTH
|
||
if not auth_config.get('ACCESS_KEY_ID'):
|
||
self.client = None
|
||
self._initialized = True
|
||
return
|
||
|
||
config = Config(
|
||
access_key_id=auth_config['ACCESS_KEY_ID'],
|
||
access_key_secret=auth_config['ACCESS_KEY_SECRET'],
|
||
endpoint='dypnsapi.aliyuncs.com',
|
||
)
|
||
self.client = Client(config)
|
||
self._initialized = True
|
||
|
||
def get_mobile(self, access_token):
|
||
"""
|
||
用 App 端 SDK 获取的 token 换取真实手机号
|
||
:param access_token: App 端 SDK 返回的 token
|
||
:return: (phone: str|None, error_msg: str)
|
||
"""
|
||
if not self.client:
|
||
logger.warning('号码认证 SDK 未配置')
|
||
return None, '号码认证服务未配置'
|
||
|
||
try:
|
||
request = GetMobileRequest(access_token=access_token)
|
||
response = self.client.get_mobile(request)
|
||
|
||
if response.body.code == 'OK':
|
||
mobile = response.body.get_mobile_result_dto.mobile
|
||
logger.info('号码认证成功: %s', mobile)
|
||
return mobile, ''
|
||
else:
|
||
msg = response.body.message or response.body.code
|
||
logger.error('号码认证失败: %s', msg)
|
||
return None, msg
|
||
except Exception as e:
|
||
logger.error('号码认证异常: %s', str(e))
|
||
return None, str(e)
|
||
|
||
|
||
def get_phone_auth_client():
|
||
"""获取号码认证客户端单例"""
|
||
return PhoneAuthClient()
|