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 from alibabacloud_credentials.client import Client as CredentialClient # type: ignore use_chain = bool(settings.ALIYUN_USE_DEFAULT_CREDENTIALS) or (not settings.ALIBABA_CLOUD_ACCESS_KEY_ID or not settings.ALIBABA_CLOUD_ACCESS_KEY_SECRET) if not use_chain and (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") if str(settings.ALIYUN_SMS_SIGN_NAME).upper().startswith("SMS_"): raise RuntimeError("短信签名配置错误:签名不应为模板码") if settings.ALIYUN_SMS_TEMPLATE_CODE_VERIFY and settings.ALIYUN_SMS_TEMPLATE_CODE_REPORT and settings.ALIYUN_SMS_TEMPLATE_CODE_VERIFY == settings.ALIYUN_SMS_TEMPLATE_CODE_REPORT: raise RuntimeError("短信模板配置错误:验证码模板与报告模板重复") if use_chain: credential = CredentialClient() config = open_api_models.Config(credential=credential) else: 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={} sign={} template={}", phone, settings.ALIYUN_SMS_SIGN_NAME, 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: 返回体映射字典 """ key = settings.ALIYUN_SMS_TEMPLATE_PARAM_CODE_KEY or "code" template = settings.ALIYUN_SMS_TEMPLATE_CODE_VERIFY logger.info("sms.send_code using key={} template={} phone={}", key, template, phone) return self.send_by_template(phone, template, {key: code}) def send_report(self, phone: str) -> Dict[str, Any]: """发送报告通知短信 Args: phone: 接收短信手机号 Returns: 返回体映射字典 """ template = settings.ALIYUN_SMS_TEMPLATE_CODE_REPORT logger.info("sms.send_report using template={} phone={}", template, phone) return self.send_by_template(phone, template, {}) sms_client = SMSClient()