144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
import json
|
|
import time
|
|
import hmac
|
|
import hashlib
|
|
import base64
|
|
import uuid
|
|
import logging
|
|
import aiohttp
|
|
from typing import List, Dict, Any, Optional
|
|
from app.core.config import settings
|
|
|
|
class UniSMSClient:
|
|
"""UniSMS短信客户端"""
|
|
|
|
def __init__(self):
|
|
self.access_key_id = settings.UNISMS_ACCESS_KEY_ID
|
|
self.access_key_secret = settings.UNISMS_ACCESS_KEY_SECRET
|
|
self.base_url = "https://uni.apistd.com"
|
|
self.endpoint = "/2022-12-28/sms/send"
|
|
self.signature_method = "HMAC-SHA256"
|
|
self.signature_version = "1"
|
|
|
|
def _generate_signature(self, string_to_sign: str) -> str:
|
|
"""生成签名"""
|
|
key = self.access_key_secret.encode('utf-8')
|
|
message = string_to_sign.encode('utf-8')
|
|
sign = hmac.new(key, message, digestmod=hashlib.sha256).digest()
|
|
return base64.b64encode(sign).decode('utf-8')
|
|
|
|
def _build_request_headers(self, body: Dict[str, Any]) -> Dict[str, str]:
|
|
"""构建请求头"""
|
|
timestamp = str(int(time.time()))
|
|
nonce = str(uuid.uuid4()).replace('-', '')
|
|
content_md5 = hashlib.md5(json.dumps(body).encode('utf-8')).hexdigest()
|
|
|
|
# 构建待签名字符串
|
|
string_to_sign = "\n".join([
|
|
self.endpoint,
|
|
timestamp,
|
|
nonce,
|
|
content_md5
|
|
])
|
|
|
|
# 生成签名
|
|
signature = self._generate_signature(string_to_sign)
|
|
|
|
# 构建请求头
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"X-Uni-Timestamp": timestamp,
|
|
"X-Uni-Nonce": nonce,
|
|
"X-Uni-Content-MD5": content_md5,
|
|
"X-Uni-Signature-Method": self.signature_method,
|
|
"X-Uni-Signature-Version": self.signature_version,
|
|
"X-Uni-Signature": signature,
|
|
"X-Uni-AccessKeyId": self.access_key_id,
|
|
}
|
|
|
|
return headers
|
|
|
|
async def send_sms(
|
|
self,
|
|
to: str,
|
|
signature: str,
|
|
template_id: str,
|
|
template_data: Dict[str, Any] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
发送短信
|
|
|
|
Args:
|
|
to: 接收短信的手机号码
|
|
signature: 短信签名
|
|
template_id: 短信模板ID
|
|
template_data: 短信模板参数
|
|
|
|
Returns:
|
|
Dict[str, Any]: 发送结果
|
|
"""
|
|
try:
|
|
# 构建请求体
|
|
body = {
|
|
"to": to,
|
|
"signature": signature,
|
|
"templateId": template_id,
|
|
}
|
|
|
|
if template_data:
|
|
body["templateData"] = template_data
|
|
|
|
# 构建请求头
|
|
headers = self._build_request_headers(body)
|
|
|
|
# 发送请求
|
|
url = f"{self.base_url}{self.endpoint}"
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(url, json=body, headers=headers) as response:
|
|
result = await response.json()
|
|
logging.info(f"UniSMS响应: {result}")
|
|
|
|
if result.get("status") != "success":
|
|
logging.error(f"发送短信失败: {result}")
|
|
return {
|
|
"success": False,
|
|
"error": result.get("message", "发送短信失败")
|
|
}
|
|
|
|
return {
|
|
"success": True,
|
|
"message_id": result.get("data", {}).get("messageId", ""),
|
|
"fee": result.get("data", {}).get("fee", 0)
|
|
}
|
|
|
|
except Exception as e:
|
|
logging.error(f"发送短信异常: {str(e)}")
|
|
return {
|
|
"success": False,
|
|
"error": f"发送短信异常: {str(e)}"
|
|
}
|
|
|
|
async def send_verification_code(self, phone: str, code: str) -> Dict[str, Any]:
|
|
"""
|
|
发送验证码短信
|
|
|
|
Args:
|
|
phone: 手机号码
|
|
code: 验证码
|
|
|
|
Returns:
|
|
Dict[str, Any]: 发送结果
|
|
"""
|
|
# 使用验证码短信模板
|
|
template_id = settings.UNISMS_VERIFICATION_TEMPLATE_ID
|
|
signature = settings.UNISMS_SIGNATURE
|
|
template_data = {"code": code}
|
|
|
|
return await self.send_sms(
|
|
to=phone,
|
|
signature=signature,
|
|
template_id=template_id,
|
|
template_data=template_data
|
|
) |