xiaoxia-saas/packages/adapters/sms/sms_service.py

"""
短信服务实现（Noop + 阿里云）
"""

from __future__ import annotations

import logging
import os

logger = logging.getLogger(__name__)


class NoopSmsService:
    """空实现短信服务 - 开发/测试环境用，只打日志不真发"""

    def send_verification_code(self, phone: str, code: str) -> bool:
        logger.info("[NoopSMS] 发送验证码到 %s: %s", phone, code)
        return True

    def send_template_sms(self, phone: str, template_id: str, params: dict) -> bool:
        logger.info("[NoopSMS] 发送模板短信到 %s, template=%s, params=%s", phone, template_id, params)
        return True


class AliyunSmsService:
    """阿里云短信服务"""

    def __init__(
        self,
        access_key_id: str | None = None,
        access_key_secret: str | None = None,
        sign_name: str | None = None,
        verify_template_id: str | None = None,
    ):
        self.access_key_id = access_key_id or os.environ.get("ALIYUN_SMS_ACCESS_KEY_ID", "")
        self.access_key_secret = access_key_secret or os.environ.get("ALIYUN_SMS_ACCESS_KEY_SECRET", "")
        self.sign_name = sign_name or os.environ.get("ALIYUN_SMS_SIGN_NAME", "小应剪辑")
        self.verify_template_id = verify_template_id or os.environ.get("ALIYUN_SMS_VERIFY_TEMPLATE_ID", "SMS_123456789")

    def send_verification_code(self, phone: str, code: str) -> bool:
        return self.send_template_sms(phone, self.verify_template_id, {"code": code})

    def send_template_sms(self, phone: str, template_id: str, params: dict) -> bool:
        try:
            import json

            from alibabacloud_dysmsapi20170525 import models as dysmsapi_models
            from alibabacloud_dysmsapi20170525.client import Client as DysmsapiClient
            from alibabacloud_tea_openapi import models as open_api_models

            config = open_api_models.Config(
                access_key_id=self.access_key_id,
                access_key_secret=self.access_key_secret,
            )
            config.endpoint = "dysmsapi.aliyuncs.com"
            client = DysmsapiClient(config)

            request = dysmsapi_models.SendSmsRequest(
                phone_numbers=phone,
                sign_name=self.sign_name,
                template_code=template_id,
                template_param=json.dumps(params),
            )
            response = client.send_sms(request)
            body = response.body
            if body.code == "OK":
                logger.info("阿里云短信发送成功: phone=%s, template=%s", phone, template_id)
                return True
            else:
                logger.error("阿里云短信发送失败: code=%s, message=%s", body.code, body.message)
                return False
        except ImportError:
            logger.error("阿里云短信 SDK 未安装，请 pip install alibabacloud-dysmsapi20170525")
            return False
        except Exception as e:
            logger.error("阿里云短信发送异常: %s", e, exc_info=True)
            return False


def get_sms_service() -> "NoopSmsService | AliyunSmsService":
    """获取短信服务实例"""
    provider = os.environ.get("SMS_PROVIDER", "noop").lower()
    if provider == "aliyun":
        return AliyunSmsService()
    return NoopSmsService()