#!/usr/bin/env python3 import concurrent.futures import json import pathlib import statistics import time from datetime import datetime from datetime import timedelta import requests ROOT = pathlib.Path(__file__).resolve().parents[1] API_DIR = ROOT / "api" OLD_DATA = ROOT.parent / "20260622-full-retest" / "data" BASE = "http://8.147.119.215:8080" SESSION = requests.Session() SESSION.trust_env = False RUN_ID = datetime.now().strftime("%Y%m%d%H%M%S") def post(path, data=None, token=None, timeout=30): headers = {} if token: headers["token"] = token started = time.time() try: resp = SESSION.post(BASE + path, data=data or {}, headers=headers, timeout=timeout) elapsed = time.time() - started try: body = resp.json() except Exception: body = {"raw": resp.text[:1000]} return {"status": resp.status_code, "elapsed": elapsed, "body": body} except Exception as exc: return {"status": 0, "elapsed": time.time() - started, "body": {"code": 1, "message": repr(exc)}} def load_tsv(path): rows = [] with open(path, encoding="utf-8") as f: for line in f: line = line.rstrip("\n") if not line: continue rows.append(line.split("\t")) return rows def login_token(): body = post("/p/login/login", {"account": "admin", "pwd": "Cpt@1234"})["body"] token = body["data"]["_token"] (API_DIR / "30-login-for-460-retest.json").write_text(json.dumps(body, ensure_ascii=False, indent=2)) return token def equipment_map(token): res = post("/p/equipment/lists", {"type": 10, "page": 1, "pageSize": 100}, token=token) rows = res["body"]["data"]["rows"] selected = {row["code"]: row for row in rows if str(row.get("code", "")).startswith("CSFZX460")} (API_DIR / "31-equipment-460-list-for-retest.json").write_text(json.dumps(res, ensure_ascii=False, indent=2)) return selected def start_heartbeat_userdata(codes): summary = [] first_sample = None started = time.time() for code in codes: start = post("/p/api/equipmentStart", {"code": code}) heartbeat = post("/p/api/heartbeat", {"code": code}) userdata = post("/p/api/userdata", {"code": code, "type": 1}, timeout=60) data_wrapper = userdata["body"].get("data") or {} data = data_wrapper.get("face_data", data_wrapper if isinstance(data_wrapper, list) else []) item = { "code": code, "start_code": start["body"].get("code"), "start_message": start["body"].get("message"), "heartbeat_code": heartbeat["body"].get("code"), "heartbeat_message": heartbeat["body"].get("message"), "userdata_code": userdata["body"].get("code"), "userdata_total": len(data), "test_name_count": sum(1 for x in data if str(x.get("name", "")).startswith("测试人员")), "test_card_count": sum(1 for x in data if str(x.get("card_id", "")).startswith("CSFZX_CARD_")), } summary.append(item) if first_sample is None: first_sample = { "code": code, "sample_count": len(data), "first_5": data[:5], "last_5": data[-5:], } result = {"elapsed_sec": time.time() - started, "summary": summary} (API_DIR / "32-device-start-heartbeat-userdata-summary.json").write_text(json.dumps(result, ensure_ascii=False, indent=2)) (API_DIR / "33-device-userdata-first-sample.json").write_text(json.dumps(first_sample, ensure_ascii=False, indent=2)) return result def push_dishes_60(token, equipments, dishes): results = [] for idx in range(60): code = f"CSFZX460{idx + 1:03d}" equipment = equipments[code] dish = dishes[idx] heartbeat = post("/p/api/heartbeat", {"code": code}) res = post( "/p/equipment/pushDishes", { "equipment_id": equipment["id"], "left_dishes_uuid": dish[0], "right_dishes_uuid": "", }, token=token, ) results.append( { "code": code, "equipment_id": equipment["id"], "dish_uuid": dish[0], "dish_name": dish[1], "heartbeat": heartbeat, "push": res, } ) (API_DIR / "34-heartbeat-and-push-dishes-60-response.json").write_text(json.dumps(results, ensure_ascii=False, indent=2)) return results def meal_flow_one(idx, staff_rows, dishes, plate_prefix="LOAD"): code = f"CSFZX460{((idx - 1) % 60) + 1:03d}" staff_uuid, staff_name, mobile = staff_rows[idx - 1] dish_uuid, dish_name, weight, price = dishes[(idx - 1) % 60] plate = f"CSFZX_{plate_prefix}{RUN_ID[-6:]}_{idx:04d}" bind = post( "/p/api/bindPlate", { "staff_uuid": staff_uuid, "plate_code": plate, "equipment_code": code, "card_id": f"CSFZX_CARD_{idx:04d}", }, timeout=60, ) create_date = (datetime.now() + timedelta(seconds=5)).strftime("%Y-%m-%d %H:%M:%S") dish_info = { "create_date": create_date, "details": [{"uuid": dish_uuid, "weight": float(weight)}], } push = post( "/p/api/zhctPushMeal", { "code": plate, "equipment_code": code, "dish_info": json.dumps(dish_info, ensure_ascii=False), }, timeout=90, ) ok = ( bind["status"] == 200 and push["status"] == 200 and bind["body"].get("code") == 0 and push["body"].get("code") == 0 ) return { "idx": idx, "code": code, "plate": plate, "staff_uuid": staff_uuid, "staff_name": staff_name, "staff_mobile": mobile, "dish_uuid": dish_uuid, "dish_name": dish_name, "bind": bind, "push": push, "ok": ok, "elapsed": bind["elapsed"] + push["elapsed"], } def sample_and_load(staff_rows, dishes): sample = [meal_flow_one(i, staff_rows, dishes, "SM") for i in range(1, 6)] (API_DIR / "35-sample-5-meal-flow.json").write_text(json.dumps(sample, ensure_ascii=False, indent=2)) started = time.time() results = [] with concurrent.futures.ThreadPoolExecutor(max_workers=60) as pool: futures = [pool.submit(meal_flow_one, i, staff_rows, dishes, "LD") for i in range(1, 1501)] for fut in concurrent.futures.as_completed(futures): results.append(fut.result()) elapsed = time.time() - started latencies = [row["elapsed"] for row in results] latencies_sorted = sorted(latencies) summary = { "total": len(results), "ok": sum(1 for row in results if row["ok"]), "fail": sum(1 for row in results if not row["ok"]), "elapsed_sec": elapsed, "throughput_per_sec": len(results) / elapsed if elapsed else 0, "latency_avg": statistics.mean(latencies) if latencies else 0, "latency_p50": statistics.median(latencies) if latencies else 0, "latency_p95": latencies_sorted[int(len(latencies_sorted) * 0.95) - 1] if latencies_sorted else 0, "latency_max": max(latencies) if latencies else 0, } payload = {"summary": summary, "results": results} (API_DIR / "36-load-1500-meal-flow-results-full.json").write_text(json.dumps(payload, ensure_ascii=False, indent=2)) (API_DIR / "36-load-1500-meal-flow-summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2)) return summary def main(): token = login_token() equipments = equipment_map(token) codes = [f"CSFZX460{i:03d}" for i in range(1, 61)] missing = [code for code in codes if code not in equipments] if missing: raise SystemExit(f"missing equipments: {missing}") staff_rows = load_tsv(OLD_DATA / "csfzx_load_staff_1500.tsv") dishes = load_tsv(OLD_DATA / "csfzx_load_dishes_60.tsv") result = { "device_start_heartbeat_userdata": start_heartbeat_userdata(codes), "push_dishes_60": push_dishes_60(token, equipments, dishes), "load_1500": sample_and_load(staff_rows, dishes), } (API_DIR / "37-460-load-retest-overall.json").write_text(json.dumps(result, ensure_ascii=False, indent=2)) print(json.dumps(result["load_1500"], ensure_ascii=False, indent=2)) if __name__ == "__main__": main()