2019-09-27

實現公司提出的需求，寫了以下的腳本，模擬多車輛同時查詢、入場、出場一套流程

'''

@Description: In User Settings Edit

@Author: coco

@Date: 2019-09-24 10:00:03

@LastEditTime: 2019-09-27 11:33:43

@LastEditors: coco

'''

import random

import threading

import time

import requests

import logging

import redis

pool = redis.ConnectionPool(host='localhost', port=6379)

redis_conn = redis.Redis(connection_pool=pool)

logger = logging.getLogger()

logger.setLevel(logging.INFO)

rq = time.strftime('%Y%m%d%H%M', time.localtime(time.time()))

log_path = './'

log_name = 'test4.log'

logfile = log_name

fh = logging.FileHandler(logfile, mode='w', encoding='utf8')

fh.setLevel(logging.DEBUG)

formatter = logging.Formatter(

    "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")

fh.setFormatter(formatter)

logger.addHandler(fh)

carNos = ['OO55', 'BY334', 'DD088', 'RR3456', 'TY2322',

          'QQ334', 'UI334', 'PY453', 'GY778', 'RTG34',

          'YU439', 'YR323', 'ER003', 'P001', 'GG01',

          'GG011', 'P002', 'TTR45', 'IO887', 'YU667',

          'PP667', 'RV565', 'AQT89', 'AS562', 'PH565',

          'POL78', 'QE55', 'RI675', 'WW888', 'YO865',

          'YUJ66', 'ZT666']

def send_query(carNo):

    try:

        url = 'http://xx.xx.xx.xx:xxxx/api?app_id=bd6832c144394620&server_name=Parking.NoSense.Pay&method=Parking.NoSense.Pay.ApiProxy.CarPlateSignContractQuery&timestamp=20190916181649&nonce=1805027239&signature=5f8e18d9bb584203a0a5f874a3b29cae&parameters={"parking_id":"2","channel_id":"1","plate_no":"'+carNo+'","vehicle_type":"1"}'

        r = requests.get(url, timeout=5)

        data = r.text

        response_time = r.elapsed.total_seconds()*1000

        print('查詢:'+data)

        logger.info({"Action": "查詢", "Response": data,

                     "Url": url, "ResponseTime": response_time})

    except Exception as err:

        print(err)

        logger.error(err)

def send_enter(carNo):

    try:

        url = 'http://xx.xx.xx.xx:xxxx/api?app_id=bd6832c144394620&server_name=Parking.NoSense.Pay&method=Parking.NoSense.Pay.ApiProxy.CarEnterParking&timestamp=20190916181649&nonce=1805027239&signature=5f8e18d9bb584203a0a5f874a3b29cae&parameters={"parking_id":"2","channel_id":"1","plate_no":"' + carNo+'","vehicle_type":"1","in_time":"20190917151355"}'

        r = requests.get(url, timeout=5)

        data = r.text

        response_time = r.elapsed.total_seconds()*1000

        print('入場:'+data)

        logger.info({"Action": "入場", "Response": data,

                     "Url": url, "ResponseTime": response_time})

    except Exception as err:

        print(err)

        logger.error(err)

def send_out(carNo):

    try:

        url = 'http://xx.xx.xx.xx:xxxx/api?app_id=bd6832c144394620&server_name=Parking.NoSense.Pay&method=Parking.NoSense.Pay.ApiProxy.CarOutParking&timestamp=20190916181649&nonce=1805027239&signature=5f8e18d9bb584203a0a5f874a3b29cae&parameters={"parking_id":"2","channel_id":"1","out_trade_number":"10201909170910119","subject":"無感","total_amt":"0","plate_no":"' + \

            carNo+'","vehicle_type":"1","in_time":"20190917151355","out_time":"20190917154556","asyn_url":"","must_charge":"1"}'

        r = requests.get(url, timeout=5)

        data = r.text

        response_time = r.elapsed.total_seconds()*1000

        print('出場:'+data)

        logger.info({"Action": "出場", "Response": data,

                     "Url": url, "ResponseTime": response_time})

    except Exception as err:

        print(err)

        logger.error(err)

def exe_query(carNo):

    t1 = threading.Thread(target=send_query, args=(carNo,))

    t1.start()

def exe_enter(carNo):

    t2 = threading.Thread(target=send_enter, args=(carNo,))

    t2.start()

def exe_out(carNo):

    t3 = threading.Thread(target=send_out, args=(carNo,))

    t3.start()

def test1():

    i = 0

    for i in range(len(carNos)):

        exe_query(carNos[i])

def test2():

    j = 0

    for j in range(len(carNos)):

        exe_enter(carNos[j])

def test3():

    k = 0

    for k in range(len(carNos)):

        exe_out(carNos[k])

def executer():

    time.sleep(5)

    test1()

    time.sleep(5)

    test2()

    time.sleep(15)

    test3()

def lanch():

    i = 0

    counter = 10000

    while i < counter:

        executer()

        i += 1

lanch()