101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
"""
|
||
腾讯云短信服务
|
||
"""
|
||
import random
|
||
import string
|
||
from typing import Optional
|
||
from tencentcloud.common import credential
|
||
from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException
|
||
from tencentcloud.sms.v20210111 import sms_client, models
|
||
from app.config import get_settings
|
||
from app.utils.logger import logger
|
||
|
||
|
||
class SMSService:
|
||
"""腾讯云短信服务类"""
|
||
|
||
def __init__(self):
|
||
"""初始化短信服务"""
|
||
settings = get_settings()
|
||
self.app_id = settings.tencent_sms_app_id
|
||
self.secret_id = settings.tencent_sms_secret_id
|
||
self.secret_key = settings.tencent_sms_secret_key
|
||
self.sign_id = settings.tencent_sms_sign_id
|
||
self.template_id = settings.tencent_sms_template_id
|
||
|
||
# 初始化客户端
|
||
try:
|
||
# 使用密钥认证
|
||
cred = credential.Credential(self.secret_id, self.secret_key)
|
||
self.client = sms_client.SmsClient(cred, "ap-guangzhou")
|
||
logger.info("腾讯云短信服务初始化成功")
|
||
except Exception as e:
|
||
logger.error(f"腾讯云短信服务初始化失败: {e}")
|
||
self.client = None
|
||
|
||
def generate_code(self, length: int = 6) -> str:
|
||
"""
|
||
生成验证码
|
||
|
||
Args:
|
||
length: 验证码长度，默认6位
|
||
|
||
Returns:
|
||
验证码字符串
|
||
"""
|
||
return ''.join(random.choices(string.digits, k=length))
|
||
|
||
async def send_code(self, phone: str, code: str) -> bool:
|
||
"""
|
||
发送验证码短信
|
||
|
||
Args:
|
||
phone: 手机号
|
||
code: 验证码
|
||
|
||
Returns:
|
||
是否发送成功
|
||
"""
|
||
if not self.client:
|
||
logger.error("短信客户端未初始化")
|
||
return False
|
||
|
||
if not self.template_id:
|
||
logger.warning("短信模板ID未配置，跳过发送（开发模式）")
|
||
logger.info(f"【开发模式】验证码: {code} (手机号: {phone})")
|
||
return True
|
||
|
||
try:
|
||
req = models.SendSmsRequest()
|
||
req.SmsSdkAppId = self.app_id
|
||
req.SignName = "成都爱嘉辰科技"
|
||
req.TemplateId = self.template_id
|
||
req.TemplateParamSet = [code] # 只传递验证码参数
|
||
req.PhoneNumberSet = [f"+86{phone}"]
|
||
|
||
resp = self.client.SendSms(req)
|
||
|
||
# 检查发送结果
|
||
if resp.SendStatusSet and len(resp.SendStatusSet) > 0:
|
||
status = resp.SendStatusSet[0]
|
||
if status.Code == "Ok":
|
||
logger.info(f"短信发送成功: {phone}")
|
||
return True
|
||
else:
|
||
logger.error(f"短信发送失败: {status.Code} - {status.Message}")
|
||
return False
|
||
else:
|
||
logger.error("短信发送失败: 无响应状态")
|
||
return False
|
||
|
||
except TencentCloudSDKException as e:
|
||
logger.error(f"腾讯云SDK异常: {e}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"发送短信异常: {e}")
|
||
return False
|
||
|
||
|
||
# 创建全局实例
|
||
sms_service = SMSService()
|