alien_py_cloud/common/aliyun_sms_server/sms_client.py

import json
import logging
from datetime import datetime
from typing import Dict, Optional

from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_dysmsapi20170525 import models as dysmsapi_20170525_models
from alibabacloud_tea_util import models as util_models

from alien_gateway.config import settings

logger = logging.getLogger(__name__)


def _create_client():
    access_key_id = getattr(settings, "ALIYUN_ACCESS_KEY_ID", None)
    access_key_secret = getattr(settings, "ALIYUN_ACCESS_KEY_SECRET", None)
    if not access_key_id or not access_key_secret:
        raise ValueError("请设置 ALIYUN_ACCESS_KEY_ID 和 ALIYUN_ACCESS_KEY_SECRET")

    config = open_api_models.Config(
        access_key_id=access_key_id,
        access_key_secret=access_key_secret,
    )
    config.endpoint = "dysmsapi.aliyuncs.com"
    from alibabacloud_dysmsapi20170525.client import Client as Dysmsapi20170525Client
    return Dysmsapi20170525Client(config)


def query_send_status(phone: str, biz_id: str, send_date: Optional[str] = None) -> None:
    if not phone or not biz_id:
        logger.warning("短信查询失败：phone 或 biz_id 为空")
        return

    if not send_date:
        send_date = datetime.now().strftime("%Y%m%d")

    try:
        client = _create_client()
        query_request = dysmsapi_20170525_models.QuerySendDetailsRequest(
            phone_number=phone,
            biz_id=biz_id,
            send_date=send_date,
            page_size=10,
            current_page=1,
        )
        resp = client.query_send_details_with_options(query_request, util_models.RuntimeOptions())
        # logger.info(
        #     "短信查询响应 phone=%s code=%s message=%s total_count=%s",
        #     phone,
        #     getattr(resp.body, "code", None),
        #     getattr(resp.body, "message", None),
        #     getattr(resp.body, "total_count", None),
        # )
        # logger.info("短信查询详情 %s", getattr(resp.body, "sms_send_detail_dto_list", None))
    except Exception as e:
        logger.error(f"短信查询失败 phone={phone}: {e}", exc_info=True)


def send_sms(phone: str, template_param: Dict[str, object], sign_name: str, template_code: str) -> None:
    if not phone:
        logger.warning("短信发送失败：联系电话为空")
        return

    sign_name = sign_name
    template_code = template_code
    if not sign_name or not template_code:
        logger.error("短信配置缺失：请设置 ALIYUN_SMS_SIGN_NAME_CONTRACT/ALIYUN_SMS_TEMPLATE_CODE_CONTRACT")
        return

    try:
        client = _create_client()
        send_sms_request = dysmsapi_20170525_models.SendSmsRequest(
            phone_numbers=phone,
            sign_name=sign_name,
            template_code=template_code,
            template_param=json.dumps(template_param, ensure_ascii=False),
        )
        resp = client.send_sms_with_options(send_sms_request, util_models.RuntimeOptions())
        logger.info(
            "商家入驻协议快到期提醒： 短信发送响应 phone=%s code=%s message=%s biz_id=%s request_id=%s",
            phone,
            getattr(resp.body, "code", None),
            getattr(resp.body, "message", None),
            getattr(resp.body, "biz_id", None),
            getattr(resp.body, "request_id", None),
        )
        if getattr(resp.body, "code", None) != "OK":
            logger.error(
                "短信发送失败 phone=%s code=%s message=%s",
                phone,
                getattr(resp.body, "code", None),
                getattr(resp.body, "message", None),
            )
        else:
            biz_id = getattr(resp.body, "biz_id", None)
            if biz_id:
                query_send_status(phone, biz_id)
    except Exception as e:
        logger.error(f"短信发送失败 phone={phone}: {e}", exc_info=True)