SMANAGE/gvsds/wechat/sms.py

"""模块：阿里云短信

直接 HTTP 调用阿里云短信 API（dysmsapi），避免 SDK 依赖。
签名计算用 HMAC-SHA1 + Base64（阿里云 RPC API 规范）。

配置（secrets.ini [aliyun_sms] 段）：
    ACCESS_KEY / ACCESS_SECRET / SIGN_NAME
配置（app.ini [aliyun_sms] 段）：
    VERIFICATION_CODE_TEMPLATE  验证码模板 code

常用 API：
    from gvsds.wechat import SMS
    SMS.SendCode(phone, code)                     # 发送验证码
    SMS.Send(phone, template_code, params)        # 通用单发
    SMS.SendBatch(phones, template_code, params)  # 批量发送

返回：{'success': bool, 'biz_id': str, 'msg': str}
"""
import base64
import hashlib
import hmac
import json
import logging
import time
import urllib.parse
import uuid

from gvsds.app_config import get as _app_get
from gvsds.secrets import get as _secret_get

logger = logging.getLogger(__name__)

_SMS_ENDPOINT = 'https://dysmsapi.aliyuncs.com'
_API_VERSION = '2017-05-25'


class SMS:
    """阿里云短信发送器（HTTP 直接调用）。"""

    @staticmethod
    def _GetConfig():
        """读取阿里云短信配置。"""
        access_key = _secret_get('aliyun_sms', 'ACCESS_KEY')
        access_secret = _secret_get('aliyun_sms', 'ACCESS_SECRET')
        sign_name = _secret_get('aliyun_sms', 'SIGN_NAME')
        if not access_key or not access_secret or not sign_name:
            raise RuntimeError('[SMS] 未配置 aliyun_sms ACCESS_KEY/ACCESS_SECRET/SIGN_NAME')
        return access_key, access_secret, sign_name

    @staticmethod
    def _PercentEncode(value):
        """阿里云规范的特殊百分号编码。"""
        if value is None:
            return ''
        if not isinstance(value, str):
            value = str(value)
        encoded = urllib.parse.quote(value, safe='')
        encoded = encoded.replace('+', '%20').replace('*', '%2A').replace('%7E', '~')
        return encoded

    @classmethod
    def _Sign(cls, params, access_secret):
        """计算阿里云 RPC API 签名（HMAC-SHA1 + Base64）。"""
        sorted_items = sorted(params.items())
        canonical = '&'.join(
            f'{cls._PercentEncode(k)}={cls._PercentEncode(v)}'
            for k, v in sorted_items
        )
        sign_string = 'GET&' + cls._PercentEncode('/') + '&' + cls._PercentEncode(canonical)
        hmac_obj = hmac.new(
            (access_secret + '&').encode('utf-8'),
            sign_string.encode('utf-8'),
            hashlib.sha1,
        )
        return base64.b64encode(hmac_obj.digest()).decode('utf-8')

    @classmethod
    def _Invoke(cls, params):
        """调用阿里云短信 API 并解析响应。"""
        access_key, access_secret, _ = cls._GetConfig()
        common = {
            'Format': 'JSON',
            'Version': _API_VERSION,
            'AccessKeyId': access_key,
            'SignatureMethod': 'HMAC-SHA1',
            'Timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
            'SignatureVersion': '1.0',
            'SignatureNonce': uuid.uuid4().hex,
        }
        all_params = {**common, **params}
        all_params['Signature'] = cls._Sign(all_params, access_secret)
        try:
            import requests
            resp = requests.get(_SMS_ENDPOINT, params=all_params, timeout=10)
            data = resp.json()
        except Exception as e:
            logger.error('[SMS] 请求阿里云失败: %s', e, exc_info=True)
            return {'success': False, 'biz_id': '', 'msg': f'请求异常: {e}'}
        biz_id = data.get('BizId', '')
        code = data.get('Code', '')
        if code == 'OK':
            return {'success': True, 'biz_id': biz_id, 'msg': 'OK'}
        msg = data.get('Message', 'unknown')
        logger.warning('[SMS] 发送失败 Code=%s Message=%s', code, msg)
        return {'success': False, 'biz_id': biz_id, 'msg': f'{code}: {msg}'}

    @staticmethod
    def _StringifyParams(params):
        """把 dict 转成 JSON 字符串。"""
        if params is None:
            return '{}'
        if isinstance(params, str):
            return params
        return json.dumps(params, ensure_ascii=False, separators=(',', ':'))

    @classmethod
    def SendCode(cls, phone, code):
        """发送验证码（使用配置中的 VERIFICATION_CODE_TEMPLATE）。"""
        template_code = _app_get('aliyun_sms', 'VERIFICATION_CODE_TEMPLATE')
        if not template_code:
            raise RuntimeError('[SMS] 未配置 aliyun_sms VERIFICATION_CODE_TEMPLATE')
        return cls.Send(phone, template_code, {'code': str(code)})

    @classmethod
    def Send(cls, phone, template_code, params=None):
        """通用单发。params 是 dict，如 {'code': '123456'}。"""
        if not phone:
            return {'success': False, 'biz_id': '', 'msg': 'phone 为空'}
        _, _, sign_name = cls._GetConfig()
        req_params = {
            'Action': 'SendSms',
            'PhoneNumbers': str(phone),
            'SignName': sign_name,
            'TemplateCode': template_code,
            'TemplateParam': cls._StringifyParams(params),
        }
        return cls._Invoke(req_params)

    @classmethod
    def SendBatch(cls, phones, template_code, params=None):
        """批量发送（同模板同参数广播给所有号码）。

        阿里云 SendBatchSms 要求 PhoneNumberJson/SignNameJson/TemplateParamJson
        均为 JSON 数组，且长度一致。此处将同一签名、同一参数套用到所有号码。
        """
        phones = list(phones or [])
        if not phones:
            return {'success': False, 'biz_id': '', 'msg': 'phones 为空'}
        _, _, sign_name = cls._GetConfig()
        phone_list = [str(p) for p in phones]
        sign_list = [sign_name] * len(phone_list)
        param_dict = params if isinstance(params, dict) else {}
        param_list = [param_dict] * len(phone_list)
        req_params = {
            'Action': 'SendBatchSms',
            'PhoneNumberJson': json.dumps(phone_list, ensure_ascii=False),
            'SignNameJson': json.dumps(sign_list, ensure_ascii=False),
            'TemplateCode': template_code,
            'TemplateParamJson': json.dumps(param_list, ensure_ascii=False),
        }
        return cls._Invoke(req_params)