""" 短信服务实现(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()