# # -*- encoding: utf-8 -*- # """ # @File : UnicomObject.py # @Time : 2022/6/17 11:03 # @Author : stephen # @Email : zhangdongming@asj6.wecom.work # @Software: PyCharm # """ # import base64 # import json # # import requests # from Crypto.Cipher import AES # from decimal import Decimal # # from pysmx.SM3 import hash_msg as sm3_msg # # from Ansjer.config import unicomAppUrl, unicomAppId, unicomAppSecret, unicomTenantId, \ # unicomEncodeKey, unicomIvKey, unicomUserName, unicomPassword, unicomPushKey # from Object.utils.SymmetricCryptoUtil import AESencrypt # # """ # 联通4Gapi # 具体参数查看接口文档 # https://www.showdoc.com.cn/unicomJYHapi/8158648460007467 # """ # # # class UnicomObjeect: # def __init__(self): # self.appUrl = unicomAppUrl # self.appId = unicomAppId # self.appSecret = unicomAppSecret # self.tenantId = unicomTenantId # self.encodeKey = unicomEncodeKey # self.ivKey = unicomIvKey # self.username = unicomUserName # self.password = unicomPassword # self.pushKey = unicomPushKey # self.headers = {'Tenant': self.tenantId, 'content-type': 'application/x-www-form-urlencoded'} # # # pip install snowland-smx # def createSign(self, reverse=False, **sign_params): # """ # 调用接口(API)时需要对请求参数进行签名（sign）验证， # 算法： # 根据参数名称将你的所有请求参数按照字母先后顺序排序: key = value & key = value，对除签名外的所有请求参数按 # key # 做的升序排列。 # 如：将 # foo = 1, bar = 2, baz = 3 # 排序为 # bar = 2, baz = 3, foo = 1 # 参数名和参数值链接后，得到拼装字符串（注意：参数名与参数值左右两边不能包含空格） # bar = 2 & baz = 3 & foo = 1 # 将 # pushkey拼接到参数字符尾部进行 # SM3 # 加密，再转化成大写，格式是 # (SM3(key1=value1 & key2=value2 &...& key= pushKey)).upcase # @param reverse: # @param sign_params: # @return: # """ # dict_2 = dict(sorted(sign_params.items(), key=lambda item: item[0], reverse=reverse)) # data_list = [] # for item in dict_2.items(): # if item[0] and item[1]: # data_list.append("{}={}".format(item[0], item[1])) # val = '&'.join(data_list) # push_key = '&key={}'.format(self.pushKey) # val = val + push_key # return sm3_msg(val).upper() # # def get_login_authorization(self): # """ # 获取登录授权 # 注意登录认证Authorization和登录后的请求认证不一样 # 算法：appId+":"+appSecret base64转码生成 前面加Basic # @return: "Basic " + base64_data # """ # voucher = self.appId + ':' + self.appSecret # base64_data = str(base64.b64encode(voucher.encode("utf-8")), "utf-8") # return "Basic " + base64_data # # def get_encode_password(self): # """ # 获取对称加密 # @return: encrypt_pwd # """ # aes = AESencrypt(self.encodeKey.encode('utf-8'), AES.MODE_CBC, self.ivKey.encode('utf-8'), # paddingMode="ZeroPadding", # characterSet='utf-8') # encrypt_pwd = aes.encryptFromString(self.password) # return encrypt_pwd # # def generate_token(self): # """ # 生成令牌 # @return: token # """ # url = self.appUrl + '/auc/oauth/token' # pwd = self.get_encode_password() # body = {'username': self.username, 'password': pwd, 'grant_type': 'password', 'scope': 'server'} # headers = self.headers # headers['Authorization'] = self.get_login_authorization() # response_data = requests.post(url, data=body, headers=headers) # response_data = json.loads(response_data.text) # return response_data['access_token'] # # def refresh_token(self, refresh_token): # """ # 刷新令牌 # @param refresh_token: # @return: # """ # url = self.appUrl + '/auc/oauth/token?grant_type=refresh_token' # body = {'refresh_token': refresh_token, 'grant_type': 'refresh_token'} # headers = self.headers # headers['Authorization'] = self.get_login_authorization() # response_data = requests.post(url, data=body, headers=headers) # return response_data.text # # def business_unify_headers(self): # """ # 业务统一headers # 在请求头中增加key为Authorization，value为"Bearer " + token # @return: headers # """ # token = self.generate_token() # headers = self.headers # headers['Authorization'] = 'Bearer ' + token # return headers # # # 业务api 注意 get与post 请求数据类型不同 post使用：application/json # def verify_device(self, **re_params): # """ # 验证设备 # @param re_params: # @return: # """ # url = self.appUrl + '/platform/api/device/verify-device' # return requests.get(url, params=re_params, headers=self.business_unify_headers()) # # def query_device_status(self, **re_params): # """ # 查询设备状态 # @param re_params: # @return: # """ # url = self.appUrl + '/platform/api/device/device-status' # return requests.get(url, params=re_params, headers=self.business_unify_headers()) # # def update_device_state(self, **re_data): # """ # 修改设备状态 # @param re_data: # @return: # """ # url = self.appUrl + '/platform/api/device/device-state' # headers = self.business_unify_headers() # headers['content-type'] = 'application/json' # return requests.post(url, data=json.dumps(re_data), headers=headers) # # def query_device_usage_history(self, **re_params): # """ # 查询设备用量历史 # @param re_params: # @return: # """ # url = self.appUrl + '/platform/api/usage/device-usage-history' # return requests.get(url, params=re_params, headers=self.business_unify_headers()) # # def query_current_renew_list_usage_details(self, **re_params): # """ # 查询设备当前队列用量详情 # @param re_params: # @return: # """ # url = self.appUrl + '/platform/api/usage/current-renewlist-usage-details' # return requests.get(url, params=re_params, headers=self.business_unify_headers()) # # def get_device_batch_detail(self, **re_data): # """ # 查询设备当前队列用量详情 # @return: # """ # url = self.appUrl + '/platform/api/device/batch-detail' # headers = self.business_unify_headers() # headers['content-type'] = 'application/json' # return requests.post(url, data=json.dumps(re_data), headers=headers) # # def query_package_list(self, **re_params): # """ # 查询套餐列表 # @return: # """ # url = self.appUrl + '/platform/api/package/list' # return requests.get(url, params=re_params, headers=self.business_unify_headers()) # # def query_renewal_list(self, **re_params): # """ # 续费套餐列表 # @param re_params: # @return: # """ # url = self.appUrl + '/platform/api/package/list' # return requests.get(url, params=re_params, headers=self.business_unify_headers()) # # def async_buy_package(self, **re_data): # """ # 查询设备当前队列用量详情 # @return: # """ # url = self.appUrl + '/platform/api/package/async-buy-package' # headers = self.business_unify_headers() # headers['content-type'] = 'application/json' # return requests.post(url, data=json.dumps(re_data), headers=headers) # # # if __name__ == '__main__': # price = '12.13' # print(float(price)) # discount = '6' # dd = Decimal(price) - Decimal(discount) # print(dd.quantize(Decimal('0.00'))) # # # data = {'foo': 1, 'bar': 2, 'baz': 3} # # print(unicom_api.createSign(**data)) # unicom_api = UnicomObjeect() # # result = unicom_api.generate_token() # # result = unicom_api.refresh_token('5d0c0f30-99bd-4f17-9614-3524495b05d4') # params = {'iccids': '89860620180077842020'} # # response = unicom_api.verify_device(**params) # # response = unicom_api.query_device_status(**params) # # response = unicom_api.update_device_state(**params) # # response = unicom_api.query_device_usage_history(**params) # # response = unicom_api.query_current_renew_list_usage_details(**params) # # unicom_api.get_device_batch_detail() # response = unicom_api.query_package_list() # # response = unicom_api.query_renewal_list(**params) # # if response.status_code == 200: # res_dict = json.loads(response.text) # print(res_dict) # response_json = { # "success": True, # "msg": "操作成功", # "code": 0, # "data": { # "iccid": "8986062018007784202", # "completeIccid": "89860620180077842020", # "status": 1 # } # } # print(response_json['data']['iccid']) # print(response.status_code)