Maas-Model-platform/phone/sms_service.py

"""
阿里云短信服务

提供验证码发送、余额预警短信发送功能
"""
import os
import json
import random
import logging
from typing import Optional

logger = logging.getLogger(__name__)

SMS_ACCESS_KEY_ID = os.getenv("SMS_ACCESS_KEY_ID", os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID", ""))
SMS_ACCESS_KEY_SECRET = os.getenv("SMS_ACCESS_KEY_SECRET", os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET", ""))
SMS_SIGN_NAME = os.getenv("SMS_SIGN_NAME", "智创空间")
SMS_TEMPLATE_VERIFY = os.getenv("SMS_TEMPLATE_CODE_VERIFY", "")   # 验证码模板
SMS_TEMPLATE_WARN = os.getenv("SMS_TEMPLATE_CODE_WARN", "")       # 余额预警模板

# 验证码有效期（秒）
CODE_TTL = 300  # 5分钟
# 同一手机号发送间隔（秒）
SEND_INTERVAL = 60


def _get_client():
    from alibabacloud_dysmsapi20170525.client import Client
    from alibabacloud_tea_openapi import models as open_api_models
    config = open_api_models.Config(
        access_key_id=SMS_ACCESS_KEY_ID,
        access_key_secret=SMS_ACCESS_KEY_SECRET,
    )
    config.endpoint = "dysmsapi.aliyuncs.com"
    return Client(config)


def send_sms(phone: str, template_code: str, params: dict) -> bool:
    """发送短信，返回是否成功"""
    try:
        from alibabacloud_dysmsapi20170525 import models as sms_models
        client = _get_client()
        request = sms_models.SendSmsRequest(
            phone_numbers=phone,
            sign_name=SMS_SIGN_NAME,
            template_code=template_code,
            template_param=json.dumps(params, ensure_ascii=False),
        )
        resp = client.send_sms(request)
        if resp.body.code == "OK":
            logger.info(f"短信发送成功: phone={phone}, template={template_code}")
            return True
        else:
            logger.warning(f"短信发送失败: phone={phone}, code={resp.body.code}, msg={resp.body.message}")
            return False
    except Exception as e:
        logger.error(f"短信发送异常: phone={phone}, error={e}")
        return False


class SmsCodeService:
    """验证码服务（基于 Redis）"""

    REDIS_PREFIX_CODE = "sms:code:"
    REDIS_PREFIX_INTERVAL = "sms:interval:"

    def __init__(self):
        from app.core.redis import redis_manager
        self._redis_manager = redis_manager

    @property
    def redis(self):
        return self._redis_manager.get_client()

    async def send_code(self, phone: str) -> tuple[bool, str]:
        """
        发送验证码
        返回 (success, message)
        """
        if not self.redis:
            return False, "短信服务暂不可用"

        # 检查发送间隔
        interval_key = f"{self.REDIS_PREFIX_INTERVAL}{phone}"
        if await self.redis.exists(interval_key):
            ttl = await self.redis.ttl(interval_key)
            return False, f"请 {ttl} 秒后再试"

        # 生成6位验证码
        code = str(random.randint(100000, 999999))

        # 发送短信
        if not SMS_TEMPLATE_VERIFY:
            logger.warning("未配置验证码短信模板 SMS_TEMPLATE_CODE_VERIFY")
            # 开发模式：不实际发送，直接存储
        else:
            ok = send_sms(phone, SMS_TEMPLATE_VERIFY, {"code": code})
            if not ok:
                return False, "短信发送失败，请稍后重试"

        # 存储验证码
        code_key = f"{self.REDIS_PREFIX_CODE}{phone}"
        await self.redis.setex(code_key, CODE_TTL, code)
        # 设置发送间隔
        await self.redis.setex(interval_key, SEND_INTERVAL, "1")

        logger.info(f"验证码已发送: phone={phone}, code={code}")
        return True, "验证码已发送"

    async def verify_code(self, phone: str, code: str, delete_after: bool = True) -> bool:
        """验证验证码，验证成功后默认删除"""
        if not self.redis:
            logger.warning(f"Redis 不可用，验证码验证失败: phone={phone}")
            return False
        code_key = f"{self.REDIS_PREFIX_CODE}{phone}"
        stored = await self.redis.get(code_key)
        logger.info(f"验证码校验: phone={phone}, input={code}, stored={stored}")
        if stored and stored == code:
            if delete_after:
                await self.redis.delete(code_key)
            return True
        return False


# 全局单例
sms_code_service = SmsCodeService()


class TenantWarnSmsService:
    """企业余额预警短信服务"""

    # 已发送预警记录（避免重复发送），key: tenant_id:level
    REDIS_PREFIX_WARN = "sms:warn:"
    # 同一档位预警间隔（秒）：24小时
    WARN_INTERVAL = 86400

    def __init__(self):
        from app.core.redis import redis_manager
        self._redis_manager = redis_manager

    @property
    def redis(self):
        return self._redis_manager.get_client()

    async def send_warn(self, tenant_id: int, phone: str, company_name: str,
                        balance: float, level: str) -> bool:
        """发送余额预警短信，同一档位24小时内不重复发送"""
        if not SMS_TEMPLATE_WARN:
            logger.warning("未配置预警短信模板 SMS_TEMPLATE_CODE_WARN")
            return False

        # 检查是否已发送过该档位
        if self.redis:
            warn_key = f"{self.REDIS_PREFIX_WARN}{tenant_id}:{level}"
            if await self.redis.exists(warn_key):
                return False  # 已发送，跳过

        ok = send_sms(phone, SMS_TEMPLATE_WARN, {
            "companyName": company_name,
            "remainAmount": f"{balance:.2f}",
        })

        if ok and self.redis:
            warn_key = f"{self.REDIS_PREFIX_WARN}{tenant_id}:{level}"
            await self.redis.setex(warn_key, self.WARN_INTERVAL, "1")

        return ok

    async def check_and_warn(self, db) -> list[dict]:
        """
        检查所有企业余额，对达到预警阈值的企业发送短信
        返回已发送的记录列表
        """
        from app.models.tenant import Tenant, TenantApplication
        from decimal import Decimal
        from sqlalchemy import and_

        sent = []
        tenants = db.query(Tenant).filter(Tenant.status == "active").all()

        for tenant in tenants:
            balance = float(tenant.balance or 0)
            level = None
            if balance <= 0:
                level = "critical"
            elif balance <= 100:
                level = "low"
            elif balance <= 200:
                level = "medium"
            elif balance <= 500:
                level = "warning"

            if not level:
                continue

            # 获取联系电话（从 TenantApplication 里取）
            app = db.query(TenantApplication).filter(
                TenantApplication.tenant_id == tenant.id
            ).order_by(TenantApplication.id.desc()).first()

            phone = app.contact_phone if app else None
            if not phone:
                continue

            ok = await self.send_warn(
                tenant_id=tenant.id,
                phone=phone,
                company_name=tenant.company_name,
                balance=balance,
                level=level,
            )
            if ok:
                sent.append({
                    "tenant_id": tenant.id,
                    "company_name": tenant.company_name,
                    "phone": phone,
                    "balance": balance,
                    "level": level,
                })

        return sent


tenant_warn_sms = TenantWarnSmsService()