payment-platform/backend/app/utils/sms_util.py

import enum
import os
import json
from typing import Optional, Dict, Any
from collections import namedtuple
from functools import lru_cache

from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_credentials.models import Config as CredentialConfig
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client
from alibabacloud_dysmsapi20170525 import models as dysmsapi_20170525_models
from alibabacloud_tea_util import models as util_models
from pydantic import BaseModel, Field, AliasChoices
from pydantic_settings import SettingsConfigDict, BaseSettings

from app.config.path_conf import ENV_DIR
from app.core.logger import log
from app.core.exceptions import CustomException

_SmsTemplate = namedtuple("_SmsTemplate", ["template_name", "template_code", "template_param_fn"])

class SmsTemplateEnum(enum.Enum):
    VERIFICATION_CODE = _SmsTemplate(
        template_name="verify",
        template_code="SMS_333796424",
        template_param_fn=lambda code: f'{{"code": "{code}"}}'
    )

    @staticmethod
    def get_template_by_name(template_name: str) -> "_SmsTemplate":
        for template in SmsTemplateEnum:
            if template.value.template_name == template_name:
                return template.value
        raise CustomException(f"未找到模板: template_name={ template_name}")


class SendSmsRequest(BaseModel):
    phone_numbers: str = Field(..., description="支持向不同的手机号码发送短信，手机号码之间以半角逗号（,）分隔。"
                                                "上限为 1000 个手机号码。批量发送相对于单条发送，及时性稍有延迟。"
                                                "验证码类型的短信，建议单条发送。")
    sign_name: str = Field(default="湖南钱程似锦技术服务", description="短信签名名称。")
    template_code: str = Field(..., description="短信模板 Code。")
    template_param: Optional[str] = Field(default=None, description="短信模板变量对应的实际值，请传入JSON 字符串。"
                                                                    "当您选择的模板内容含有变量时，此参数必填。参数个数应与模板内变量个数一致。")


class AliyunConfig(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=ENV_DIR / f".env.{os.getenv('ENVIRONMENT')}",
        env_file_encoding="utf-8",
        extra="ignore",
        case_sensitive=True,
    )

    access_key_id: str = Field(
        default=None, validation_alias=AliasChoices("ALIBABA_CLOUD_ACCESS_KEY_ID"), description="必填参数，从环境变量中获取AccessKey ID ")
    access_key_secret: str = Field(
        default=None, validation_alias=AliasChoices("ALIBABA_CLOUD_ACCESS_KEY_SECRET"), description="必填参数，从环境变量中获取AccessKey Secret")

    @staticmethod
    @lru_cache(maxsize=1)
    def get_credential_config() -> "AliyunConfig":
        """"""
        return AliyunConfig()


class SmsSender:

    @classmethod
    def get_client(cls) -> Dysmsapi20170525Client:
        credentials_config = CredentialConfig(
            type='access_key',
            access_key_id=AliyunConfig.get_credential_config().access_key_id,
            access_key_secret=AliyunConfig.get_credential_config().access_key_secret,
        )

        credential = CredentialClient(config=credentials_config)

        config = open_api_models.Config(
            credential=credential,
            endpoint='dysmsapi.aliyuncs.com'
        )

        return Dysmsapi20170525Client(config)

    @classmethod
    async def send_sms(cls, sms_request: SendSmsRequest) -> bool:
        """
        发送短信
        
        参数:
        - sms_request (SendSmsRequest): 短信发送请求对象
        
        返回:
        - Dict[str, Any]: 短信发送响应结果
        
        异常:
        - Exception: 发送失败时抛出异常
        """
        client = SmsSender.get_client()
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(**sms_request.model_dump(exclude_none=True))
        
        runtime = util_models.RuntimeOptions()
        
        phone_numbers = sms_request.phone_numbers
        log.info(f"开始发送短信: 手机号={phone_numbers}, 模板={sms_request.template_code}, 参数={sms_request.template_param}")
        
        try:
            resp = await client.send_sms_with_options_async(send_sms_request, runtime)
            # 将响应转换为字典
            resp_dict = json.loads(json.dumps(resp, default=str))
            # 检查发送结果
            if hasattr(resp.body, "code") and resp.body.code == 'OK':
                log.info("短信发送成功: 手机号={}, BizID={}, 请求ID={}",
                         phone_numbers, resp.body.biz_id, resp.body.request_id)
                return True

            log.warning("短信发送异常: 手机号={}, 响应={}", phone_numbers, resp_dict)
            raise CustomException(f"{resp.body.message}")

        except Exception as error:
            # 打印异常栈
            error_message = str(error.message) if hasattr(error, 'message') else str(error)
            raise CustomException(f"短信发送失败: {error_message}")