著名建站公司西安百度公司开户

需求：

1. 调用 /api/GDPostFeiYouTemu/NotMailTemuOrderPush
2. 生成一个 随机数 赋值给orderCode和referenceNumber（且需要唯一）
3. ECB 加密 Json 参数
4. 可配置的 持续调用时间和调用次数
5. 生成测试报告，并保存在文件夹中

代码：

import json
import random
import string
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from base64 import b64encode
import requests
from concurrent.futures import ThreadPoolExecutor
import time
import logging
from datetime import datetime# 配置日志
logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s',handlers=[logging.FileHandler('request_log.log'),logging.StreamHandler()]
)# 加密函数（保持不变）
def encrypt_data(data, key):key = key.encode('utf-8')[:16]  # AES-128cipher = AES.new(key, AES.MODE_ECB)ct_bytes = cipher.encrypt(pad(data.encode('utf-8'), AES.block_size))return b64encode(ct_bytes).decode('utf-8')# 生成唯一随机字符串（添加时间戳增强唯一性）
def generate_unique_random_string(length=10):timestamp = datetime.now().strftime("%f")  # 微秒级时间戳chars = ''.join(random.choices(string.ascii_uppercase + string.digits, k=length))return f"{timestamp}_{chars}"# 构造请求数据（优化唯一性）
def prepare_request_data():base_data = {"head": {"method": "batch_create_order","version": "1.0.0","requestTime": "2022-07-20 10:06:31"},"uaccount": "xxxxx","body": [{"mailCode": "TEST201072000MAILCODE","orderCode": "TE_" + generate_unique_random_string(8),"length": 20.3,"width": 18.1,"height": 6.3,"weight": 0.53,"receiverCompName": "AMZ-tt","receiverName": "Kaila Gilam 123456","receiverProv": "Atlantico","receiverCity": "Barranquilla","receiverArea": "Fayetteville","receiverHouseNo": "15","receiverAddress": "138 Lowery Dr 138 Lowery Dr","receiverAddress2": "apto 2B","receiverMobile": "","receiverTel": "3013357358","receiverEmail": "[e-mail]","receiverPostCode": "110121","senderCompName": "CAN109","senderName": "CAN101","senderEngProv": "guangdong","senderEngCity": "guangzhou","senderAddress": "baiyun helong","senderPostCode": "80000","senderTel": "123456456","senderMobile": "147258369","countryCode": "CO","countryName": "哥伦比亚","productCode": "COL-X","tariffType": "ddp","vatNumber": "5236523","taxNo": "3923982135","remark": "","referenceNumber": "BG-" + generate_unique_random_string(8),"declValue": "50","declCurrency": "RMB","cardType": "2","cardNo": "48452153625425361252","inner": [{"gName": "加厚压缩袋","gEngName": "compression bag","innerContext": "","innerQty": 5,"innerWeight": 0.22,"innerPrice": 3.5,"currencyCode": "USD","madePlace": "CN","hscode": "3923290000","model": "无","gBrand": "无","originCountry": "","remark": "","sku": "100010","material": "塑料","purpose": "物品","taxNo": "3923","salesAddress": "https://suijimimashengcheng.bmcx.com/","distribution": "compression","electricityFlag": "","unitCode": "11","secUnitCode": "","innerDeclValue": "25","innerDeclCurrency": "RMB"}]}]}return base_data# 发送请求（添加重试机制和日志）
def send_request(url, key):try:data = prepare_request_data()encrypted = encrypt_data(json.dumps(data), key)start_time = time.time()response = requests.post(url, json={"data": encrypted}, timeout=10)duration = time.time() - start_timelog_msg = f"Status: {response.status_code} | Duration: {duration:.2f}s | Order: {data['body'][0]['orderCode']}"if response.status_code != 200:logging.error(log_msg)return (False, response.status_code)else:logging.info(log_msg)return (True, response.status_code)except Exception as e:logging.error(f"Request failed: {str(e)}")return (False, str(e))# 压力测试主函数
def stress_test():# 配置参数config = {"url": "http://txxxxxx.xxxx.com/api/GDPostFeiYouTemu/NotPush","key": "bJtR4ZSNK4p","threads_per_second": 1,"duration_seconds": 1}success_count = 0failure_count = 0total_requests = config["threads_per_second"] * config["duration_seconds"]logging.info(f"Starting stress test: {total_requests} total requests")with ThreadPoolExecutor(max_workers=50) as executor:  # 稍微多些工作线程防止堆积futures = []start_time = time.time()# 创建每分钟任务for _ in range(config["duration_seconds"]):# 提交每秒的任务批for _ in range(config["threads_per_second"]):future = executor.submit(send_request,config["url"],config["key"])futures.append(future)# 精确控制时间间隔time.sleep(1 - (time.time() - start_time) % 1)# 统计结果for future in futures:result = future.result()if result[0]:success_count += 1else:failure_count += 1# 生成测试报告end_time = time.time()test_duration = end_time - start_timereport = f"""======== Stress Test Report ========Total Requests: {total_requests}Successful: {success_count}Failed: {failure_count}Success Rate: {(success_count / total_requests) * 100:.2f}%Test Duration: {test_duration:.2f} secondsAvg Throughput: {total_requests / test_duration:.2f} req/s===================================="""logging.info(report)if __name__ == "__main__":stress_test()

运行结果：

D:\PYTHON-学习\邮政接口压力测试脚本\pythonProject1\.venv\Scripts\python.exe D:\PYTHON-学习\邮政接口压力测试脚本\pythonProject1\优化非邮下单接口.py 
2025-03-11 16:27:35,452 - INFO - Starting stress test: 1 total requests
2025-03-11 16:27:37,474 - INFO - Status: 200 | Duration: 2.02s | Order: TE_452975_72S1AYET
2025-03-11 16:27:37,475 - INFO - ======== Stress Test Report ========Total Requests: 1Successful: 1Failed: 0Success Rate: 100.00%Test Duration: 2.02 secondsAvg Throughput: 0.49 req/s====================================进程已结束，退出代码为 0