|
@@ -1,146 +0,0 @@
|
|
|
-#!/usr/bin/env python3
|
|
|
-# -*- coding: utf-8 -*-
|
|
|
-import json
|
|
|
-
|
|
|
-
|
|
|
-class CustomUtils():
|
|
|
- def response_analysis(self, response):
|
|
|
- """
|
|
|
- 解析接口返回值:返回值分三种情况如下:
|
|
|
- {"jsonrpc": "2.0","id": 1,"result": true}
|
|
|
- {"jsonrpc": "2.0","id": 1,"error": {"code": 1002,"message": "Permission validation error"}}
|
|
|
- {"jsonrpc": "2.0","id": 1,"result": {"LoginState": 0,"Token": "0126f5d7ef044088b075246c65a410ba","LockRemainingTimes":[{}],"PasswordExpired": false,"AccountName": ""}}
|
|
|
-
|
|
|
- 返回值示例:
|
|
|
- {'resultTag':'','content':'','message':''}
|
|
|
- """
|
|
|
- result = {'resultTag':'','content':'','code':'','message':''}
|
|
|
- try:
|
|
|
- if response.status_code == 200:
|
|
|
- content = str(response.content, encoding = 'UTF-8')
|
|
|
- # 若传入的数据为str类型需要将它转成dict类型
|
|
|
- if isinstance(content, str):
|
|
|
- content = json.loads(content)
|
|
|
- # 若包含error则接口返回失败
|
|
|
- if 'error' in content:
|
|
|
- result['resultTag'] = 'false';
|
|
|
- result['code'] = content['error']['code'];
|
|
|
- result['message'] = content['error']['message'];
|
|
|
- else :
|
|
|
- # 若result为true时情况
|
|
|
- if isinstance(content['result'], dict):
|
|
|
- result['resultTag'] = 'true';
|
|
|
- result['content'] = content['result'];
|
|
|
- else :
|
|
|
- result['resultTag'] = content['result'];
|
|
|
- else:
|
|
|
- raise Exception('Exception:'+str(response.status_code));
|
|
|
- except Exception as ERROR:
|
|
|
- print('Exception:'+str(ERROR));
|
|
|
- raise ERROR;
|
|
|
- return result;
|
|
|
-
|
|
|
- def generate_params(self, method, *args):
|
|
|
-
|
|
|
- """
|
|
|
- 构建body体参数
|
|
|
- {"jsonrpc": "2.0","method": "CommonSignUpAsync","params": [{"AnyAccount": "fnyml","AnyCode": "3245","Password": "222c33333","HeaderMap":{}}],"id": 1}
|
|
|
-
|
|
|
- def loop(request):
|
|
|
- # 元组遍历方式一:for
|
|
|
- intupe_one = ('1', '2', '3', '4')
|
|
|
- for value in tupe_one:
|
|
|
- print(value)
|
|
|
- # 元组遍历方式二:内置函数enumerate()
|
|
|
- tupe_tow = ('1', '2', '3', '4')
|
|
|
- for index, value in enumerate(tupe_tow):
|
|
|
- print("下标:%s 值:%s" % (str(index), value))
|
|
|
- # 元组遍历方式三:使用range()函数
|
|
|
- tupe_three = ('1', '2', '3', '4')
|
|
|
- for index in range(len(tupe_three)):
|
|
|
- print(tupe_three[index])
|
|
|
- # 元组遍历方式四: 使用iter()内置函数,返回迭代器对象
|
|
|
- tupe_four = ('1', '2', '3', '4')for value in iter(tupe_four):
|
|
|
- print(value)
|
|
|
- # 列表遍历方式一:for
|
|
|
- inlist_one = ('1', '2', '3', '4')
|
|
|
- for value in list_one
|
|
|
- :print(value)
|
|
|
- # 列表遍历方式二:内置函数enumerate()
|
|
|
- list_tow = ('1', '2', '3', '4')
|
|
|
- for index, value in enumerate(list_tow):
|
|
|
- print("下标:%s 值:%s" % (str(index), value))
|
|
|
- # 列表遍历方式三:使用range()函数
|
|
|
- list_three = ('1', '2', '3', '4')
|
|
|
- for index in range(len(list_three)):
|
|
|
- print(list_three[index])
|
|
|
- # 列表遍历方式四: 使用iter()内置函数,返回迭代器对象
|
|
|
- list_four = ('1', '2', '3', '4')
|
|
|
- for value in iter(list_four):
|
|
|
- print(value)
|
|
|
- # 字典遍历方式一:for
|
|
|
- indict_one = {"one": 1, "two": 2, "three": 3}
|
|
|
- for value in dict_one:
|
|
|
- print("K:%s Value:%d" % (value, dict_one[value]))
|
|
|
- # 字典遍历方式二:使用dict的keys()方法
|
|
|
- dict_two = {"one": 1, "two": 2, "three": 3}
|
|
|
- for key in dict_two.keys():
|
|
|
- print("K:%s Value:%d" % (key, dict_two[key]))
|
|
|
- # 字典遍历方式三: 使用dict的values()方法
|
|
|
- dict_three = {"one": 1, "two": 2, "three": 3}
|
|
|
- for value in dict_three.values():
|
|
|
- print("value:" + str(value))
|
|
|
- # 字典遍历方式四: 使用dict的items()方法
|
|
|
- dict_four = {"one": 1, "two": 2, "three": 3}
|
|
|
- for item in dict_four.items():
|
|
|
- print(item)
|
|
|
- # 字典遍历方式五: 使用dict的items()方法,然后直接解包元组
|
|
|
- dict_five = {"one": 1, "two": 2, "three": 3}
|
|
|
- for key, value in dict_five.items():
|
|
|
- print(key + ":" + str(value))
|
|
|
- return response_success(message='列表、字典、元组遍历方式验证')
|
|
|
- """
|
|
|
- body = {"jsonrpc": "2.0","method": "","params": [],"id": 1}
|
|
|
- try:
|
|
|
- body['method'] = method;
|
|
|
- for value in args:
|
|
|
- body['params'].append(value);
|
|
|
- except Exception as ERROR:
|
|
|
- print('Exception:'+str(ERROR));
|
|
|
- return body;
|
|
|
-
|
|
|
- def result_analysis(self, result, resultTag, errorCode, message):
|
|
|
- """
|
|
|
- resultTag:返回标记:true/false
|
|
|
- errorCode:错误码
|
|
|
- message:错误内容
|
|
|
- """
|
|
|
- try:
|
|
|
- tag = result['resultTag']
|
|
|
- code = result['code']
|
|
|
- mes = result['message']
|
|
|
- if not (resultTag is None):
|
|
|
- if resultTag is tag:
|
|
|
- return True;
|
|
|
- else:
|
|
|
- raise Exception('Exception:result tag not eq');
|
|
|
- elif not (errorCode is None) and not (message is None):
|
|
|
- if errorCode is code and message is mes:
|
|
|
- return True;
|
|
|
- else:
|
|
|
- raise Exception('Exception:errorCode or message not eq');
|
|
|
- else:
|
|
|
- raise Exception('Exception:params error');
|
|
|
- except Exception as ERROR:
|
|
|
- raise ERROR;
|
|
|
-
|
|
|
-
|
|
|
-if __name__ == '__main__':
|
|
|
- utils = CustomUtils()
|
|
|
- s = {"jsonrpc": "2.0","id": '1',"result": 'true'};
|
|
|
- ss = {"jsonrpc": "2.0","id": 1,"error": {"code": 1002,"message": "Permission validation error"}}
|
|
|
- sss = {"jsonrpc": "2.0","id": 1,"result": {"LoginState": 0,"Token": "0126f5d7ef044088b075246c65a410ba","LockRemainingTimes":[{}],"PasswordExpired": 'false',"AccountName": ""}}
|
|
|
- print(utils.response_analysis(s));
|
|
|
- print(utils.response_analysis(ss));
|
|
|
- print(utils.response_analysis(sss));
|
|
|
- print(utils.generate_params('CommonSignUpAsync',{"AnyAccount": "fnyml","AnyCode": "3245","Password": "222c33333","HeaderMap":{}},{"AnyAccount": "fnyml","AnyCode": "3245","Password": "222c33333","HeaderMap":{}}));
|