ASJServer/Object/TelecomObject.py

# -*- encoding: utf-8 -*-
"""
@File    : TelecomObject.py
@Time    : 2024/1/11 10:40
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import json

import requests
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


class TelecomObject:
    def __init__(self):
        self.user_id = CT_USER_ID
        self.password = CT_PASSWORD
        self.secret_key = CT_SECRET_KEY
        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):
        """
        卡主状态查询接口

        :param iccid: 卡的ICCID号。
        :return: 包含卡状态的结果字典。
        :raises ValueError: 如果响应为空或HTTP请求失败。
        """
        try:
            if not iccid:
                raise ValueError("error ICCID不能为空")
            method = 'queryCardMainStatus'
            arr = self.get_required_arr(iccid, method)
            # 密码加密
            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(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 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

    def query_product_information(self, access_number):
        """
        产品资料查询接口
        :param access_number: 接入号码access_number。
        :return: SIM卡的产品资料，包含基础信息、套餐、状态及 断网类型等
        :raises ValueError: 如果响应为空或HTTP请求失败。
        """
        try:
            if not access_number:
                raise ValueError("*****TelecomObject.query_product_information error****access_number不能为空")
            method = 'prodInstQuery'
            arr = [method, self.user_id, access_number, self.password]
            re_params = self.get_params_dict_by_access_number(method, access_number, arr)
            response = self.session.get(self.url, params=re_params)

            if response.status_code != 200:
                LOGGER.info(f"*****TelecomObject.query_product_information 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_product_information 查询产品资料异常:{access_number},error{result}")
                return None
            else:
                LOGGER.info("***TelecomObject.query_product_information 无法识别的响应类型: {}".format(content_type))
                return None
            return result

        except Exception as e:
            LOGGER.info('***TelecomObject.query_product_information:errLine:{}, errMsg:{}'
                        .format(e.__traceback__.tb_lineno, repr(e)))
            return None