|
|
@@ -0,0 +1,645 @@
|
|
|
+# -*- 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
|
