manyi/script/test_swap_api.py

# -*- coding: utf-8 -*-
# !/usr/bin/env python
import datetime
import os
import requests
import random
import logging
import hmac
import hashlib
import json
import itsdangerous
from concurrent.futures import ThreadPoolExecutor


from base import init_env

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

init_env(interactive = False)
from apps.thirdparties.dingding import DingDingRobot
from apps.web.device.models import Device
from apps import serviceCache
from django.conf import settings
from apps.web.constant import RechargeRecordVia
from apps.web.core.db import copy_document_classes
from apps.web.dealer.proxy import DealerIncomeProxy
from apilib.monetary import RMB
from apps.web.user.models import RechargeRecord
from bson import ObjectId
from apps.web.common.models import WithdrawRecord, WithdrawRefundRecord
from apps.web.dealer.models import Dealer
from apps.web.core.bridge.wechat.WechatClientProxy import MyWeChatComponent

from apps.web.south_intf.swap_carcharger import SwapContract
from apps.web.api.utils import AES_CBC_PKCS5padding_encrypt, AES_CBC_PKCS5padding_decrypt
from apps.web.constant import DeviceOnlineStatus

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'