123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285 |
- # -*- coding: utf-8 -*-
- # !/usr/bin/env python
- import base64
- import os
- import time
- import uuid
- import requests
- from django.conf import settings
- os.environ.setdefault("DJANGO_SETTINGS_MODULE", "configs.production")
- from base import init_env, md5
- init_env(interactive = False)
- from apps.web.agent.models import Agent
- from apps.web.dealer.models import Dealer
- # TEST_DEALER_ID = '5b9ae99ad89a177846459999'
- TEST_DEALER_ID = '5d2c23090030483301d8914d'
- class ApiSender(object):
- # PAY_HOST = 'http://211.159.224.10'
- PAY_HOST = 'http://www.washpayer.com'
- def __init__(self, dealer_id, password):
- dealer = Dealer.objects(id = dealer_id).get()
- agent = Agent.objects(id = str(dealer.agentId)).get()
- self.test_case = {
- 'dealerId': dealer_id,
- 'username': str(dealer.username),
- 'password': password,
- 'agentId': dealer.agentId,
- 'domain': agent.domain,
- 'agentSign': agent.agentSign,
- 'mySign': agent.mySign
- }
- def get(self, endpoint, data = None, **kwargs):
- return self._request(
- method = 'get',
- endpoint = endpoint,
- data = data,
- **kwargs)
- def post(self, endpoint, data = None, **kwargs):
- return self._request(
- method = 'post',
- endpoint = endpoint,
- data = data,
- **kwargs
- )
- def _decode_result(self, res):
- try:
- import simplejson as json
- result = json.loads(res.content.decode('utf-8', 'ignore'), strict = False)
- except (TypeError, ValueError):
- return res
- return result
- def _request(self, method, endpoint, data = None, headers = None, **kwargs):
- url = '{base}{endpoint}'.format(
- base = self.PAY_HOST,
- endpoint = endpoint
- )
- headers = {
- 'Content-Type': 'application/json',
- 'authorization': 'Basic {}'.format(
- base64.b64encode('{}:{}'.format(self.test_case['username'], md5(self.test_case['password'])))),
- 'sign': self.test_case['agentSign'],
- 'User-Agent':'MI 6X Build'
- }
- if data:
- data.update({'sign': self.test_case['agentSign']})
- import simplejson as json
- body = json.dumps(data, ensure_ascii = False)
- body = body.encode('utf-8')
- kwargs['data'] = body
- kwargs['timeout'] = kwargs.get('timeout', 15)
- with requests.sessions.Session() as session:
- res = session.request(
- url = url,
- method = method,
- headers = headers,
- **kwargs)
- try:
- res.raise_for_status()
- except requests.RequestException as reqe:
- return {
- 'errcode': 'EXCEPTION',
- 'errmsg': str(reqe)
- }
- return self._decode_result(res)
- def test_online():
- data = {
- 'deviceCode': 'test_zhiyangji_4g'
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/status/online', data = data
- )
- print(res)
- def test_start_device_old():
- data = {
- 'package': '1',
- 'notify_url': 'http://develop.5tao5ai.com/api/test/notify/order',
- 'attachParas':
- {
- 'createTime': '2019-11-27 16:47:12',
- 'coins': '1',
- 'extOrderNo': str(uuid.uuid4()),
- 'devNo': '860344049185789',
- 'domain': '47.108.67.205:8080',
- 'channel': '123456',
- 'agentSign': 'nUiUwUGj8IoiDopmMIQxcMdiUj3wI6Nz',
- 'mySign': 'nUiUwUGj8IoiDopmMIQxcMdiUj3wI6Nz',
- 'logicalCode': 'test_zhiyangji_4g'
- },
- 'deviceCode': 'test_zhiyangji_4g'}
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/start', data = data
- )
- print(res)
- def test_start_device_old2():
- data = {
- 'package': {
- 'unit': u'分钟',
- 'time': 5
- },
- 'notify_url': 'http://develop.5tao5ai.com/api/test/notify/order',
- 'attachParas':
- {
- 'createTime': '2019-11-27 16:47:12',
- 'coins': '1',
- 'extOrderNo': str(uuid.uuid4()),
- 'devNo': '860344049185789',
- 'domain': '47.108.67.205:8080',
- 'channel': '123456',
- 'agentSign': 'nUiUwUGj8IoiDopmMIQxcMdiUj3wI6Nz',
- 'mySign': 'nUiUwUGj8IoiDopmMIQxcMdiUj3wI6Nz',
- 'logicalCode': 'test_zhiyangji_4g'
- },
- 'deviceCode': 'test_zhiyangji_4g'}
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/start', data = data
- )
- print(res)
- def test_start_device_new():
- data = {
- 'package': '1',
- # 'notify_url': 'http://develop.5tao5ai.com/api/test/notify/order',
- 'createTime': '2019-11-27 16:47:12',
- 'extOrderNo': str(uuid.uuid4()),
- 'channel': '123456',
- 'attachParas': {},
- 'deviceCode': 'test_zhiyangji'}
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = 'lcds2017').post(
- endpoint = '/api/v1/device/start', data = data
- )
- print(res)
- def test_start_device_new2():
- data = {
- 'package': {
- 'coins': 3,
- 'time': 5
- },
- # 'notify_url': 'http://develop.5tao5ai.com/api/test/notify/order',
- 'createTime': '2019-11-27 16:47:12',
- 'extOrderNo': str(uuid.uuid4()),
- 'channel': '123456',
- 'attachParas': {},
- 'deviceCode': 'test_zhiyangji_4g'}
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/start', data = data
- )
- print(res)
- def test_get_packages():
- data = {
- 'deviceCode': 'test_sanjiang_4g'
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/packages', data = data
- )
- print(res)
- def test_get_device_list_post():
- data = {
- 'pageSize': 10,
- 'pageIndex': 3
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/list/by/dealer', data = data
- )
- print(res)
- def test_get_device_list_get():
- data = {
- 'pageSize': 10,
- 'pageIndex': 1
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').get(
- endpoint = '/api/v1/device/list/by/dealer?pageSize=10&pageIndex=2')
- print(res)
- def test_get_charger():
- data = {
- 'deviceCode': 'test_zhiyangji'
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/charger', data = data
- )
- print(res)
- def test_get_device_status():
- data = {
- 'deviceCode': 'test_sanjiang_4g'
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/status', data = data
- )
- print(res)
- def test_stop_device():
- data = {
- 'deviceCode': 'test_zhiyangji'
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = 'lcds2017').post(
- endpoint = '/api/v1/device/stop', data = data
- )
- print(res)
- def test_sijiang_start():
- data = {
- 'package': '2',
- 'extOrderNo': str(uuid.uuid4()),
- 'createTime': '2019-11-29 11:26:52',
- 'attachParas': {'chargeIndex': '8'},
- 'channel': 'charge',
- 'Sign': 'a3ohkE846AbtDMuBf7zeeo9VJAUO94It',
- 'deviceCode': 'test_sanjiang_4g'
- }
- res = ApiSender(dealer_id = TEST_DEALER_ID, password = '123456').post(
- endpoint = '/api/v1/device/start', data = data
- )
- print(res)
- # test_online()
- # test_start_device_1()
- test_start_device_new()
- # test_start_device_new2()
- # test_start_device_old()
- # test_start_device_old2()
- # test_get_package/s()
- # test_start_device_2()
- # test_get_device_list_get()
- # test_get_device_list_post()
- # test_get_charger()
- # test_get_device_status()
- # test_stop_device()
- # test_sijiang_start()
|