manyi/script/test_swap_api.py

# -*- coding: utf-8 -*-
# !/usr/bin/env python
import datetime
import hashlib
import hmac
import json
import logging
import os
import random

import requests

from base import init_env

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "configs.testing")

init_env(interactive = False)

from apps.web.api.utils import AES_CBC_PKCS5padding_encrypt, AES_CBC_PKCS5padding_decrypt

logger = logging.getLogger(__name__)

# newObj = SwapContract(
#                       source = u'快电',
#                       swapLabel = 'wfl_kd',
#                       operatorType = 'agent',
#                       operatorInnerId = '5b5d87ebaa4fca310e25c5b1',
#                       OperatorID = 'MA4KWBHB5',
#                       OperatorName = u'微付乐',
#                       OperatorTell = '18963981712',
#                       OperatorTel2 = '19063981712',
#                       OperatorRegAddress = u'光谷物联港',
#                       OperatorNote = u'oooo',
#                       northToken = 'waerasasfwr322',
#                       northPort = 'localhost:1883',
#                       
#                       secretFromUs = 'UePNOXPHzb4Jz6cD',
#                       sigSecretFromUs = 'y6J5Qneg7e9tZ2XI',
#                       dataSecretFromUs = 'G6WRJDkZUFW3XSmP',
#                       dataSecretIVFromUs = 'vN2yZWX54Z9zaIZQ',
#                       
#                       operatorId2Us = 'MA005DBW1',
#                       secret2Us = 'UePNOXPHzb4Jz6cD',
#                       sigSecret2Us = 'y6J5Qneg7e9tZ2XI',
#                       dataSecret2Us = 'G6WRJDkZUFW3XSmP',
#                       dataSecretIV2Us = 'vN2yZWX54Z9zaIZQ',
#                       )
# 
# newObj.save()
kdOperId = 'MA005DBW1'
kdSecret = 'UePNOXPHzb4Jz6cD'
sigSecret2Us = 'y6J5Qneg7e9tZ2XI'
dataSecret2Us = 'G6WRJDkZUFW3XSmP'
dataSecretIV2Us = 'vN2yZWX54Z9zaIZQ'
testUrl = 'http://www.washpayer.com' #'http://211.159.224.10' #
def get_sig( data):
    sigSecret = sigSecret2Us
    return hmac.new(sigSecret, data, hashlib.md5).hexdigest().upper()

def get_token():
    url = '%s/api/swap/wfl_kd/query_token' % testUrl
    data = {
        "OperatorID": kdOperId,
        "OperatorSecret": kdSecret
    }
    result = send_request(url, **data)
    ret = result.get("Ret")
    print 'ret is ',ret
    if ret != 0:
        return

    responseJson = result.get("Data")
    print 'get token data :',responseJson
    responseData = json.loads(
        AES_CBC_PKCS5padding_decrypt(s=responseJson, dataSecret=dataSecret2Us, dataSecretIV=dataSecretIV2Us) or "{}"
    )

    # 防止解析出错
    token = responseData.get("AccessToken", "")
    print 'token is :',token
    return token
    
def send_request( url, **kwargs):
    """
    主动发送HTTP请求获取数据 秘钥以及签名
    :param url:
    :param kwargs:
    :return:
    """
    headers = {"Content-Type": "application/json;charset=utf-8"}
    token = kwargs.pop("token", None)
    
    if token: headers.update({"Authorization": "Bearer {}".format(token)})
    timeout = kwargs.pop("timeout", 60)
    # 主动推送 加密以及向量为 dataSecret 和 dataSecretIV
    data = AES_CBC_PKCS5padding_encrypt(
        json.dumps(kwargs), dataSecret=dataSecret2Us, dataSecretIV=dataSecretIV2Us
    )
    
    data = {
        "OperatorID": kdOperId,
        "TimeStamp": datetime.datetime.now().strftime("%Y%m%d%H%M%S"),
        "Seq": "{:0>4}".format(random.randint(1, 1)),
        "Data": data
    }
    
    sig = get_sig(data.get("OperatorID") + data.get("Data") + data.get("TimeStamp") + data.get("Seq"))
    data.update({"Sig": sig})
    
    try:
        response = requests.post(url = url, json = data, headers = headers, timeout = timeout)
    except requests.Timeout:
        return dict()
    except Exception as e:
        return dict()
    
    if response.status_code != 200:
        return dict()
    
    return response.json()

# 测试query stations info
token = get_token()
# LastQueryTime = '2022-02-01 00:00:00'
# pageNo = 1
# pageSize = 10
# result = send_request('%s/api/swap/wfl_kd/query_stations_info' % testUrl,token=token,LastQueryTime=LastQueryTime,pageNo=pageNo,pageSize=pageSize)

LastQueryTime = '2022-02-01 00:00:00'
pageNo = 1
pageSize = 10
result = send_request('%s/api/swap/wfl_kd/query_stations_info' % testUrl,token=token,LastQueryTime=LastQueryTime,pageNo=pageNo,pageSize=pageSize)

# from apps.web.constant import Const
# dev = Device.get_dev_by_l('11112203240001')
# SwapContract.notify_2_all_northers_port_status(dev, 1, Const.DEV_WORK_STATUS_FAULT)

# StationIDs = ['d6d5e1621d073a0f']
# result = send_request('%s/api/swap/wfl_kd/query_station_status'% testUrl,token=token,StationIDs=StationIDs)

# StationID = 'd6d5e1621d073a0f'
# StartTime = '2000-02-01'
# EndTime = '2023-02-01'
# result = send_request('%s/api/swap/wfl_kd/query_station_stats'% testUrl,token=token,StationID=StationID,StartTime=StartTime,EndTime = EndTime)

# EquipAuthSeq = 'd6d5e1621d073a0f'
# ConnectorID = '623c3d54b7c98903d90b4e8e'
# result = send_request('%s/api/swap/wfl_kd/query_equip_auth'% testUrl,token=token,EquipAuthSeq=EquipAuthSeq,ConnectorID=ConnectorID)

# EquipBizSeq = 'd6d5e1621d073a0f'
# ConnectorID = '623c3d54b7c98903d90b4e8e'
# result = send_request('%s/api/swap/wfl_kd/query_equip_business_policy'% testUrl,token=token,EquipBizSeq=EquipBizSeq,ConnectorID=ConnectorID)

# StartChargeSeq = 'd6d5e1621d073a0f2222'
# ConnectorID = '623c3d54b7c98903d90b4e8e'
# QRCode = ''
# result = send_request('%s/api/swap/wfl_kd/query_start_charge'% testUrl,token=token,StartChargeSeq=StartChargeSeq,ConnectorID=ConnectorID,QRCode=QRCode)

# StartChargeSeq = 'd6d5e1621d073a0f2222'
# result = send_request('%s/api/swap/wfl_kd/query_equip_charge_status'% testUrl,token=token,StartChargeSeq=StartChargeSeq)

# StartChargeSeq = 'd6d5e1621d073a0f2222'
# ConnectorID = '623c3d54b7c98903d90b4e8e'
# result = send_request('%s/api/swap/wfl_kd/query_stop_charge'% testUrl,token=token,StartChargeSeq=StartChargeSeq,ConnectorID=ConnectorID)

# dev = Device.get_dev_by_l('11112203240001')
# SwapContract.notify_2_all_northers_port_network_status(dev,DeviceOnlineStatus.DEV_STATUS_OFFLINE)

# SwapContract.notify_2_all_northers_port_status(dev,1,1)

# orderStatus ={
#                 'StartChargeSeq':'123412341234',
#                 'StartChargeSeqStat':4 ,
#                 'ConnectorID':'2323423',
#                 'ConnectorStatus':2,
#                 'CurrentA':2,
#                 'VoltageA':220.0,
#                 'Soc':0,
#                 'StartTime':'2022-02-23 00:00:23',
#                 'EndTime':'2022-02-23 00:00:23',
#                 'TotalPower':3,
#                 'TotalMoney':4,
#                 'ConnectorId':'223232323',
#                 }
#         
# SwapContract.notify_2_all_northers_order_status(dev,orderStatus)

# result = {
#                           'StartChargeSeq':'123412341234',
#                 'StartChargeSeqStat':4 ,
#                           'ConnectorID':'2323423',
#                 'ConnectorStatus':2,
#                           'StartTime':'2022-02-23 00:00:23',
#                           'EndTime':'2022-02-23 00:00:23',
#                           'TotalPower':3,
#                           'TotalElecMoney':4,
#                           'TotalSeviceMoney':3,
#                           'TotalMoney':4,
#                           'StopReason':2
#                           }
# SwapContract.notify_2_all_northers_order_info(dev,result)

print 'OK'