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

import enum
import os
import json
from typing import Optional
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

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

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


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) -> None:
        client = SmsSender.get_client()
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(**sms_request.model_dump(exclude_none=True))
        runtime = util_models.RuntimeOptions()
        try:
            resp = await client.send_sms_with_options_async(send_sms_request, runtime)
            print(json.dumps(resp, default=str, indent=2))
        except Exception as error:
            # 此处仅做打印展示，请谨慎对待异常处理，在工程项目中切勿直接忽略异常。
            # 错误 message
            print(error.message)
            # 诊断地址
            print(error.data.get("Recommend"))