ASJServer/Object/QuecCloudObject.py

# -*- encoding: utf-8 -*-
"""
@File    : QuecCloudObject.py
@Time    : 2025/11/13 14:31
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import hashlib
import logging
# 1. 标准库导入
import time
from typing import Dict, Any, Optional, Final, List

# 2. 第三方库导入
import requests
from requests.exceptions import RequestException

# 3. 本地应用导入（假设日志配置在项目config模块中）
from Ansjer.config import LOGGER

# 类型注解日志对象
LOGGER: logging.Logger


class QuecCloudApiClient:
    """
    移远通信安防按量计费OpenAPI工具类
    封装系统级参数处理、签名算法及流量池相关业务查询接口
    遵循项目代码规范，支持可扩展、可复用的API调用
    """
    # 基础配置常量（全大写蛇形命名）
    BASE_URL: Final[str] = "https://api.quectel.com/openapi/router"
    TIME_DIFF_THRESHOLD: Final[int] = 300  # UTC时间戳最大误差（5分钟）
    DEFAULT_TIMEOUT: Final[int] = 10  # 请求超时时间（秒）
    MAX_PAGE_SIZE: Final[int] = 100  # 分页查询最大每页条数

    def __init__(self, app_key: str, secret: str):
        """
        初始化API客户端

        Args:
            app_key (str): 应用键（从移远平台API管理页面获取）
            secret (str): 应用密钥（与appKey对应，需妥善保管）
        """
        self.app_key = app_key
        self.secret = secret
        self.session = requests.Session()  # 复用会话提升性能

    def _generate_sign(self, params: Dict[str, Any]) -> str:
        """
        生成签名串（遵循文档1.6签名算法规范）

        Args:
            params (Dict[str, Any]): 待签名的所有参数（系统级+业务级）

        Returns:
            str: 十六进制SHA1签名字符串

        Raises:
            ValueError: 当参数为空或签名过程失败时抛出
        """
        try:
            # 1. 过滤空值参数，按参数名ASCII升序排列
            # 注意：必须将所有参数值转换为字符串，并过滤None值
            filtered_params = {}
            for k, v in params.items():
                if v is not None:
                    # 将所有值转换为字符串
                    filtered_params[k] = str(v)

            # 按参数名升序排序
            sorted_params = sorted(filtered_params.items(), key=lambda x: x[0])

            # 2. 拼接参数名和参数值
            param_str = "".join([f"{k}{v}" for k, v in sorted_params])

            # 3. 首尾添加secret
            sign_str = f"{self.secret}{param_str}{self.secret}"

            # 调试：打印签名相关信息
            LOGGER.debug(f"参与签名的参数: {sorted_params}")
            LOGGER.debug(f"签名字符串: {sign_str}")

            # 4. SHA1加密并转为十六进制字符串
            sha1 = hashlib.sha1()
            sha1.update(sign_str.encode("UTF-8"))
            signature = sha1.hexdigest().upper()

            LOGGER.debug(f"生成的签名: {signature}")
            return signature

        except Exception as e:
            error_msg = f"签名生成失败: {str(e)}"
            LOGGER.error(f"{error_msg} 行号: {e.__traceback__.tb_lineno}")
            raise ValueError(error_msg) from e

    def _get_utc_timestamp(self) -> int:
        """
        获取当前UTC时间戳（秒），确保与服务器时间误差不超过5分钟

        Returns:
            int: UTC时间戳

        Raises:
            RuntimeError: 当系统时间异常时抛出
        """
        timestamp = int(time.time())
        # 此处可添加时间校准逻辑（如对接NTP服务器），避免本地时间偏差
        return timestamp

    def _send_request(self, method: str, business_params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """
        发送API请求（统一请求处理逻辑）

        Args:
            method (str): 服务方法名（如fc.function.pool.package.getall）
            business_params (Optional[Dict[str, Any]]): 业务级参数，可选

        Returns:
            Dict[str, Any]: 接口返回的JSON数据

        Raises:
            RequestException: 网络请求异常时抛出
            ValueError: 接口返回错误时抛出
        """
        # 1. 构建系统级参数
        system_params = {
            "appKey": self.app_key,
            "t": self._get_utc_timestamp(),
            "method": method
            # 注意：sign参数不参与签名生成
        }

        # 2. 合并所有参数（过滤空值）
        all_params = system_params.copy()
        if business_params:
            # 过滤掉业务参数中的None值
            all_params.update({k: v for k, v in business_params.items() if v is not None})

        # 3. 生成签名（使用所有参数，但不包括sign本身）
        sign = self._generate_sign(all_params)

        # 4. 将签名添加到请求参数中
        all_params["sign"] = sign

        LOGGER.info(f"API请求参数 - 方法: {method}, 时间戳: {all_params['t']}")
        LOGGER.debug(f"完整请求参数: {all_params}")

        try:
            # 5. 发送POST请求（支持form-data格式）
            response = self.session.post(
                url=self.BASE_URL,
                data=all_params,
                timeout=self.DEFAULT_TIMEOUT,
                headers={"Content-Type": "application/x-www-form-urlencoded"}
            )
            response.raise_for_status()  # 抛出HTTP状态码异常
            result = response.json()
            LOGGER.info(f"API响应结果 - 方法: {method}, 状态: {result.get('resultCode')}")

            # 6. 校验接口返回状态
            result_code = result.get("resultCode", -1)
            if result_code != 0:
                error_msg = f"接口调用失败 - 错误码: {result_code}, 错误信息: {result.get('errorMessage', '未知错误')}"
                LOGGER.error(error_msg)
                raise ValueError(error_msg)

            return result
        except RequestException as e:
            error_msg = f"网络请求异常 - 方法: {method}, 错误: {str(e)}"
            LOGGER.error(f"{error_msg} 行号: {e.__traceback__.tb_lineno}")
            raise RequestException(error_msg) from e
        except ValueError as e:
            # 已记录日志，直接向上抛出
            raise
        except Exception as e:
            error_msg = f"请求处理异常 - 方法: {method}, 错误: {str(e)}"
            LOGGER.error(f"{error_msg} 行号: {e.__traceback__.tb_lineno}")
            raise RuntimeError(error_msg) from e

    def query_enterprise_pool_packages(self) -> List[Dict[str, Any]]:
        """
        3.1.1 企业套餐查询接口
        查询企业与移远约定的流量池计费套餐列表

        Returns:
            List[Dict[str, Any]]: 套餐信息列表，包含id、name、flowsize等字段

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> packages = client.query_enterprise_pool_packages()
            >>> print(packages[0]["name"])
            "100G/360天"
        """
        method = "fc.function.pool.package.getall"
        result = self._send_request(method=method)
        # 按文档规范，返回数据在data字段中
        return result.get("data", [])

    def query_enterprise_pools(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None,
            poolcode: Optional[str] = None,
            page_no: Optional[int] = 1,
            page_size: Optional[int] = 10
    ) -> Dict[str, Any]:
        """
        3.1.2 企业流量池查询接口
        查询企业账户下的流量池列表，支持多条件过滤和分页

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一
            poolcode (Optional[str]): 指定流量池编号，精准查询
            page_no (Optional[int]): 分页页码，默认1
            page_size (Optional[int]): 分页大小，默认10，最大100

        Returns:
            Dict[str, Any]: 流量池查询结果，包含分页信息和流量池列表

        Raises:
            ValueError: 当分页参数无效时抛出

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> result = client.query_enterprise_pools(poolcode="PC1629946707")
            >>> print(result["data"][0]["poolname"])
            "测试流量池"
        """
        # 参数校验
        if page_size and page_size > self.MAX_PAGE_SIZE:
            raise ValueError(f"分页大小不能超过{self.MAX_PAGE_SIZE}")

        # 构建业务参数
        business_params = {
            "msisdn": msisdn,
            "iccid": iccid,
            "poolcode": poolcode,
            "pageNo": page_no,
            "pageSize": page_size
        }

        method = "fc.function.pool.list"
        result = self._send_request(method=method, business_params=business_params)
        return result

    def query_single_card_info(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        4.1.1 单卡信息查询接口
        查询物联卡的详细信息，包括基本状态、流量使用、定向信息等

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一

        Returns:
            Dict[str, Any]: 单卡详细信息

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> card_info = client.query_single_card_info(iccid="89860446091891282224")
            >>> print(card_info["msisdn"], card_info["status"])
        """
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        business_params = {
            "msisdn": msisdn,
            "iccid": iccid
        }

        method = "fc.function.card.info"
        result = self._send_request(method=method, business_params=business_params)

        # 移除resultCode和errorMessage，直接返回数据主体
        result.pop("resultCode", None)
        result.pop("errorMessage", None)
        return result

    def query_single_card_realtimestatus(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        4.1.8.资产实时状态
        查询资产的实时状态

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一

        Returns:
            Dict[str, Any]: 单卡详细信息

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出
        """
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        business_params = {
            "msisdn": msisdn,
            "iccid": iccid
        }

        method = "fc.function.card.realtimestatus"
        result = self._send_request(method=method, business_params=business_params)

        # 移除resultCode和errorMessage，直接返回数据主体
        result.pop("resultCode", None)
        result.pop("errorMessage", None)
        return result

    def query_section_day_flow(
            self,
            section: Optional[str] = None,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        4.1.7.单卡日流量查询
        资产指定时间段的流量查询

        Args:
            section (Optional[str]): 查询时间段，20201020
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一

        Returns:
            Dict[str, Any]: 单卡详细信息

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出
        """
        if not section:
            raise ValueError("section参数必须提供")
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        business_params = {
            "section": section,
            "msisdn": msisdn,
            "iccid": iccid
        }

        method = "fc.function.billing.section.dayflow"
        result = self._send_request(method=method, business_params=business_params)

        # 移除resultCode和errorMessage，直接返回数据主体
        result.pop("resultCode", None)
        result.pop("errorMessage", None)
        return result

    def query_orientation_info(
            self,
            strategycode: Optional[str] = None
    ) -> List[Dict[str, Any]]:
        """
        4.1.11 定向信息查询接口
        查询企业名下的定向策略信息

        Args:
            strategycode (Optional[str]): 定向策略编号，不传则查询所有定向信息

        Returns:
            List[Dict[str, Any]]: 定向策略信息列表

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> strategies = client.query_orientation_info()
            >>> print(f"查询到{len(strategies)}个定向策略")
        """
        business_params = {
            "strategycode": strategycode
        }

        method = "fc.function.orientation.info"
        result = self._send_request(method=method, business_params=business_params)
        return result.get("data", [])

    def manage_card_flow(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None,
            notify_emails: Optional[str] = None,
            alarm_threshold1: Optional[str] = None,
            alarm_threshold2: Optional[str] = None,
            alarm_threshold3: Optional[str] = None,
            alarm_type: Optional[int] = None,
            alarm_status: Optional[int] = None,
            pause_flow_threshold: Optional[int] = None,
            pause_status: Optional[int] = None,
            pause_type: Optional[int] = None
    ) -> bool:
        """
        4.2.1 单卡流量管理接口
        支持对单卡的流量使用设置预警、阈值，达到预警进行告警，达到阈值进行停机

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一
            notify_emails (Optional[str]): 通知邮箱，多个邮箱以英文逗号隔开
            alarm_threshold1 (Optional[str]): 预警阶段1
            alarm_threshold2 (Optional[str]): 预警阶段2
            alarm_threshold3 (Optional[str]): 预警阶段3
            alarm_type (Optional[int]): 预警阶段类型，1百分比，2数值
            alarm_status (Optional[int]): 预警启用状态，0停用，1启用
            pause_flow_threshold (Optional[int]): 停机阈值(MB)
            pause_status (Optional[int]): 停机阈值启用状态，0停用，1启用
            pause_type (Optional[int]): 停机类型，1当月流量，2当月流量池内流量

        Returns:
            bool: 操作是否成功

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> success = client.manage_card_flow(
            ...     iccid="89860446091891282224",
            ...     notify_emails="admin@example.com",
            ...     alarm_threshold1="1024",
            ...     alarm_type=2,
            ...     alarm_status=1,
            ...     pause_flow_threshold=20480,
            ...     pause_status=1,
            ...     pause_type=1
            ... )
        """
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        # 构建流量管理参数JSON
        param_json = {}

        if notify_emails is not None:
            param_json["notifyemails"] = notify_emails
        if alarm_threshold1 is not None:
            param_json["alarmthreshold1"] = alarm_threshold1
        if alarm_threshold2 is not None:
            param_json["alarmthreshold2"] = alarm_threshold2
        if alarm_threshold3 is not None:
            param_json["alarmthreshold3"] = alarm_threshold3
        if alarm_type is not None:
            param_json["alarmtype"] = alarm_type
        if alarm_status is not None:
            param_json["alarmstatus"] = alarm_status
        if pause_flow_threshold is not None:
            param_json["pauseflowthreshold"] = pause_flow_threshold
        if pause_status is not None:
            param_json["pausestatus"] = pause_status
        if pause_type is not None:
            param_json["pausetype"] = pause_type

        import json
        param_json_str = json.dumps(param_json, ensure_ascii=False)

        business_params = {
            "msisdn": msisdn,
            "iccid": iccid,
            "paramJson": param_json_str
        }

        method = "fc.function.billing.card.alarm.new"
        self._send_request(method=method, business_params=business_params)
        return True

    def suspend_single_card(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None
    ) -> bool:
        """
        4.2.2 单卡停机接口
        主动停机处于正常状态的卡

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一

        Returns:
            bool: 操作是否成功

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> success = client.suspend_single_card(iccid="89860446091891282224")
        """
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        business_params = {
            "msisdn": msisdn,
            "iccid": iccid
        }

        method = "fc.function.card.pause"
        self._send_request(method=method, business_params=business_params)
        return True

    def resume_single_card(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None
    ) -> bool:
        """
        4.2.3 单卡复机接口
        主动复机处于停机状态的卡

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一

        Returns:
            bool: 操作是否成功

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> success = client.resume_single_card(iccid="89860446091891282224")
        """
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        business_params = {
            "msisdn": msisdn,
            "iccid": iccid
        }

        method = "fc.function.card.resume"
        self._send_request(method=method, business_params=business_params)
        return True

    def close_gprs_service(self, msisdn: Optional[str] = None, iccid: Optional[str] = None) -> bool:
        """
        4.2.4 单卡GPRS关闭接口
        主动关闭物联卡的GPRS(数据服务)。只能处理隶属于该企业下的卡。

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid任选其一。
            iccid (Optional[str]): 集成电路卡识别码，与msisdn任选其一。

        Returns:
            Dict[str, Any]: 接口返回的JSON数据，包含resultCode和errorMessage字段。

        Raises:
            ValueError: 当msisdn和iccid均为空时抛出。
        """
        try:
            if not msisdn and not iccid:
                raise ValueError("必须提供msisdn或iccid参数")

            method = "fc.function.card.gprs.close"
            business_params = {}
            if msisdn:
                business_params["msisdn"] = msisdn
            if iccid:
                business_params["iccid"] = iccid
            self._send_request(method=method, business_params=business_params)
            return True
        except Exception as e:
            LOGGER.error(f"{iccid}关闭GPRS服务失败: {str(e)}")
            return False

    def open_gprs_service(self, msisdn: Optional[str] = None, iccid: Optional[str] = None) -> bool:
        """
        4.2.5 单卡GPRS开启接口
        主动开启物联卡的GPRS(数据服务)。只能处理隶属于该企业下的卡。

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid任选其一。
            iccid (Optional[str]): 集成电路卡识别码，与msisdn任选其一。

        Returns:
            Dict[str, Any]: 接口返回的JSON数据，包含resultCode和errorMessage字段。

        Raises:
            ValueError: 当msisdn和iccid均为空时抛出。
        """

        try:
            if not msisdn and not iccid:
                raise ValueError("必须提供msisdn或iccid参数")

            method = "fc.function.card.gprs.open"
            business_params = {}
            if msisdn:
                business_params["msisdn"] = msisdn
            if iccid:
                business_params["iccid"] = iccid
            self._send_request(method=method, business_params=business_params)
            return True
        except Exception as e:
            LOGGER.error(f"{iccid}开启GPRS服务失败: {str(e)}")
            return False

    def activate_flow_card(
            self,
            msisdn: Optional[str] = None,
            iccid: Optional[str] = None
    ) -> bool:
        """
        4.2.6 流量卡激活接口
        激活名下的物联卡（同一卡号30分钟内不能重复办理）

        Args:
            msisdn (Optional[str]): 物联卡号码，与iccid二选一
            iccid (Optional[str]): 集成电路卡识别码，与msisdn二选一

        Returns:
            bool: 操作是否成功

        Raises:
            ValueError: 当msisdn和iccid都未提供时抛出

        Example:
            >>> client = QuecCloudApiClient("appKey", "secret")
            >>> success = client.activate_flow_card(iccid="89860446091891282224")
        """
        if not msisdn and not iccid:
            raise ValueError("msisdn和iccid参数必须至少提供一个")

        business_params = {
            "msisdn": msisdn,
            "iccid": iccid
        }

        method = "fc.function.card.activate"
        self._send_request(method=method, business_params=business_params)
        return True