License/backend/app/services/sms.py

"""阿里云短信服务封装"""
import base64
import hashlib
import hmac
import logging
import random
import string
import uuid
from datetime import datetime, timezone
from urllib.parse import quote

import httpx
from app.config import settings

logger = logging.getLogger(__name__)

ALIYUN_ENDPOINT = "https://dysmsapi.aliyuncs.com/"


def _percent_encode(s: str) -> str:
    """RFC 3986 percent encoding"""
    return quote(s, safe="").replace("+", "%20").replace("*", "%2A").replace("%7E", "~")


def _build_signature(method: str, params: dict[str, str], secret: str) -> str:
    """阿里云 POP 签名"""
    sorted_keys = sorted(params.keys())
    canonicalized = "&".join(f"{_percent_encode(k)}={_percent_encode(params[k])}" for k in sorted_keys)
    string_to_sign = f"{method.upper()}&%2F&{_percent_encode(canonicalized)}"
    signing_key = secret + "&"
    h = hmac.new(signing_key.encode(), string_to_sign.encode(), hashlib.sha1)
    return base64.b64encode(h.digest()).decode()


async def _send_sms(phone: str, template_code: str, template_params: dict) -> bool:
    """调用阿里云 SendSms 接口"""
    if not settings.sms_access_key_id or not settings.sms_access_key_secret:
        logger.warning("阿里云短信配置未设置，跳过短信发送")
        return False

    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    params = {
        "AccessKeyId": settings.sms_access_key_id,
        "Action": "SendSms",
        "Format": "JSON",
        "PhoneNumbers": phone,
        "RegionId": "cn-hangzhou",
        "SignName": settings.sms_sign_name,
        "SignatureMethod": "HMAC-SHA1",
        "SignatureNonce": uuid.uuid4().hex,
        "SignatureVersion": "1.0",
        "TemplateCode": template_code,
        "TemplateParam": str(template_params).replace("'", '"'),
        "Timestamp": now,
        "Version": "2017-05-25",
    }

    signature = _build_signature("POST", params, settings.sms_access_key_secret)
    params["Signature"] = signature

    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.post(ALIYUN_ENDPOINT, data=params)
            result = resp.json()
            if result.get("Code") == "OK":
                logger.info("短信发送成功: %s", phone)
                return True
            else:
                logger.error("短信发送失败: %s - %s", result.get("Code"), result.get("Message"))
                return False
    except Exception:
        logger.exception("短信发送异常: %s", phone)
        return False


async def send_sms_code(phone: str, code: str) -> bool:
    """发送验证码短信"""
    return await _send_sms(phone, settings.sms_template_code_verify, {"code": code})


async def send_license_expired(phone: str, company: str) -> bool:
    """发送 License 过期通知短信"""
    return await _send_sms(phone, settings.sms_template_code_expired, {"company": company})


async def send_license_restored(phone: str, company: str) -> bool:
    """发送 License 恢复通知短信"""
    return await _send_sms(phone, settings.sms_template_code_restored, {"company": company})


def generate_verify_code() -> str:
    """生成 6 位数字验证码"""
    return "".join(random.choices(string.digits, k=6))