修改电信API接口

Object/TelecomObject.py

@@ -9,9 +9,10 @@
 import json
 
 import requests
-from requests import RequestException
+import xmltodict
 
 from Ansjer.config import CT_USER_ID, CT_PASSWORD, CT_SECRET_KEY, LOGGER
+from Object.utils import LocalDateTimeUtil
 from Object.utils.DesUtils import DesUtils
 
 
@@ -23,8 +24,30 @@ class TelecomObject:
         self.key1 = CT_SECRET_KEY[0:3]
         self.key2 = CT_SECRET_KEY[3:6]
         self.key3 = CT_SECRET_KEY[6:9]
+        self.url = 'http://api.ct10649.com:9001/m2m_ec/query.do'
         self.session = requests.Session()
 
+    def get_password_enc(self):
+        return DesUtils.str_enc(self.password, self.key1, self.key2, self.key3)
+
+    def get_sign(self, arr):
+        return DesUtils.str_enc(DesUtils.natural_ordering(arr), self.key1, self.key2, self.key3)
+
+    def get_required_arr(self, iccid, method):
+        return [iccid, self.user_id, self.password, method]
+
+    def get_params_dict_by_iccid(self, method, iccid, arr):
+        return {'method': method, 'iccid': iccid,
+                'user_id': self.user_id,
+                'passWord': self.get_password_enc(),
+                'sign': self.get_sign(arr)}
+
+    def get_params_dict_by_access_number(self, method, access_number, arr):
+        return {'method': method, 'access_number': access_number,
+                'user_id': self.user_id,
+                'passWord': self.get_password_enc(),
+                'sign': self.get_sign(arr)}
+
     def query_card_main_status(self, iccid):
         """
         卡主状态查询接口
@@ -35,27 +58,328 @@ class TelecomObject:
         """
         try:
             if not iccid:
-                raise ValueError("*****TelecomObject.query_card_main_status error****ICCID不能为空")
-            url = 'http://api.ct10649.com:9001/m2m_ec/query.do'
+                raise ValueError("error ICCID不能为空")
             method = 'queryCardMainStatus'
-            arr = [iccid, self.user_id, self.password, method]
+            arr = self.get_required_arr(iccid, method)
             # 密码加密
-            password_enc = DesUtils.str_enc(self.password, self.key1, self.key2, self.key3)
-            sign = DesUtils.str_enc(DesUtils.natural_ordering(arr), self.key1, self.key2, self.key3)
-            re_params = {'method': method, 'iccid': iccid,
-                         'user_id': self.user_id,
-                         'passWord': password_enc,
-                         'sign': sign}
+            re_params = self.get_params_dict_by_iccid(method, iccid, arr)
 
-            response = self.session.get(url, params=re_params)
+            response = self.session.get(self.url, params=re_params)
             if response.status_code != 200:
-                raise ValueError(f"*****TelecomObject.query_card_main_status error HTTP请求失败，状态码: {response.status_code}")
+                LOGGER(f"*****TelecomObject.query_card_main_status error HTTP请求失败，状态码: {response.status_code}")
+                return None
 
             result = response.json()
             LOGGER.info(f'*****TelecomObject.query_card_main_status****iccid:{iccid},result:{result}')
             return result
 
-        except RequestException:
-            raise
-        except json.JSONDecodeError:
-            raise
+        except Exception as e:
+            LOGGER.info('***TelecomObject.query_card_main_status:errLine:{}, errMsg:{}'
+                        .format(e.__traceback__.tb_lineno, repr(e)))
+
+            return None
+
+    def query_total_usage_by_date(self, access_number):
+        """
+        总使用量查询接口(时间段) 不可跨月查询
+
+        :param access_number: 11位接入号码
+        :return: 套餐使用量查询结果
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not access_number:
+                raise ValueError("error access_number不能为空")
+            start_date = LocalDateTimeUtil.get_first_day_of_month()
+            end_date = LocalDateTimeUtil.get_last_day_of_month()
+            method = 'queryTotalUsageByDate'
+            arr = [method, self.user_id, access_number, self.password, start_date, end_date]
+            re_params = self.get_params_dict_by_access_number(method, access_number, arr)
+            re_params['startDate'] = start_date
+            re_params['endDate'] = end_date
+
+            response = self.session.get(self.url, params=re_params)
+            if response.status_code != 200:
+                LOGGER.info(f"*****TelecomObject.query_total_usage_by_date error HTTP请求失败，状态码: {response.status_code}")
+                return None
+
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/xml' in content_type or 'text/xml' in content_type:
+                result = xmltodict.parse(msg)
+            elif 'application/json' in content_type or 'text/json' in content_type:
+                result = json.loads(msg)
+                LOGGER.info(f"***TelecomObject.query_total_usage_by_date查询流量异常{access_number},error{result}")
+                return None
+            else:
+                return None
+
+            return result
+
+        except Exception as e:
+            LOGGER.info('***TelecomObject.query_total_usage_by_date:errLine:{}, errMsg:{}'
+                        .format(e.__traceback__.tb_lineno, repr(e)))
+            return None
+
+    def disabled_number(self, access_number, orderTypeId):
+        """
+        停机保号/复机/测试期去激活
+
+        :param access_number: 11位接入号码
+        :param orderTypeId: 19表示停机保号，20表示停机保号后复机,21表示测试期去激活，22表示测试期去激活后回到测试激活。
+        :return: 流量总使用量查询(时间段)
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not access_number:
+                raise ValueError("*****TelecomObject.disabled_number error****access_number不能为空")
+            method = 'disabledNumber'
+            arr = [method, self.user_id, access_number, self.password, orderTypeId, '']
+            re_params = self.get_params_dict_by_access_number(method, access_number, arr)
+            re_params['orderTypeId'] = orderTypeId
+            re_params['acctCd'] = ''
+            response = self.session.get(self.url, params=re_params)
+            if response.status_code != 200:
+                LOGGER.info(f"*****TelecomObject.disabled_number error HTTP请求失败，状态码: {response.status_code}")
+                return None
+
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/xml' in content_type or 'text/xml' in content_type:
+                result = xmltodict.parse(msg)
+            elif 'application/json' in content_type or 'text/json' in content_type:
+                result = json.loads(msg)
+                LOGGER.info(f"***TelecomObject.disabled_number停机/复机异常{access_number},error{result}")
+                return None
+            else:
+                return None
+
+            return result
+
+        except Exception as e:
+            LOGGER.info('***TelecomObject.disabled_number:errLine:{}, errMsg:{}'
+                        .format(e.__traceback__.tb_lineno, repr(e)))
+            return None
+
+    def single_cut_net(self, access_number, action):
+        """
+        单独添加/恢复断网接口
+
+        :param access_number: 11位接入号码
+        :param action: ADD:单独添加断网；DEL:单独恢复断网
+        :return: 单独添加/恢复断网结果
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not access_number:
+                raise ValueError("*****TelecomObject.single_cut_net***error:access_number不能为空")
+            method = 'singleCutNet'
+            arr = [method, self.user_id, access_number, self.password, action]
+            re_params = self.get_params_dict_by_access_number(method, access_number, arr)
+            re_params['action'] = action
+            response = self.session.get(self.url, params=re_params)
+
+            if response.status_code != 200:
+                LOGGER.info(f"*****TelecomObject.query_card_main_status error HTTP请求失败，状态码: {response.status_code}")
+                return None
+
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/xml' in content_type or 'text/xml' in content_type:
+                result = xmltodict.parse(msg)
+            elif 'application/json' in content_type or 'text/json' in content_type:
+                result = json.loads(msg)
+                LOGGER.info(f"***TelecomObject.query_card_main_status停机/复机异常{access_number},error{result}")
+                return None
+            else:
+                return None
+
+            return result
+
+        except Exception as e:
+            LOGGER.info('***TelecomObject.single_cut_net:errLine:{}, errMsg:{}'
+                        .format(e.__traceback__.tb_lineno, repr(e)))
+            return None
+
+    def get_telephone(self, iccid):
+        """
+        接入号码查询接口
+
+        :param iccid: 卡的ICCID号。
+        :return: 接入号码查询结果
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not iccid:
+                raise ValueError("*****TelecomObject.get_telephone error****iccid不能为空")
+            method = 'getTelephone'
+            arr = [method, self.user_id, iccid, self.password]
+            re_params = self.get_params_dict_by_iccid(method, iccid, arr)
+            response = self.session.get(self.url, params=re_params)
+
+            if response.status_code != 200:
+                LOGGER.info(f"*****TelecomObject.get_telephone error HTTP请求失败，状态码: {response.status_code}")
+                return None
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            return json.loads(msg)
+
+        except Exception as e:
+            LOGGER.info('***TelecomObject.get_telephone:errLine:{}, errMsg:{}'
+                        .format(e.__traceback__.tb_lineno, repr(e)))
+            return None
+
+    def query_traffic(self, iccid, need_dtl='0'):
+        """
+        表示流量总使用量查询(当月)
+
+        :param iccid: 卡的ICCID号。
+        :param need_dtl: 0
+        :return: 表示流量总使用量查询(当月)。
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not iccid:
+                raise ValueError("*****TelecomObject.query_traffic error****ICCID不能为空")
+            method = 'queryTraffic'
+            arr = [method, self.user_id, iccid, self.password]
+            re_params = self.get_params_dict_by_iccid(method, iccid, arr)
+            re_params['needDtl'] = need_dtl
+            response = self.session.get(self.url, params=re_params)
+            if response.status_code != 200:
+                LOGGER.info(
+                    f"*****TelecomObject.query_traffic error HTTP请求失败，状态码: {response.status_code}")
+                return None
+
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/xml' in content_type or 'text/xml' in content_type:
+                result = xmltodict.parse(msg)
+            elif 'application/json' in content_type or 'text/json' in content_type:
+                result = json.loads(msg)
+                LOGGER.info(f"***TelecomObject.query_traffic查询流量异常{iccid},error{result}")
+                return None
+            else:
+                return None
+
+            return result
+
+        except Exception as e:
+            LOGGER.info(
+                '***TelecomObject.query_traffic:errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno, repr(e)))
+            return None
+
+    def query_package_batch_flow(self, iccid, month_date):
+        """
+        套餐使用量批量查询接口
+
+        :param iccid: 卡的ICCID号。
+        :param month_date: 例如20160501，表示查询2016年5月份的套餐；如果不填，则为查询实时套餐
+        :return: 套餐使用量查询结果
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not iccid:
+                raise ValueError("*****TelecomObject.query_package_batch_flow error****ICCID不能为空")
+            method = 'queryPakagePlus'
+            arr = [method, self.user_id, iccid, self.password]
+            re_params = self.get_params_dict_by_access_number(method, iccid, arr)
+            response = self.session.get(self.url, params=re_params)
+            if response.status_code != 200:
+                LOGGER.info(f"*****TelecomObject.query_package_batch_flow error HTTP请求失败，状态码: {response.status_code}")
+                return None
+
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/xml' in content_type or 'text/xml' in content_type:
+                result = xmltodict.parse(msg)
+            elif 'application/json' in content_type or 'text/json' in content_type:
+                result = json.loads(msg)
+                LOGGER.info(f"***TelecomObject.query_package_batch_flow查询流量异常{iccid},error{result}")
+                return None
+            else:
+                LOGGER.info("***TelecomObject.query_package_batch_flow无法识别的响应类型: {}".format(content_type))
+                return None
+
+            return result
+
+        except Exception as e:
+            LOGGER.info(
+                '***TelecomObject.query_package_batch_flow:errLine:{}, errMsg:{}'.format(e.__traceback__.tb_lineno,
+                                                                                         repr(e)))
+            return None
+
+    def query_traffic_by_date(self, iccid):
+        """
+        流量总使用量查询(时间段) 会被限流 成功则返回XML
+
+        :param iccid: 卡的ICCID号。
+        :return: 流量总使用量查询(时间段)
+        :raises ValueError: 如果响应为空或HTTP请求失败。
+
+        """
+        try:
+            if not iccid:
+                raise ValueError("*****TelecomObject.query_traffic_by_date error****ICCID不能为空")
+            method = 'queryTrafficByDate'
+            arr = [method, self.user_id, iccid, self.password]
+            re_params = self.get_params_dict_by_iccid(method, iccid, arr)
+            re_params['startDate'] = '20240101'
+            re_params['endDate'] = '20240115'
+            re_params['needDtl'] = '0'
+            response = self.session.get(self.url, params=re_params)
+            if response.status_code != 200:
+                LOGGER.info(f"*****TelecomObject.query_traffic_by_date error HTTP请求失败，状态码: {response.status_code}")
+                return None
+
+            msg = response.text
+            if not msg:
+                return None
+
+            # 检查响应类型
+            content_type = response.headers.get('Content-Type', '')
+            if 'application/xml' in content_type or 'text/xml' in content_type:
+                result = xmltodict.parse(msg)
+            elif 'application/json' in content_type or 'text/json' in content_type:
+                result = json.loads(msg)
+                LOGGER.info(f"***TelecomObject.query_traffic_by_date 查询流量异常{iccid},error{result}")
+                return None
+            else:
+                LOGGER.info("***TelecomObject.query_traffic_by_date 无法识别的响应类型: {}".format(content_type))
+                return None
+
+            return result
+
+        except Exception as e:
+            LOGGER.info('***TelecomObject.query_traffic_by_date:errLine:{}, errMsg:{}'
+                        .format(e.__traceback__.tb_lineno, repr(e)))
+            return None

Object/utils/LocalDateTimeUtil.py

@@ -318,3 +318,24 @@ def get_date_time(timestamp, timezone_offset):
 
     # 转换为日期时间对象
     return datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone(offset))
+
+
+def get_first_day_of_month(strformat="%Y%m%d"):
+    """
+    获取当月第一天日期
+    @param strformat:
+    @return: 返回当月第一天日期
+    """
+    today = datetime.date.today()
+    return datetime.date(today.year, today.month, 1).strftime(strformat)
+
+
+def get_last_day_of_month(strformat="%Y%m%d"):
+    """
+    获取当月最后一天日期
+    @param strformat: 日期格式化
+    @return: 返回当月最后一天日期
+    """
+    today = datetime.date.today()
+    last_day = calendar.monthrange(today.year, today.month)[1]
+    return datetime.date(today.year, today.month, last_day).strftime(strformat)