"""
|
短信发送服务
|
"""
|
|
import hashlib
|
import time
|
import requests
|
from typing import List, Optional
|
from loguru import logger
|
from config import settings
|
|
|
class SmsService:
|
"""短信发送服务"""
|
|
def __init__(self):
|
self.api_url = settings.sms_api_url
|
self.username = settings.sms_username
|
self.password = settings.sms_password
|
self.session = requests.Session()
|
self.session.headers.update({
|
"Accept": "application/json",
|
"Content-Type": "application/json;charset=utf-8"
|
})
|
|
def _generate_sign(self, timestamp: int) -> str:
|
"""
|
生成签名
|
计算规则：MD5(userName+timestamp+MD5(password))
|
|
Args:
|
timestamp: 时间戳（毫秒）
|
|
Returns:
|
签名字符串
|
"""
|
# 计算密码的MD5
|
password_md5 = hashlib.md5(self.password.encode('utf-8')).hexdigest()
|
|
# 组合字符串：userName+timestamp+MD5(password)
|
combined_str = f"{self.username}{timestamp}{password_md5}"
|
|
# 计算最终签名
|
sign = hashlib.md5(combined_str.encode('utf-8')).hexdigest()
|
|
return sign
|
|
def send_sms(self, phone_list: List[str], content: str) -> bool:
|
"""
|
发送短信
|
|
Args:
|
phone_list: 手机号码列表
|
content: 短信内容
|
|
Returns:
|
发送成功返回True，失败返回False
|
"""
|
if not settings.sms_enabled:
|
logger.info("短信发送功能已禁用")
|
return True
|
|
if not phone_list:
|
logger.warning("手机号码列表为空")
|
return False
|
|
try:
|
# 生成时间戳（毫秒）
|
timestamp = int(time.time() * 1000)
|
|
# 生成签名
|
sign = self._generate_sign(timestamp)
|
|
# 构建请求参数
|
payload = {
|
"userName": self.username,
|
"content": content,
|
"phoneList": phone_list,
|
"timestamp": timestamp,
|
"sign": sign
|
}
|
|
logger.info(f"发送短信: phones={phone_list}, content_length={len(content)}")
|
|
# 发送请求
|
response = self.session.post(self.api_url, json=payload, timeout=30)
|
response.raise_for_status()
|
|
result = response.json()
|
|
if result.get("code") == 0:
|
msg_id = result.get("msgId")
|
sms_count = result.get("smsCount")
|
logger.info(f"短信发送成功: msgId={msg_id}, smsCount={sms_count}, phones={phone_list}")
|
return True
|
else:
|
logger.error(
|
f"短信发送失败: code={result.get('code')}, message={result.get('message')}, phones={phone_list}"
|
)
|
return False
|
|
except requests.exceptions.RequestException as e:
|
logger.error(f"短信发送网络错误: phones={phone_list}, error={str(e)}")
|
return False
|
except Exception as e:
|
logger.error(f"短信发送异常: phones={phone_list}, error={str(e)}")
|
return False
|
|
def send_notification(self, content: str) -> bool:
|
"""
|
发送通知短信到配置的手机号码列表
|
|
Args:
|
content: 短信内容
|
|
Returns:
|
发送成功返回True，失败返回False
|
"""
|
return self.send_sms(settings.sms_phone_numbers, content)
|
|
def test_connection(self) -> bool:
|
"""
|
测试短信服务连接（发送测试短信）
|
|
Returns:
|
连接成功返回True，失败返回False
|
"""
|
if not settings.sms_enabled:
|
logger.info("短信发送功能已禁用")
|
return True
|
|
# 发送测试短信到第一个号码
|
if settings.sms_phone_numbers:
|
test_phone = [settings.sms_phone_numbers[0]]
|
test_content = "【测试】短信服务连接测试"
|
return self.send_sms(test_phone, test_content)
|
else:
|
logger.warning("没有配置短信接收号码")
|
return False
|
|
|
# 全局短信服务实例
|
sms_service = SmsService()
|
