shaguabijia-app-server/app/integrations/sms/aliyun.py

"""阿里云号码认证(dypns)短信 provider —— Mode A(阿里云托管验证码)。

与极光(自管码)最大不同:**本服务不生成/不存储验证码**,验证码由阿里云生成+存储+下发+校验。
  - 发码:调 SendSmsVerifyCode,TemplateParam 用 `{"code":"##code##","min":...}` 占位,阿里云生成。
  - 校验:调 CheckSmsVerifyCode,阿里云返回 PASS / UNKNOWN。核验免费。
→ 天然消除极光路径「内存存码、多 worker 不共享」的技术债(发码/校验可落不同 worker,阿里云统一裁决)。

**唯一本地态**:per-phone 连续失败计数(`_verify_attempts`),用于复刻极光「单码失败
`SMS_MAX_VERIFY_ATTEMPTS` 次即作废」的防爆破语义 —— 刻意与极光一致,避免两 provider 行为不同
导致排查困惑。其多 worker 降级特性与极光现状同级;另有 API 层登录频控(设备+IP)做硬兜底。

单号发送频控(冷却)交给阿里云 `Interval` 参数(命中→FREQUENCY_FAIL→429),本地不再维护冷却。

SDK 交互隔离在 `_call_send` / `_call_check` 两个薄封装(惰性 import + 惰性建 client,仿 wxpay
惰性加载),单测 monkeypatch 这两个即可,不触真 SDK / 网络。
"""
from __future__ import annotations

import json
import logging
import time
from threading import Lock

from app.core.config import settings

from .base import SmsError, mock_verify

logger = logging.getLogger("shagua.sms.aliyun")

# 阿里云路径唯一本地态:per-phone 连续失败次数(与极光同语义,防爆破)。
_verify_attempts: dict[str, int] = {}   # phone -> 连续失败次数
_verify_seen: dict[str, float] = {}     # phone -> 最近触碰 epoch(仅供 GC 老化)
_lock = Lock()
_GC_THRESHOLD = 10000                    # 超此阈值,send 时顺手清老于验证码有效期的计数(仿极光 _gc)

# 发码错误码 → (HTTP 码, 用户提示)。未列出的一律 503(供应商不可用)。
_SEND_ERRORS: dict[str, tuple[int, str]] = {
    "MOBILE_NUMBER_ILLEGAL": (400, "手机号无效"),
    "BUSINESS_LIMIT_CONTROL": (429, "今日发送次数过多,请明天再试"),
    "FREQUENCY_FAIL": (429, "发送过于频繁,请稍后再试"),
}
# 需运维介入的配置/开通类错误:打 critical 日志(融合认证未开通 / 参数非法)。
_SEND_CRITICAL_CODES = frozenset({"FUNCTION_NOT_OPENED", "INVALID_PARAMETERS"})

_client = None  # 惰性构建的 SDK client(模块级缓存)


# ============================ 对外:发码 / 校验 ============================

def send_code(phone: str) -> int:
    """发送验证码(阿里云生成+下发)。

    Returns: 距下次可发的秒数(= ALIYUN_SMS_INTERVAL_SEC,冷却由阿里云 Interval 侧执行)。
    Raises: SmsError(手机号无效 400 / 过频·天级流控 429 / 未配置·未开通·其他 503)。
    """
    if settings.SMS_MOCK:
        logger.info("[SMS-aliyun-MOCK] to %s**** (不真发)", phone[:3])
        return settings.ALIYUN_SMS_INTERVAL_SEC
    if not settings.aliyun_sms_configured:
        raise SmsError("短信服务未配置(缺阿里云凭证)", status_code=503)

    result = _call_send(phone)  # 传输/SDK 异常在内部抛 SmsError(503)

    if result["success"] and result["code"] == "OK":
        now = time.time()
        with _lock:
            _gc(now)                          # 顺手清老计数(超阈值才扫)
            _verify_attempts.pop(phone, None)  # 新码 = 新失败预算
            _verify_seen.pop(phone, None)
        logger.info("[SMS-aliyun] sent to %s****", phone[:3])
        return settings.ALIYUN_SMS_INTERVAL_SEC

    code = result["code"]
    logger.error("[SMS-aliyun] send failed code=%s msg=%s", code, result["message"])
    if code in _SEND_CRITICAL_CODES:
        logger.critical("[SMS-aliyun] %s —— 需运维处理(融合认证未开通 / 参数非法)", code)
    status, msg = _SEND_ERRORS.get(code, (503, "短信服务暂不可用,请稍后重试"))
    raise SmsError(msg, status_code=status)


def verify_code(phone: str, code: str) -> bool:
    """校验验证码(阿里云裁决)。

    - **mock**:放行任意 N 位数字(provider 无关,同极光)。
    - **real**:先查本地失败计数(达上限即本地作废,不调阿里云,与极光一致)→ 调 CheckSmsVerifyCode:
      PASS 清计数返 True(一次性);UNKNOWN 计数 +1 返 False;接口异常抛 SmsError(503)。
    """
    if settings.SMS_MOCK:
        ok = mock_verify(code)
        logger.info("[SMS-aliyun-MOCK] verify %s for %s****", "ok" if ok else "fail", phone[:3])
        return ok

    # 失败计数是 best-effort:网络调用不持锁(不能锁跨 IO),故并发下同号可能多放行个位数次。
    # 无碍——API 层登录频控(设备+IP 5/时)是硬上限,阿里云码有效期 + DuplicatePolicy 亦兜底。
    with _lock:
        if _verify_attempts.get(phone, 0) >= settings.SMS_MAX_VERIFY_ATTEMPTS:
            return False  # 已作废:保持计数(直到 send_code 重置),与极光「达上限即作废」一致

    result = _call_check(phone, code)  # 传输/SDK 异常在内部抛 SmsError(503)

    if not (result["success"] and result["code"] == "OK"):
        # 接口层失败(非码错):降级 503,别误报「验证码错误」(400),便于区分排查。
        logger.error("[SMS-aliyun] check failed code=%s msg=%s", result["code"], result["message"])
        raise SmsError("短信服务暂不可用,请稍后重试", status_code=503)

    if result["verify_result"] == "PASS":
        with _lock:
            _verify_attempts.pop(phone, None)  # 验过即清(一次性)
            _verify_seen.pop(phone, None)
        return True

    # UNKNOWN:码错 / 过期 → 失败计数 +1(累计到上限即作废)
    with _lock:
        _verify_attempts[phone] = _verify_attempts.get(phone, 0) + 1
        _verify_seen[phone] = time.time()
    return False


def _gc(now: float) -> None:
    """超阈值时清理老于验证码有效期的失败计数(码早已在阿里云侧失效,计数无意义)。仅持锁调用。"""
    if len(_verify_attempts) <= _GC_THRESHOLD:
        return
    cutoff = now - settings.ALIYUN_SMS_VALID_TIME_SEC
    for p in [p for p, ts in _verify_seen.items() if ts < cutoff]:
        _verify_attempts.pop(p, None)
        _verify_seen.pop(p, None)


# ============================ SDK 接缝(单测 monkeypatch 这两个)============================

def _get_client():
    """惰性构建 dypns SDK client(仿 wxpay 惰性加载:jiguang-only 部署不加载 alibabacloud)。"""
    global _client
    if _client is None:
        from alibabacloud_dypnsapi20170525.client import Client
        from alibabacloud_tea_openapi import models as open_api_models

        cfg = open_api_models.Config(
            access_key_id=settings.ALIYUN_SMS_ACCESS_KEY_ID,
            access_key_secret=settings.ALIYUN_SMS_ACCESS_KEY_SECRET,
            read_timeout=settings.ALIYUN_SMS_TIMEOUT_SEC * 1000,     # SDK 单位 ms
            connect_timeout=settings.ALIYUN_SMS_TIMEOUT_SEC * 1000,
        )
        cfg.endpoint = settings.ALIYUN_SMS_ENDPOINT
        _client = Client(cfg)
    return _client


def _call_send(phone: str) -> dict:
    """调 SendSmsVerifyCode。返回归一化 {success, code, message};import/建 client/调用 任一失败抛 SmsError(503)。"""
    valid_min = max(1, settings.ALIYUN_SMS_VALID_TIME_SEC // 60)
    template_param = json.dumps({"code": "##code##", "min": str(valid_min)}, ensure_ascii=False)
    try:
        # import + 建 req + 调用 全在 try 内:任一 provider 侧失败都归一成 503(保「provider 出问题→503」不变式)
        from alibabacloud_dypnsapi20170525 import models as dypns_models
        req = dypns_models.SendSmsVerifyCodeRequest(
            phone_number=phone,
            sign_name=settings.ALIYUN_SMS_SIGN_NAME,
            template_code=settings.ALIYUN_SMS_TEMPLATE_CODE,
            template_param=template_param,
            code_length=settings.ALIYUN_SMS_CODE_LENGTH,
            valid_time=settings.ALIYUN_SMS_VALID_TIME_SEC,
            interval=settings.ALIYUN_SMS_INTERVAL_SEC,
            scheme_name=settings.ALIYUN_SMS_SCHEME_NAME or None,
        )
        body = _get_client().send_sms_verify_code(req).body
    except Exception as e:
        logger.exception("[SMS-aliyun] send_sms_verify_code 调用异常 phone=%s****", phone[:3])
        raise SmsError("短信服务暂不可用,请稍后重试", status_code=503) from e
    return {"success": bool(body.success), "code": body.code, "message": body.message}


def _call_check(phone: str, code: str) -> dict:
    """调 CheckSmsVerifyCode。返回归一化 {success, code, message, verify_result};import/建 client/调用 任一失败抛 SmsError(503)。"""
    try:
        # import + 建 req + 调用 全在 try 内:任一 provider 侧失败都归一成 503(保「provider 出问题→503」不变式)
        from alibabacloud_dypnsapi20170525 import models as dypns_models
        req = dypns_models.CheckSmsVerifyCodeRequest(
            phone_number=phone,
            verify_code=code,
            scheme_name=settings.ALIYUN_SMS_SCHEME_NAME or None,
        )
        body = _get_client().check_sms_verify_code(req).body
    except Exception as e:
        logger.exception("[SMS-aliyun] check_sms_verify_code 调用异常 phone=%s****", phone[:3])
        raise SmsError("短信服务暂不可用,请稍后重试", status_code=503) from e
    verify_result = getattr(body.model, "verify_result", None) if body.model else None
    return {
        "success": bool(body.success),
        "code": body.code,
        "message": body.message,
        "verify_result": verify_result,
    }