ASJServer/Object/WXTechObject.py

# -*- encoding: utf-8 -*-
"""
@File    : WXTechObject.py
@Time    : 2023/5/6 15:28
@Author  : stephen
@Email   : zhangdongming@asj6.wecom.work
@Software: PyCharm
"""
import hashlib
import json
import random
import time

import requests

from Ansjer.config import WX_TECH_URL, WX_TECH_APP_KEY, TWX_TECH_SECRET, LOGGER
from Object.UnicomObject import UnicomObjeect


class WXTechObject:
    def __init__(self):
        self.headers = {'content-type': 'application/json;charset=utf-8'}

    @staticmethod
    def getSign(secret, **data):
        dict_2 = dict(sorted(data.items(), key=lambda item: item[0], reverse=False))
        data_list = []
        for item in dict_2.items():
            if item[0] and item[1]:
                data_list.append("{}={}".format(item[0], item[1]))
        sign_param = '&'.join(data_list)
        push_key = '&secret={}'.format(secret)
        param = sign_param + push_key
        return hashlib.sha1(param.encode("utf-8")).hexdigest().upper()

    @staticmethod
    def get_system_params():
        """
        请求接口必传系统参数
        """
        nonce = ''.join(str(random.choice(range(1, 10))) for _ in range(5))
        return {
            'appkey': WX_TECH_APP_KEY,
            'nonce': int(nonce),
            'timestamp': int(time.time()),
        }

    @staticmethod
    def get_request_params(**params):
        """
        将业务参数与系统参数合并后进行加签得到请求参数体
        """
        # 获取请求系统参数
        data = WXTechObject().get_system_params()
        # 系统参数与业务参数合并
        params = dict(data, **params)
        # 合并参数后进行升序加密得到sign
        sign = WXTechObject.getSign(TWX_TECH_SECRET, **params)
        params['sign'] = sign
        params = json.dumps(params)
        LOGGER.info('*****五兴API请求数据：{}'.format(params))
        return params

    def get_cards_info(self, **params):
        """
        根据卡号，获取单卡信息及流量使用信息
        @param params: iccid、operator
        @return: 卡信息
        """
        url = WX_TECH_URL + '/api/v2/card/getCardsInfo'
        rq_body = self.get_request_params(**params)
        response = requests.post(url, data=rq_body, headers=self.headers)
        assert response.status_code == 200
        return UnicomObjeect().get_text_dict(response)

    def create_order_package(self, **params):
        """
        卡片套餐订购,主套餐是立即生效，只有流量包可以顺延生效
        @param params: iccid、operator、startType、packageCode
        @return: 卡信息
        """
        url = WX_TECH_URL + '/api/v2/package/orderPackage'
        rq_body = self.get_request_params(**params)
        response = requests.post(url, data=rq_body, headers=self.headers)
        assert response.status_code == 200
        return UnicomObjeect().get_text_dict(response)


if __name__ == '__main__':
    pass
    # data = {'iccid': '8986112128003439900', 'operator': 3}
    # print(WXTechObject().get_cards_info(**data))
    # data = {'iccid': '8986112128003439900', 'operator': 3}
    # print(WXTechObject().get_cards_info(**data))