""" 集成压测脚本（带压测报告生成并通过钉钉发送摘要） 说明： - 该脚本基于之前的稳定 worker/队列 实现，运行后会记录每条请求的时间、状态码和延迟。 - 运行结束后会调用Util目录下的压测报告生成器（文件名: stress_test_report_generator.py），输出 HTML/JSON/CSV/（可选）DOCX 等文件。 - 生成后会把关键统计摘要通过 DingTalk 机器人发送（调用 DingTalkHelper.send_message）。 - 安装依赖：aiohttp, tqdm, numpy/pandas/matplotlib/python-docx（可选） - 确保 DingTalkHelper 类在你的 `Util.dingtalk_helper` 中可用，且 ACCESS_TOKEN/SECRET 正确。 """ import sys import os import asyncio import aiohttp import time import traceback import datetime from tqdm import tqdm def get_parent_directory(file_path, levels=1): """获取指定层级的父目录""" path = os.path.abspath(file_path) for _ in range(levels): path = os.path.dirname(path) return path parent_dir = get_parent_directory(__file__, 5) # 获取上五级目录 sys.path.append(parent_dir) from 测试组.脚本.造数脚本2.Util.random_util import RandomUtil from 测试组.脚本.造数脚本2.Util.dingtalk_helper import DingTalkHelper # --- 配置 --- ACCESS_TOKEN = '4625f6690acd9347fae5b3a05af598be63e73d604b933a9b3902425b8f136d4d' SECRET = 'SEC3b6937550bd297b5491855f6f40c2ff1b41bc8c495e118ba9848742b1ddf8f19' apiname = "创建动物房笼架" url = "http://192.168.6.190:5561/api/base/room/shelf/save" headers = { "token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJleHAiOjE3NjYyMTk5MzIsInVzZXJuYW1lIjoiZ2x5In0.fN-yJ0SVrEd_GyIvPmaBL2nx1M3a1jloSOsXyu3E5_w", "Content-Type": "application/json" } NUM_WORKERS = 100 TOTAL_REQUESTS = 40000 MAX_RETRIES = 3 REQUEST_TIMEOUT = 60 OUTPUT_DIR = './load_test_report' # --- 初始化 --- dingtalk_helper = DingTalkHelper(ACCESS_TOKEN, SECRET) LARGE_CONTENT = "压测中" * 500 FILES_PATH = "/userfiles/1588133301094375425/程序附件/notify/notify/2025/10/15/173933/cs.jpg" def create_animal_data(idx: int): random_code = RandomUtil.generate_random_number_string(0, 999999999) random_data = RandomUtil.generate_random_date("2023-01-01", "2025-12-19") return { "id": "", "room": { "id": "2001936175429365762" }, "facility": { "id": "" }, "name": f"hyb压测笼架{random_code}", "code": f"hyb压测笼架{random_code}", "status": "1", "internalPrice": "", "externalPrice": "", "negotiatedPrice": "", "row": 10, "col": 10, "leftRight": "5", "topBottom": "5", "remarks": "", "priceConfig": { "id": "1860888070482100226" }, "variety": {}, "shelfCleanDate": random_data, "otherCleanDate": random_data } async def perform_request(session: aiohttp.ClientSession, index: int, max_retries: int = MAX_RETRIES): attempt = 0 last_err = None while attempt < max_retries: data = create_animal_data(index) start = time.time() try: async with session.post(url, json=data, headers=headers) as resp: text = await resp.text() latency_ms = (time.time() - start) * 1000.0 status = resp.status if status == 200: return { 'index': index, 'timestamp': time.time(), 'status_code': status, 'latency_ms': latency_ms, 'response_size': len(text) if text is not None else None, 'error': None } else: last_err = f'status_{status}:{text}' attempt += 1 await asyncio.sleep(min(10, 2 ** attempt)) except Exception as e: latency_ms = (time.time() - start) * 1000.0 last_err = f'{type(e).__name__}:{str(e)}' attempt += 1 await asyncio.sleep(min(10, 2 ** attempt)) # 最终失败 return { 'index': index, 'timestamp': time.time(), 'status_code': 0, 'latency_ms': latency_ms if 'latency_ms' in locals() else 0, 'response_size': None, 'error': last_err } async def worker(name: int, queue: asyncio.Queue, session: aiohttp.ClientSession, gen, pbar, success_counter: dict, failed_list: list, lock: asyncio.Lock): while True: idx = await queue.get() if idx is None: queue.task_done() break try: res = await perform_request(session, idx) # 记录到报告生成器 gen.record_result( index=res['index'], timestamp=res['timestamp'], status_code=int(res['status_code']), latency_ms=float(res['latency_ms']), response_size=res.get('response_size'), error=res.get('error') ) async with lock: if res['status_code'] and 200 <= res['status_code'] < 300: success_counter['count'] += 1 else: failed_list.append((res['index'], res.get('error'))) pbar.update(1) except Exception as e: async with lock: failed_list.append((idx, f'Worker异常:{type(e).__name__}:{e}')) pbar.update(1) finally: queue.task_done() async def batch_create_animals(total: int, num_workers: int): # 动态加载报告生成器模块（支持中文文件名） gen = None try: import importlib.util script_dir = os.path.dirname(os.path.abspath(__file__)) report_path = os.path.join(script_dir, 'H:\\项目\\archive\\测试组\\脚本\\造数脚本2\\Util\\stress_test_report_generator.py') if os.path.exists(report_path): spec = importlib.util.spec_from_file_location('report_module', report_path) report_module = importlib.util.module_from_spec(spec) spec.loader.exec_module(report_module) LoadTestReportGenerator = getattr(report_module, 'LoadTestReportGenerator') else: # 备用：尝试直接导入模块名（若你的文件名已改为 ascii） from report_generator import LoadTestReportGenerator # type: ignore gen = LoadTestReportGenerator(test_name='压测任务', report_title='压测详细报告') except Exception as e: print('无法加载压测报告生成器，请确认stress_test_report_generator.py 文件位置正确。\n', e) raise timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT) connector = aiohttp.TCPConnector(limit=num_workers, limit_per_host=num_workers, force_close=False) async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session: queue = asyncio.Queue() for i in range(1, total + 1): await queue.put(i) for _ in range(num_workers): await queue.put(None) success_counter = {'count': 0} failed_list = [] lock = asyncio.Lock() with tqdm(total=total, desc='创建进度') as pbar: workers = [ asyncio.create_task(worker(i, queue, session, gen, pbar, success_counter, failed_list, lock)) for i in range(num_workers) ] await asyncio.gather(*workers) # 任务完成，生成报告 os.makedirs(OUTPUT_DIR, exist_ok=True) outputs = gen.generate_report(OUTPUT_DIR, formats=['html', 'json', 'csv', 'docx']) stats = gen.compute_stats() # 构造钉钉摘要消息（中文） now_str = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') msg = [f'【{apiname} 压测报告】', f'生成时间：{now_str}'] msg.append(f"总请求数：{stats.get('total_requests',0)}，成功：{stats.get('success_count',0)}，失败：{stats.get('fail_count',0)}，成功率：{stats.get('success_rate',0):.2%}") msg.append(f"总耗时(s)：{stats.get('duration_seconds',0):.2f}，平均吞吐(req/s)：{stats.get('throughput_rps',0):.2f}") lat = stats.get('latency_ms', {}) msg.append(f"延迟(ms) - 平均：{lat.get('avg',0):.2f}，P90：{lat.get('p90',0):.2f}，P95：{lat.get('p95',0):.2f}，P99：{lat.get('p99',0):.2f}") # 列出生成的报告文件 file_list = [] for k, v in outputs.items(): if k == 'charts': for cname, cpath in v.items(): file_list.append(os.path.abspath(cpath)) else: file_list.append(os.path.abspath(v)) msg.append('生成文件：') for p in file_list: msg.append(p) final_msg = '\n'.join(msg) # 发送钉钉消息 try: dingtalk_helper.send_message(final_msg) except Exception as e: print('发送钉钉消息失败：', e) print('\n[SUMMARY] 已生成报告并发送钉钉摘要。') print('成功数：', success_counter['count'], ' 失败数：', len(failed_list)) if failed_list: print('失败示例（最多显示50条）：') for idx, err in failed_list[:50]: print(f' #{idx} => {err}') if __name__ == '__main__': # 运行前建议先用小规模测试 TOTAL_REQUESTS = TOTAL_REQUESTS NUM_WORKERS = NUM_WORKERS asyncio.run(batch_create_animals(TOTAL_REQUESTS, NUM_WORKERS))