邹方成
cc352d3184
refactor: 优化API路由和响应模型 feat(admin): 添加App用户管理接口 feat(sms): 实现阿里云短信服务集成 feat(email): 添加SMTP邮件发送功能 feat(upload): 支持文件上传接口 feat(rate-limiter): 实现手机号限流器 fix: 修复计算步骤入库问题 docs: 更新API文档和测试计划 chore: 更新依赖和配置
91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
import json
|
|
from typing import Optional, Dict, Any
|
|
|
|
from app.settings import settings
|
|
from app.log import logger
|
|
|
|
|
|
class SMSClient:
|
|
def __init__(self) -> None:
|
|
"""初始化短信客户端
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
self.client = None
|
|
|
|
def _ensure_client(self) -> None:
|
|
"""确保客户端初始化
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
if self.client is not None:
|
|
return
|
|
from alibabacloud_tea_openapi import models as open_api_models # type: ignore
|
|
from alibabacloud_dysmsapi20170525.client import Client as DysmsClient # type: ignore
|
|
if not settings.ALIBABA_CLOUD_ACCESS_KEY_ID or not settings.ALIBABA_CLOUD_ACCESS_KEY_SECRET:
|
|
raise RuntimeError("短信凭证未配置：请设置 ALIBABA_CLOUD_ACCESS_KEY_ID/ALIBABA_CLOUD_ACCESS_KEY_SECRET")
|
|
if not settings.ALIYUN_SMS_SIGN_NAME:
|
|
raise RuntimeError("短信签名未配置：请设置 ALIYUN_SMS_SIGN_NAME")
|
|
config = open_api_models.Config(
|
|
access_key_id=settings.ALIBABA_CLOUD_ACCESS_KEY_ID,
|
|
access_key_secret=settings.ALIBABA_CLOUD_ACCESS_KEY_SECRET,
|
|
)
|
|
config.endpoint = settings.ALIYUN_SMS_ENDPOINT
|
|
self.client = DysmsClient(config)
|
|
|
|
def send_by_template(self, phone: str, template_code: str, template_param: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
"""按模板发送短信
|
|
|
|
Args:
|
|
phone: 接收短信手机号
|
|
template_code: 模板 Code
|
|
template_param: 模板变量字典
|
|
|
|
Returns:
|
|
返回体映射字典
|
|
"""
|
|
from alibabacloud_dysmsapi20170525 import models as sms_models # type: ignore
|
|
self._ensure_client()
|
|
req = sms_models.SendSmsRequest(
|
|
phone_numbers=phone,
|
|
sign_name=settings.ALIYUN_SMS_SIGN_NAME,
|
|
template_code=template_code,
|
|
template_param=json.dumps(template_param or {}),
|
|
)
|
|
logger.info("sms.send start phone={} template={}", phone, template_code)
|
|
try:
|
|
resp = self.client.send_sms(req)
|
|
body = resp.body.to_map() if hasattr(resp, "body") else {}
|
|
logger.info("sms.send response code={} request_id={} phone={}", body.get("Code"), body.get("RequestId"), phone)
|
|
return body
|
|
except Exception as e:
|
|
logger.error("sms.provider_error err={}", repr(e))
|
|
return {"Code": "ERROR", "Message": str(e)}
|
|
|
|
def send_code(self, phone: str, code: str) -> Dict[str, Any]:
|
|
"""发送验证码短信
|
|
|
|
Args:
|
|
phone: 接收短信手机号
|
|
code: 验证码
|
|
|
|
Returns:
|
|
返回体映射字典
|
|
"""
|
|
return self.send_by_template(phone, "SMS_498190229", {"code": code})
|
|
|
|
def send_report(self, phone: str) -> Dict[str, Any]:
|
|
"""发送报告通知短信
|
|
|
|
Args:
|
|
phone: 接收短信手机号
|
|
|
|
Returns:
|
|
返回体映射字典
|
|
"""
|
|
return self.send_by_template(phone, "SMS_498140213", {})
|
|
|
|
|
|
sms_client = SMSClient() |