test(scripts): 补测"没测过的四项"+ 清理工具扩展（F-99 / F-5）

一、load-test.py：把 docs/09 §9.5 里四项"没测过"的变成"测过"
   · --base：另起一个 --fill.storage.provider=local 的实例（18080）量上传，
     不把测试文件灌进真实七牛桶；
   · --fixture-only：量名册导入本身（造 Excel 与导入接口分开计时 ——
     脚本自己的成本不能算到系统头上）；
   · --only export-big：造 N 份已通过的单（门禁绑定→草稿→上传→提交→
     初审（班主任）→终审（管理员）→前置检查→创建→轮询），再逐卷解开验证 ZIP；
   · --soak/--rps：长跑，每分钟采 RSS（从系统取，不是应用自报的堆）与
     PG 连接数，比较前 1/3 与后 1/3 的均值 —— 判据是趋势，不是"跑完没崩"；
   · 上传夹具换成真 JPEG（原来的"JPEG 头+随机字节"在服务端解码即失败，
     量到的根本不是压缩成本）。

   数字（docs/09 §9.5 有完整复现命令）：
     上传 20 并发 40/40，串行 321ms（2850KB→1329KB，47%），并发 p50 620ms；
     5000 行导入 20.2s（RSS 782MB）；150 人 194.6MB 导出打包 13.2s，
     分卷 4 卷每卷可单独解开；长跑 51 样本 15300 请求 0 失败，
     RSS 603.9→582.9MB 平稳，PG 连接定时任务高峰后正常回落。

二、F-99：这一轮压测工具自己撒了 8 次谎（都是"度量/夹具出问题时
   现象长得像产品缺陷"）：前置未断言（--only upload 单跑全报"没有草稿"）、
   假 JPEG、猜列名（part_no/file_name/file_path）、拿 HTTP 码当业务码、
   跨租户同名 admin（库里 50 多个，改从 token 的 sub 取 uid）、
   夹具接口不查返回码（PUT orgs 被拒→任务没发布→学生端整齐报 40008）、
   PowerShell -First 截管道把"被堵"变成"卡住"、清理只删库行不删文件。
   修法：bind_users/call_ok（夹具每个调用都断言业务码）、make_img 真图、
   列名按 DDL、cleanup 连磁盘文件一起删。

三、dev-clean-e2e.py（F-5 落地）：
   · --orphan-files：清"库里没有任何一行指向"的磁盘文件（判据来自
     collect_file.std_path/raw_path + collect_export_file.file_path）——
     实测 uploads 里 281 个文件/276MB 而库里只剩 16 行，已清 255 个/275.4MB；
   · --demo-tasks：清【演示·旧】任务（只匹配带「旧」的 ——
     【演示】审核台是当前演示态在用的，原来那句 LIKE 会把它一起删掉）；
   · e2e 租户的业务行按 information_schema 动态枚举（写死清单必然漏新表）；
   · --only-orphan-files：最安全的一档，只碰磁盘不碰任何库行。
   ⚠️ e2e 租户与"E2E 24数控1班"组织刻意未删：当前演示态正挂在其中一个班下面。

验证：python scripts/verify-all.py --quick → 通过 5 / 失败 0 / 跳过 0；
      docs/09 §9.1~§9.5 已同步（含机器规格：12 物理核 / 59.6GB）。

scripts/dev-clean-e2e.py

@@ -1,31 +1,47 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 """
-清理 e2e-m0.py 产生的测试数据。
+开发库与本地磁盘的**测试残留**清理。
 
 E2E 每跑一次会：
   1. 新建一个租户（编码 e2e######）并给它建角色与账号；
   2. **在 demo 租户下**建三个组织（学校/院系/班级，编码带时间戳）。
 
-第 2 点容易被忽略——这些组织挂在正常租户下，不清掉会让组织树里
+第 2 点容易被忽略——这些组织挂在正常租户下，不清掉会让组织树里堆满
-堆满 "E2E 示范学校 586923" 这种名字。
+"E2E 示范学校 586923" 这种名字。
+
+跑几十轮之后真正的问题不是"列表里有测试名字"，而是两笔越滚越大的账：
+
+  · **租户垃圾**：`sys_tenant` 里 48 个 `e2e*`（每个还带着用户、角色、组织）；
+  · **磁盘垃圾**：`backend/data/uploads` 里几百 MB 的文件 ——
+    **删库不删盘**。压测与 E2E 会删掉自己的库行，但文件留在磁盘上，
+    而"哪些文件还有主"只有库知道（`collect_file.std_path/raw_path`、
+    `collect_export_file.file_path`）。实测：库里 16 行文件、磁盘上 281 个文件。
 
 用法
 ----
-    python scripts/dev-clean-e2e.py            # 只看会删什么（默认 dry-run）
+    python scripts/dev-clean-e2e.py                  # 只看会删什么（默认 dry-run）
-    python scripts/dev-clean-e2e.py --apply    # 真删
+    python scripts/dev-clean-e2e.py --apply          # 真删：e2e 租户 + 测试组织 + 它们的业务行
+    python scripts/dev-clean-e2e.py --apply --demo-tasks      # 连【演示·旧】任务一起删
+    python scripts/dev-clean-e2e.py --apply --orphan-files    # 连"没有库行指向"的磁盘文件一起删
 
 ⚠️ 只在开发库上跑，不要对生产执行。
+⚠️ `--orphan-files` 删的是**磁盘文件**，不可恢复；它靠"库里没有任何一行指向它"来判定，
+   所以只要库是完整的，它就不会删到在用的文件。默认 dry-run，先看一眼再 `--apply`。
 """
 from __future__ import annotations
 
 import argparse
+import os
 import subprocess
 import sys
 
 if hasattr(sys.stdout, "reconfigure"):
     sys.stdout.reconfigure(encoding="utf-8")
 
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+UPLOADS = os.path.join(ROOT, "backend", "data", "uploads")
+
 # 子表在前、主表在后（无外键约束，顺序只为可读性与排查方便）
 TENANT_CHILD_TABLES = [
     ("角色授权", "sys_role_permission"),
@@ -45,6 +61,13 @@ DEMO_ORG_FILTER = (
     "OR org_name LIKE 'E2E %' OR org_name LIKE '重复编码%' OR org_name LIKE '层级颠倒%'"
 )
 
+# 演示残留：给客户演示时反复摆过的任务，为了不污染演示态，已发布的那批被**改名挪开**
+# 而不是删掉（F-14：已发布的任务删不掉）。它们属于"当时的演示数据"，不是需求数据。
+#
+# ⚠️ 只匹配带「旧」的那一批：`【演示】审核台` 是**当前**演示态在用的任务，
+#    删了它 = 把演示态弄坏（而且它的班级就是下面那批 "E2E 24数控1班"，同理不能删）。
+DEMO_TASK_FILTER = "task_name LIKE '【演示·旧】%'"
+
 
 class Db:
     def __init__(self, container: str, user: str, name: str):
@@ -62,25 +85,86 @@ class Db:
         return [x for x in r.stdout.split("\n") if x.strip()]
 
 
+def tables_with(db: Db, column: str) -> list[str]:
+    """
+    所有带这一列的 `collect_*` 表。
+
+    ⚠️ **动态枚举，不写死清单**：写死的话，以后新增一张业务表就会漏清，
+    而"漏清"的表现是下一次跑 E2E 时某个统计数字莫名偏大 —— 排查方向完全错。
+    """
+    sql = ("SELECT c.table_name FROM information_schema.columns c "
+           "JOIN information_schema.tables t "
+           "  ON t.table_name = c.table_name AND t.table_schema = c.table_schema "
+           "WHERE c.table_schema = 'public' AND t.table_type = 'BASE TABLE' "
+           f"  AND c.table_name LIKE 'collect\\_%' AND c.column_name = '{column}' "
+           "ORDER BY c.table_name")
+    return [r.split("|")[0] for r in db.rows(sql)]
+
+
+def db_file_paths(db: Db) -> set[str]:
+    """库里所有被引用的文件路径（相对 uploads 目录）。"""
+    known: set[str] = set()
+    for table, cols in (("collect_file", ("raw_path", "std_path")),
+                        ("collect_export_file", ("file_path",))):
+        for col in cols:
+            for r in db.rows(f"SELECT {col} FROM {table} WHERE {col} IS NOT NULL"):
+                p = r.split("|")[0].strip().replace("\\", "/")
+                if not p:
+                    continue
+                known.add(p)
+                if os.path.isabs(p):
+                    known.add(os.path.relpath(p, UPLOADS).replace("\\", "/"))
+    return known
+
+
+def scan_orphans(db: Db) -> tuple[list[str], int]:
+    """磁盘上有、而库里没有任何一行指向的文件。"""
+    known = db_file_paths(db)
+    orphans, size = [], 0
+    for dirpath, _dirs, files in os.walk(UPLOADS):
+        for fn in files:
+            full = os.path.join(dirpath, fn)
+            rel = os.path.relpath(full, UPLOADS).replace("\\", "/")
+            if rel not in known:
+                orphans.append(rel)
+                try:
+                    size += os.path.getsize(full)
+                except OSError:
+                    pass
+    return sorted(orphans), size
+
+
 def main() -> int:
     ap = argparse.ArgumentParser()
     ap.add_argument("--apply", action="store_true", help="真正执行删除（默认只预览）")
+    ap.add_argument("--demo-tasks", action="store_true",
+                    help="连【演示·旧】任务（及其名册/材料/填报单/文件）一起删")
+    ap.add_argument("--orphan-files", action="store_true",
+                    help="连磁盘上「没有库行指向」的文件一起删（不可恢复）")
+    ap.add_argument("--only-orphan-files", action="store_true",
+                    help="**只**清磁盘孤儿文件，不碰租户/组织/任务（最安全的一档）")
     ap.add_argument("--container", default="xt-pg")
     ap.add_argument("--user", default="fill")
     ap.add_argument("--db", default="filldb")
     args = ap.parse_args()
+    if args.only_orphan_files:
+        args.orphan_files = True
 
     db = Db(args.container, args.user, args.db)
 
-    tenants = db.rows(
+    tenants = [] if args.only_orphan_files else db.rows(
         f"SELECT id || '|' || tenant_code || '|' || tenant_name "
         f"FROM sys_tenant WHERE {TENANT_FILTER} ORDER BY id")
-    demo_orgs = db.rows(
+    demo_orgs = [] if args.only_orphan_files else db.rows(
         f"SELECT id || '|' || org_name FROM sys_org "
         f"WHERE deleted = 0 AND ({DEMO_ORG_FILTER}) ORDER BY id")
+    demo_tasks = (db.rows(f"SELECT id || '|' || task_name FROM collect_task "
+                          f"WHERE deleted = 0 AND {DEMO_TASK_FILTER} ORDER BY id")
+                  if (args.demo_tasks and not args.only_orphan_files) else [])
+    orphans, orphan_size = scan_orphans(db) if args.orphan_files else ([], 0)
 
-    if not tenants and not demo_orgs:
+    if not any((tenants, demo_orgs, demo_tasks, orphans)):
-        print("没有 e2e 测试数据，无需清理")
+        print("没有需要清理的测试数据")
         return 0
 
     if tenants:
@@ -99,6 +183,25 @@ def main() -> int:
         if len(demo_orgs) > 15:
             print(f"   ... 另有 {len(demo_orgs) - 15} 个")
 
+    if demo_tasks:
+        print(f"\n③ 演示残留任务 {len(demo_tasks)} 个（--demo-tasks）：")
+        for r in demo_tasks[:10]:
+            p = r.split("|")
+            print("   id=%-5s %s" % (p[0], p[1] if len(p) > 1 else ""))
+        if len(demo_tasks) > 10:
+            print(f"   ... 另有 {len(demo_tasks) - 10} 个")
+
+    if args.orphan_files:
+        print(f"\n④ 磁盘上没有库行指向的文件 {len(orphans)} 个 / {orphan_size / 1048576.0:.1f} MB"
+              f"（--orphan-files）")
+        by_dir: dict[str, int] = {}
+        for rel in orphans:
+            by_dir[rel.split("/")[0]] = by_dir.get(rel.split("/")[0], 0) + 1
+        for k, v in sorted(by_dir.items(), key=lambda x: -x[1]):
+            print("   %-10s %d 个" % (k, v))
+        for rel in orphans[:5]:
+            print("   例：%s" % rel)
+
     if not args.apply:
         print("\n这是预览。加 --apply 才会真正删除。")
         return 0
@@ -106,20 +209,56 @@ def main() -> int:
     print("\n开始清理：")
     total = 0
 
-    for label, table in TENANT_CHILD_TABLES:
+    if not args.only_orphan_files:
-        n = len(db.rows(f"DELETE FROM {table} WHERE {SUB_FILTER} RETURNING id"))
+        for label, table in TENANT_CHILD_TABLES:
+            n = len(db.rows(f"DELETE FROM {table} WHERE {SUB_FILTER} RETURNING id"))
+            total += n
+            print("   %-10s %-22s 删除 %d 行" % (label, table, n))
+
+        # e2e 租户的**业务行**：不删的话，租户没了、任务还在，
+        # 于是"每个请求多一次租户解析"的旧账会一直留着（F-5）
+        for table in tables_with(db, "tenant_id"):
+            n = len(db.rows(f"DELETE FROM {table} WHERE tenant_id IN "
+                            f"(SELECT id FROM sys_tenant WHERE {TENANT_FILTER}) RETURNING id"))
+            if n:
+                total += n
+                print("   %-10s %-22s 删除 %d 行" % ("业务行", table, n))
+
+        n = len(db.rows(f"DELETE FROM sys_tenant WHERE {TENANT_FILTER} RETURNING id"))
         total += n
-        print("   %-10s %-22s 删除 %d 行" % (label, table, n))
+        print("   %-10s %-22s 删除 %d 行" % ("租户", "sys_tenant", n))
 
-    n = len(db.rows(f"DELETE FROM sys_tenant WHERE {TENANT_FILTER} RETURNING id"))
+        n = len(db.rows(f"DELETE FROM sys_org WHERE {DEMO_ORG_FILTER} RETURNING id"))
-    total += n
+        total += n
-    print("   %-10s %-22s 删除 %d 行" % ("租户", "sys_tenant", n))
+        print("   %-10s %-22s 删除 %d 行" % ("测试组织", "sys_org", n))
 
-    n = len(db.rows(f"DELETE FROM sys_org WHERE {DEMO_ORG_FILTER} RETURNING id"))
+        if args.demo_tasks:
-    total += n
+            # 先按任务 id 删子表（同样动态枚举"带 task_id 的表"）
-    print("   %-10s %-22s 删除 %d 行" % ("测试组织", "sys_org", n))
+            for table in tables_with(db, "task_id"):
+                n = len(db.rows(f"DELETE FROM {table} WHERE task_id IN "
+                                f"(SELECT id FROM collect_task WHERE {DEMO_TASK_FILTER}) RETURNING id"))
+                if n:
+                    total += n
+                    print("   %-10s %-22s 删除 %d 行" % ("演示子表", table, n))
+            n = len(db.rows(f"DELETE FROM collect_task WHERE {DEMO_TASK_FILTER} RETURNING id"))
+            total += n
+            print("   %-10s %-22s 删除 %d 行" % ("演示任务", "collect_task", n))
 
-    print(f"\n完成，共删除 {total} 行。")
+    removed = 0
+    if args.orphan_files:
+        # ⚠️ 删文件前**重算一次**：上面刚删过库行，此时"还有主"的文件集合变小了，
+        #    但那些刚被删掉主的文件属于"任务数据"，本该跟着一起清 —— 重算正好覆盖它们
+        orphans, orphan_size = scan_orphans(db)
+        for rel in orphans:
+            try:
+                os.remove(os.path.join(UPLOADS, rel))
+                removed += 1
+            except OSError:
+                pass
+        print("   %-10s %-22s 删除 %d 个文件 / %.1f MB"
+              % ("磁盘文件", "backend/data/uploads", removed, orphan_size / 1048576.0))
+
+    print(f"\n完成，共删除 {total} 行" + (f"、{removed} 个文件。" if args.orphan_files else "。"))
     return 0
 
 

scripts/load-test.py

@@ -44,7 +44,13 @@ import uuid
 sys.stdout.reconfigure(encoding="utf-8")
 
 ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+# 默认打 dev 后端；`--base` 可改（例如为了量上传而在 18080 另起一个
+# `--fill.storage.provider=local` 的实例，避免把测试文件灌进真实对象存储）
 BASE = "http://localhost:8080"
+# 演示班主任（scripts/dev-seed-audit.py 摆出来的）—— 一期初审恒为开启，
+# 所以"审批闭环"这类场景必须有它
+TEACHER_USER = "demot1"
+TEACHER_PWD = "Demo1234"
 SUFFIX = uuid.uuid4().hex[:6]
 TMP_ROSTER = os.path.join(ROOT, "scripts", f"_lt-roster-{SUFFIX}.xlsx")
 TMP_IMG = os.path.join(ROOT, "scripts", f"_lt-img-{SUFFIX}.jpg")
@@ -141,7 +147,28 @@ def timed(fn):
 
 # ══════════════ 夹具 ══════════════
 
-def build_fixture(admin):
+def uid_of(token):
+    """
+    从自己的 token 里取 uid（`sub`）。
+
+    ⚠️ **不要按 username 查库**：库里同一个 username 在每个租户都有一行
+    （`admin` 有 50 多个，`select id from sys_user where username='admin'` 会随机命中
+    别人家的账号）。我第一版就是这么写的，于是把"初审人"设成了一个跨租户的用户 id。
+    """
+    p = token.split(".")[1]
+    p += "=" * (-len(p) % 4)
+    return int(json.loads(base64.urlsafe_b64decode(p))["sub"])
+
+
+def call_ok(method, path, token, body=None, what="", timeout=60):
+    """打一个接口并**断言业务码为 0**（不查返回码的调用是 F-96 里最贵的那一个）。"""
+    c, b = call(method, path, token, body, timeout=timeout)
+    if not (isinstance(b, dict) and b.get("code") == 0):
+        raise SystemExit("夹具失败：%s → %s" % (what or path, json.dumps(b, ensure_ascii=False)[:200]))
+    return b.get("data")
+
+
+def build_fixture(admin, teacher):
     """造一个任务 + N 个名册人（每人一个已知证件号）+ 对应 uid 的 MP token。"""
     _, orgs = call("GET", "/api/v1/orgs/tree", admin)
     classes = []
@@ -161,17 +188,32 @@ def build_fixture(admin):
         "taskName": f"压测任务 {SUFFIX}", "batchNo": "202609", "majorName": "数控加工",
         "collectMode": 1, "certType": 1,
         "startTime": "2026-01-01", "endTime": "2099-12-31",
+        # ⚠️ **只走终审**（needClassAudit=false）是刻意的：初审的判据是"这份单在你负责的
+        #    班级里"（AuditService.auditDenyReason），而夹具的名册导进的是组织树里第一个班，
+        #    admin 并不是那个班的班主任 —— 于是逐份通过会整齐地报 40301「不在你负责的班级里」。
+        #    要按真实两段审核走，就得再造一个"带班级范围的班主任"账号；
+        #    而导出这条路与"谁审的"无关，所以这里选便宜且不引入额外夹具的那条。
+        "auditConfig": {"needClassAudit": False, "needAdminAudit": True},
     })
     tid = b["data"]["id"]
     TASK_IDS.append(tid)
 
     call("POST", f"/api/v1/tasks/{tid}/materials", admin, {
         "materialCode": "IDCARD", "materialName": "身份证", "required": True,
-        "fileTypes": ["jpg", "jpeg", "png"], "maxSizeKb": 2048, "maxCount": 3,
+        "fileTypes": ["jpg", "jpeg", "png"], "maxSizeKb": 4096, "maxCount": 3,
         "minSizeKb": 1, "minLongSide": 100,
     })
-    call("PUT", f"/api/v1/tasks/{tid}/orgs", admin,
+    # ⚠️ **必须指定 headUserId**：初审的判据是"这个班归不归我审"
-         {"orgs": [{"orgId": org["id"], "expectedCount": GATE_USERS}]})
+    #    （`AuditScopeMapper.countMyOrg` 三级回退：head_user_id → deputy → 班级负责人），
+    #    而夹具的名册导进的是组织树里第一个班 —— 不指定的话，逐份通过会整齐地报
+    #    40301「这份填报单不在你负责的班级里」。
+    #    也**不能**靠"关掉初审"绕过：一期 `withAuditDefaults` 把 needClassAudit 强制为 true
+    #    （契约 §3.2「社会收集恒为 false，一期不启用社会收集」），传 false 会被扳回来。
+    #    指定的人还得**真有班主任角色**，否则这次 PUT 会被服务端拒（我第一版就是被它拒的，
+    #    而当时没查返回码，表现成很远的地方"任务已截止"—— 40008）。
+    call_ok("PUT", f"/api/v1/tasks/{tid}/orgs", admin,
+            {"orgs": [{"orgId": org["id"], "expectedCount": GATE_USERS,
+                       "headUserId": uid_of(teacher)}]}, "设置覆盖班级与初审人")
 
     # N 个证件号（前 17 位递增，校验位由脚本补）
     rows = []
@@ -189,9 +231,16 @@ def build_fixture(admin):
         seq = "%03d" % (i % 1000)
         idc = R.with_checksum(area + "20080101" + seq)
         rows.append((f"压测{i:04d}", idc))
+    t_excel = time.perf_counter()
     build_roster(rows)
+    excel_s = time.perf_counter() - t_excel
+    t_imp = time.perf_counter()
     body = R.upload_roster(BASE, admin, tid, TMP_ROSTER,
                            extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]})
+    imp_s = time.perf_counter() - t_imp
+    # ⚠️ 两段分开报：**造 Excel 是压测脚本自己的成本，不是系统的成本** ——
+    #    混在一起会把 openpyxl 的耗时算到"名册导入"头上（度量工具的偏差伪装成产品缺陷）
+    print("      夹具计时：造 Excel %.1fs　导入接口 %.1fs（%d 行）" % (excel_s, imp_s, len(rows)))
     # ⚠️ **判据是 data.created，不是 code**：导入了 0 行也是 code=0
     #    （体检报告在 data 里）。判错的话压测会"没有样本"，
     #    而那个现象看起来像压测脚本坏了 —— 与 F-84 是同一个坑
@@ -212,7 +261,9 @@ def build_fixture(admin):
     if len(roster_rows) != GATE_USERS:
         raise SystemExit("夹具名册行数不对：期望 %d，实际 %d"
                          % (GATE_USERS, len(roster_rows)))
-    call("POST", f"/api/v1/tasks/{tid}/publish", admin, {"force": True})
+    # ⚠️ 发布**必须查返回码**：它失败时任务还是草稿，而后续所有学生接口都报
+    #    `40008 任务已截止、已归档或已暂停` —— 排查会一路查错方向（我踩过）
+    call_ok("POST", f"/api/v1/tasks/{tid}/publish", admin, {"force": True}, "发布任务")
     cfg = R.read_dev_config()
     tokens = []
     for i, r in enumerate(roster_rows):
@@ -245,15 +296,36 @@ def build_roster(rows):
     wb.save(TMP_ROSTER)
 
 
-def make_img(size_kb=400):
+def make_img(size_kb=400, px=(2000, 1500)):
-    """造一张尽量接近真实照片的图（纯色会被压得很小，失真）。"""
+    """
-    import random
+    造一张**真实**的 JPEG（Pillow 噪声图）。
-    random.seed(7)
+
-    # 最小合法 JPEG 头 + 随机填充（不是真 JPEG，但上传接口只看扩展名与大小）
+    ⚠️ 原来是"JPEG 头 + 随机字节"的假图。它能过扩展名与大小校验，
-    with open(TMP_IMG, "wb") as f:
+    但**服务端做的根本不是同一件事**：真图要解码 → 按 EXIF 转正 → 重编码，
-        f.write(b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00")
+    假图在解码那一步就失败了（然后走的是"压缩失败"的分支）。
-        f.write(bytes(random.getrandbits(8) for _ in range(size_kb * 1024)))
+    用压测去回答"压缩是 CPU 密集"，却喂它一个不需要压缩的文件 ——
-        f.write(b"\xff\xd9")
+    量到的不是压缩成本，而是错误处理的成本。**这类"夹具不对"的红/绿都不算数。**
+
+    噪声图是有意选的：纯色图会被压到极小，量不出 CPU 成本
+    （与 e2e-m9-thumb.py 里那条"纯色会被压得很小、失真"同一个理由）。
+    """
+    try:
+        from PIL import Image
+        import random
+        random.seed(7)
+        img = Image.new("RGB", px)
+        img.putdata([(random.randrange(256), random.randrange(256), random.randrange(256))
+                     for _ in range(px[0] * px[1])])
+        img.save(TMP_IMG, "JPEG", quality=92)
+        return os.path.getsize(TMP_IMG)
+    except ImportError:
+        import random
+        random.seed(7)
+        with open(TMP_IMG, "wb") as f:
+            f.write(b"\xff\xd8\xff\xe0\x00\x10JFIF\x00\x01\x01\x00\x00\x01\x00\x01\x00\x00")
+            f.write(bytes(random.getrandbits(8) for _ in range(size_kb * 1024)))
+            f.write(b"\xff\xd9")
+        return os.path.getsize(TMP_IMG)
 
 
 def reset_gate():
@@ -335,16 +407,57 @@ def scenario_read(tid, admin, conc, rounds):
     return stats
 
 
-def scenario_upload(tid, tokens, conc):
+def scenario_upload(tid, tokens, conc, px=(2000, 1500)):
     """上传：并发上传图片（压缩是 CPU 密集）。"""
     st = Stat("材料上传")
-    make_img()
+    size = make_img(px=px)
+    print("      夹具：真实 JPEG %d×%d / %.0f KB" % (px[0], px[1], size / 1024.0))
+
+    # ⚠️ 前置条件：**必须先把人绑到名册项上**，否则 `POST /mp/submissions` 会被拒
+    #    （"没有草稿"），而 40 次上传会整齐地失败 —— 那不叫"上传慢"，叫"夹具没成立"。
+    #    以前单独跑 `--only upload` 就是这个结果：全量压测里它跟在门禁后面，
+    #    "恰好"能跑，于是这个前置依赖一直没被发现（F-96 的同一类：没断言前置就往下走）。
+    #    绑定逻辑与导出大包共用 `bind_users` —— 写两份必然分叉，而分叉的表现就是
+    #    "另一个场景的前置悄悄失效"。
+    bound = bind_users(tid, tokens, conc)
+    print("      前置：门禁绑定 %d/%d 人（上传必须先有绑定）" % (bound, len(tokens)))
+    if bound < conc:
+        raise SystemExit("上传压测的前置条件不成立：只有 %d 人绑定成功，需要 %d 人"
+                         % (bound, conc))
+
     sub_ids = {}
-    # 每个用户先建草稿
+    for t in tokens[:conc + 1]:                # 多要一份：串行基线占用第 0 个
-    for t in tokens[:conc]:
         c, b = call("POST", "/api/v1/mp/submissions", t["token"], {"taskId": tid})
         if c == 200 and isinstance(b, dict) and b.get("code") == 0:
             sub_ids[t["uid"]] = b["data"]["id"]
+        else:
+            print("      ⚠️ 建草稿失败：%s" % json.dumps(b, ensure_ascii=False)[:160])
+    # ⚠️ 同样要断言：建不出草稿时"40 个请求全失败"看起来像上传功能坏了
+    if len(sub_ids) < conc + 1:
+        raise SystemExit("上传压测的前置条件不成立：只建出 %d 份草稿，需要 %d 份"
+                         % (len(sub_ids), conc + 1))
+
+    # ── 串行基线：单请求到底多贵 ──
+    # ⚠️ 必须有这一段。并发下的 p50/p95 里混着排队，**不能反推单请求成本**（F-94）；
+    #    而"值不值得优化"只能由"单请求成本 vs 排队占比"回答。
+    warm = tokens[0]
+    w_ms, w_http, w_body = timed(lambda: upload_file(
+        "/api/v1/mp/files/upload", warm["token"], sub_ids[warm["uid"]], TMP_IMG))
+    # ⚠️ 判据是 body 里的业务码，**不是** upload_file 返回的第一个值 ——
+    #    那是 HTTP 状态码（200），拿它当"成功"判据会把成功报成失败（我自己刚踩过）
+    if isinstance(w_body, dict) and w_body.get("code") == 0:
+        # 尺寸以库里那行为准：接口回的是原始大小，规范件大小只有落库才知道
+        sizes = one("select raw_size || '/' || coalesce(final_size, 0) from collect_file "
+                    "where submission_id=%d order by id desc limit 1;" % sub_ids[warm["uid"]])
+        try:
+            raw_kb, fin_kb = [int(x) / 1024.0 for x in sizes.split("/")]
+            ratio = (fin_kb / raw_kb * 100.0) if raw_kb else 0.0
+            print("      串行基线：单请求 %.0fms　压缩 %.0fKB → %.0fKB（%.0f%%，服务端真的在压）"
+                  % (w_ms, raw_kb, fin_kb, ratio))
+        except Exception:                                     # noqa: BLE001
+            print("      串行基线：单请求 %.0fms（尺寸没取到：%s）" % (w_ms, sizes))
+    else:
+        print("      串行基线失败：HTTP %s　%s" % (w_http, json.dumps(w_body, ensure_ascii=False)[:160]))
 
     def one_req(t):
         sid = sub_ids.get(t["uid"])
@@ -357,7 +470,9 @@ def scenario_upload(tid, tokens, conc):
         return ms, int(biz if biz is not None else code), body
 
     jobs = []
-    for t in tokens[:conc]:
+    # 并发用 tokens[1:conc+1]：第 0 个已经被串行基线用掉了，混进来会让"并发里的某个用户
+    # 比别人多传一次"，而样本数对不上时最难判断的是"到底谁慢"
+    for t in tokens[1:conc + 1]:
         for _ in range(UPLOAD_PER_USER):
             jobs.append(t)
     with cf.ThreadPoolExecutor(max_workers=conc) as ex:
@@ -400,6 +515,324 @@ def upload_file(path, token, submission_id, file_path, timeout=180):
         return -1, {"msg": str(e)}
 
 
+# ══════════════ 长时间稳定性 ══════════════
+
+def base_port():
+    from urllib.parse import urlparse
+    return urlparse(BASE).port or 80
+
+
+def app_pid(port):
+    """监听该端口的进程 pid（没有就返回 None，不猜）。"""
+    try:
+        r = subprocess.run(["powershell", "-NoProfile", "-Command",
+                            "(Get-NetTCPConnection -State Listen -LocalPort %d "
+                            "-ErrorAction SilentlyContinue).OwningProcess" % port],
+                           capture_output=True, text=True, timeout=30)
+        for tok in (r.stdout or "").split():
+            if tok.strip().isdigit():
+                return int(tok.strip())
+    except Exception:                                         # noqa: BLE001
+        pass
+    return None
+
+
+def rss_mb(pid):
+    """
+    进程的物理内存（MB）。
+
+    ⚠️ 从**系统**取，不从应用的指标取：应用自己报的堆用量看不到堆外内存、
+    元空间与线程栈 —— 而"内存慢慢涨"恰恰经常长在这些地方。
+    """
+    if not pid:
+        return None
+    try:
+        r = subprocess.run(["powershell", "-NoProfile", "-Command",
+                            "(Get-Process -Id %d -ErrorAction SilentlyContinue).WorkingSet64" % pid],
+                           capture_output=True, text=True, timeout=30)
+        v = (r.stdout or "").strip()
+        return int(v) / 1048576.0 if v.isdigit() else None
+    except Exception:                                         # noqa: BLE001
+        return None
+
+
+def db_conns():
+    """PG 上到 filldb 的连接数 —— Hikari 连接泄漏会直接体现为它单调上升。"""
+    v = one("select count(*) from pg_stat_activity where datname='filldb';")
+    return int(v) if v.isdigit() else None
+
+
+def scenario_soak(tid, admin, minutes, rps):
+    """
+    长时间稳定性：低强度持续打，每分钟采样一行。
+
+    ⚠️ 要回答的是"**有没有单调上升**"，所以判据是**趋势**，不是"跑完没崩"。
+    泄漏的特征是"慢慢地越来越高"，一次性压测永远测不出来 ——
+    所以这里每分钟记 RSS 与 PG 连接数，最后比较**前 1/3 与后 1/3** 的均值。
+    延迟趋势一起看：泄漏的另一个表现是"越跑越慢"。
+    """
+    paths = [
+        ("GET", "/api/v1/tasks?pageNum=1&pageSize=20", "任务列表"),
+        ("GET", f"/api/v1/tasks/{tid}/roster?pageNum=1&pageSize=50", "名册列表"),
+        ("GET", f"/api/v1/tasks/{tid}/statistics", "任务统计"),
+        ("GET", "/api/v1/audit/todos?pageNum=1&pageSize=20", "审核待办"),
+    ]
+    pid = app_pid(base_port())
+    print("  目标后端 pid=%s（RSS 从系统取，PG 连接数从 pg_stat_activity 取）" % pid)
+    print("  强度 %.1f 请求/秒　时长 %d 分钟，每分钟一行样本" % (rps, minutes))
+    rows = []
+    n = 0
+    t_end = time.time() + minutes * 60
+    while time.time() < t_end:
+        lat, fail = [], 0
+        for _ in range(max(1, int(rps * 60))):
+            m, p, _label = paths[n % len(paths)]
+            n += 1
+            ms, code, body = timed(lambda: call(m, p, admin))
+            biz = int((body or {}).get("code", code)) if isinstance(body, dict) else code
+            lat.append(ms)
+            if biz != 0:
+                fail += 1
+            # 均匀铺开：不要"一分钟里的前 10 秒打完全部请求"，那测的是尖峰不是持续压力
+            time.sleep(max(0.0, 1.0 / rps))
+        s = sorted(lat)
+        row = {"min": len(rows) + 1, "n": len(lat), "fail": fail,
+               "p50": s[len(s) // 2], "p95": s[min(len(s) - 1, int(len(s) * 0.95))],
+               "rss": rss_mb(pid), "conns": db_conns()}
+        rows.append(row)
+        print("    %3d 分　请求 %-4d 失败 %-3d p50 %6.1fms p95 %7.1fms　RSS %7s MB　PG 连接 %4s"
+              % (row["min"], row["n"], row["fail"], row["p50"], row["p95"],
+                 ("%.0f" % row["rss"]) if row["rss"] else "—",
+                 row["conns"] if row["conns"] is not None else "—"))
+
+    def avg(rs, key):
+        vals = [r[key] for r in rs if r.get(key) is not None]
+        return sum(vals) / len(vals) if vals else None
+
+    third = max(1, len(rows) // 3)
+    head, tail = rows[:third], rows[-third:]
+    print("\n  样本 %d 分钟　请求 %d　失败 %d（%s）"
+          % (len(rows), sum(r["n"] for r in rows), sum(r["fail"] for r in rows),
+             "强度 %.1f/s、规模 %d 行" % (rps, GATE_USERS)))
+    for key, unit, label, tol in (("rss", "MB", "RSS", 10.0), ("conns", "个", "PG 连接数", 20.0)):
+        a, b = avg(head, key), avg(tail, key)
+        if a is None or b is None:
+            print("  %-10s 样本不足，没测出来" % label)
+            continue
+        pct = ((b - a) / a * 100.0) if a else 0.0
+        verdict = "平稳" if abs(pct) < tol else ("**上升** ⚠️ 疑似泄漏" if b > a else "下降")
+        print("  %-10s 前 1/3 均值 %8.1f %s → 后 1/3 均值 %8.1f %s（%+.1f%%）　%s"
+              % (label, a, unit, b, unit, pct, verdict))
+    a95, b95 = avg(head, "p95"), avg(tail, "p95")
+    if a95 and b95:
+        print("  %-10s 前 1/3 均值 %8.1f ms → 后 1/3 均值 %8.1f ms（%+.1f%%）"
+              % ("延迟 p95", a95, b95, (b95 - a95) / a95 * 100.0))
+    return rows
+
+
+def bind_users(tid, tokens, workers=10):
+    """
+    把 N 个人绑到名册项上（上传 / 提交 / 审核的**共同前置**）。
+
+    ⚠️ 抽出来是有原因的：只在 `scenario_upload` 里写一份的话，新场景（导出大包）
+    就会忘 —— 而忘了的表现是"请求整齐地全失败"，看起来像功能坏了（F-96 的同一类）。
+    """
+    ok = 0
+    fail_sample = None
+
+    def one(t):
+        nonlocal fail_sample
+        c, b = call("POST", "/api/v1/mp/roster/match", t["token"],
+                    {"taskId": tid, "realName": t["name"], "idCardNo": t["idCardNo"]})
+        code = b.get("code") if isinstance(b, dict) else -1
+        if code not in (0, 40072) and fail_sample is None:   # 0 绑定成功 / 40072 已绑定
+            fail_sample = json.dumps(b, ensure_ascii=False)[:160]
+        return code
+
+    with cf.ThreadPoolExecutor(max_workers=workers) as ex:
+        for code in ex.map(one, tokens):
+            if code in (0, 40072):
+                ok += 1
+    if fail_sample is not None:
+        print("      ⚠️ 绑定失败样本：%s" % fail_sample)
+    return ok
+
+
+def scenario_export_big(tid, tokens, admin, teacher, files_per_user=1, split_mb=0, conc=10):
+    """
+    导出**大包**：造 N 份已通过的填报单，量打包墙钟、体积、分卷与产物的完整性。
+
+    这是"交付物本身"的那条链，也是唯一一条**没法用接口断言替代**的验证：
+    打包耗时、产物体积、分卷是否每卷都能单独解开，只有真跑一遍才知道。
+    """
+    import zipfile
+
+    n = len(tokens)
+    print("      目标：%d 人 × %d 个材料%s"
+          % (n, files_per_user, "，分卷 %dMB" % split_mb if split_mb else "，不分卷"))
+
+    bound = bind_users(tid, tokens, conc)
+    print("      ① 门禁绑定 %d/%d 人" % (bound, n))
+    if bound < n:
+        print("         ⚠️ 有 %d 人没绑上 —— 后面的上传会失败，先看失败码再判断是不是导出的问题"
+              % (n - bound))
+
+    sid = {}
+    draft_fail = None
+
+    def draft(t):
+        nonlocal draft_fail
+        c, b = call("POST", "/api/v1/mp/submissions", t["token"], {"taskId": tid})
+        if not (isinstance(b, dict) and b.get("code") == 0):
+            # ⚠️ 失败**必须打出来**：上一次它只有一句"草稿 0/150"，
+            #    排查时完全看不出是绑定没生效、还是任务状态不对（F-96 的同一类）
+            if draft_fail is None:
+                draft_fail = json.dumps(b, ensure_ascii=False)[:200]
+            return t["uid"], None
+        return t["uid"], (b.get("data") or {}).get("id")
+
+    with cf.ThreadPoolExecutor(max_workers=conc) as ex:
+        for uid, s in ex.map(draft, tokens):
+            if s:
+                sid[uid] = s
+    print("      ② 草稿 %d/%d 份%s"
+          % (len(sid), n, "" if draft_fail is None else "　首个失败：%s" % draft_fail))
+    if len(sid) < n:
+        raise SystemExit("草稿没建齐（%d/%d），导出大包的夹具不成立" % (len(sid), n))
+
+    img_kb = make_img() / 1024.0
+    jobs = [(t, k) for t in tokens if t["uid"] in sid for k in range(files_per_user)]
+
+    def up(job):
+        t, _k = job
+        _c, b = upload_file("/api/v1/mp/files/upload", t["token"], sid[t["uid"]], TMP_IMG)
+        return (b or {}).get("code") if isinstance(b, dict) else -1
+
+    t_up = time.perf_counter()
+    up_fail = 0
+    with cf.ThreadPoolExecutor(max_workers=conc) as ex:
+        for code in ex.map(up, jobs):
+            if code != 0:
+                up_fail += 1
+    print("      ③ 上传 %d 张（失败 %d；%.0fKB/张；%.1fs）"
+          % (len(jobs), up_fail, img_kb, time.perf_counter() - t_up))
+    if up_fail:
+        raise SystemExit("上传有 %d 张失败，夹具不成立（先看失败码，别当成导出的问题）" % up_fail)
+
+    def submit(t):
+        c, b = call("POST", "/api/v1/mp/submissions/%d/submit" % sid[t["uid"]], t["token"],
+                    {"confirmWarnings": True})
+        # ⚠️ 判据是 data.submitted（F-84）：HTTP 200 + code 0 不等于"提交成功"
+        return ((b or {}).get("data") or {}).get("submitted") if isinstance(b, dict) else False
+
+    ok_sub = 0
+    with cf.ThreadPoolExecutor(max_workers=conc) as ex:
+        for v in ex.map(submit, [t for t in tokens if t["uid"] in sid]):
+            if v is True:
+                ok_sub += 1
+    print("      ④ 提交 %d/%d 份（判据 data.submitted）" % (ok_sub, len(sid)))
+    if ok_sub != len(sid):
+        raise SystemExit("有 %d 份没提交成功，夹具不成立" % (len(sid) - ok_sub))
+
+    ids = [sid[t["uid"]] for t in tokens if t["uid"] in sid]
+
+    def approve(sid_, token, body_, label):
+        c, b = call("POST", "/api/v1/audit/submissions/%d/approve" % sid_, token, body_, timeout=120)
+        return (b or {}).get("code") if isinstance(b, dict) else -1, b
+
+    # ⚠️ 一期**恒有两段审核**（needClassAudit 被强制 true，见 build_fixture 的说明），
+    #    所以"逐份通过"要走两次：班主任初审 → 管理员终审。
+    #    只做一段的表现是全部停在 40301「不在你负责的班级里」或"没有待你处理的审核动作"。
+    for stage, (tok, body_, label) in enumerate((
+            (teacher, {"actionSource": "BUTTON", "remark": "压测初审"}, "初审"),
+            (admin, {"actionSource": "BUTTON", "finalOpinion": "压测终审"}, "终审"))):
+        ok_n, first_fail = 0, None
+        with cf.ThreadPoolExecutor(max_workers=conc) as ex:
+            for code, b in ex.map(lambda s: approve(s, tok, body_, label), ids):
+                if code == 0:
+                    ok_n += 1
+                elif first_fail is None:
+                    first_fail = b
+        print("      ⑤-%d %s通过：%d/%d 份%s"
+              % (stage + 1, label, ok_n, len(ids),
+                 "" if first_fail is None else "　首个失败：%s"
+                 % json.dumps(first_fail, ensure_ascii=False)[:160]))
+        if first_fail is not None:
+            raise SystemExit("%s这一关没过，夹具不成立" % label)
+
+    _c, b = call("GET", "/api/v1/exports/precheck?taskId=%d&onlyApproved=true" % tid, admin)
+    pre = (b or {}).get("data") or {}
+    print("      ⑥ 前置检查：canExport=%s　候选 %s 人 / %s 个文件 / 原件 %.1f MB"
+          % (pre.get("canExport"), pre.get("candidateCount"), pre.get("fileCount"),
+             (pre.get("rawSizeBytes") or 0) / 1048576.0))
+    if pre.get("canExport") is not True:
+        raise SystemExit("前置检查不通过，大包测不了：%s"
+                         % json.dumps(pre.get("blocking"), ensure_ascii=False)[:200])
+
+    body = {"taskId": tid, "exportName": "压测大包 %s" % SUFFIX, "scope": {"onlyApproved": True}}
+    if split_mb:
+        body["splitSizeMb"] = split_mb
+    t0 = time.perf_counter()
+    _c, b = call("POST", "/api/v1/exports", admin, body, timeout=600)
+    eid = ((b or {}).get("data") or {}).get("exportId")
+    if (b or {}).get("code") != 0 or not eid:
+        raise SystemExit("创建导出失败：%s" % json.dumps(b, ensure_ascii=False)[:200])
+    print("      ⑦ 导出 id=%s，创建耗时 %.1fs → 轮询到完成（打包是异步的，量墙钟）"
+          % (eid, time.perf_counter() - t0))
+
+    detail, seen = None, None
+    while time.perf_counter() - t0 < 1800:
+        _c, b = call("GET", "/api/v1/exports/%d" % eid, admin)
+        d = (b or {}).get("data") or {}
+        if d.get("status") in (2, 3, 4):
+            detail = d
+            break
+        if d.get("progress") != seen:
+            seen = d.get("progress")
+            print("         …%s%%　%s" % (seen, d.get("stepText") or ""))
+        time.sleep(2)
+    wall = time.perf_counter() - t0
+    if not detail or detail.get("status") != 2:
+        print("      ✗ 导出没跑完：status=%s　errorMsg=%s"
+              % (detail and detail.get("status"), detail and detail.get("errorMsg")))
+        return None
+
+    s = detail.get("stats") or {}
+    print("      ⑧ **打包墙钟 %.1fs**　文件 %s 个　原件 %.1f MB → 产物 %.1f MB（%s）"
+          % (wall, s.get("fileCount"), s.get("rawSizeMb") or 0, s.get("finalSizeMb") or 0,
+             s.get("compressRatio")))
+
+    # ⚠️ 列名以 DDL 为准（V11）：`part_no` / `file_name` / `file_size` / `file_path` ——
+    #    按直觉写成 part/name/size/path 会直接 SQL 报错（我第一版就是这么写的）
+    vols = sql("select part_no, file_name, coalesce(file_size,0), file_path "
+               "from collect_export_file where export_id=%d and deleted=0 order by part_no;" % eid)
+    rows = [x for x in vols.split("\n") if x.strip()]
+    print("      ⑨ 分卷 %d 个：" % len(rows))
+    for line in rows:
+        part, name, size, path = (line.split("|") + ["", "", "", ""])[:4]
+        print("         %s 卷 %s　%.1f MB" % (name, part, int(size or 0) / 1048576.0))
+    # ⚠️ 分卷的承诺是"**每一卷都是完整可解的 ZIP**"（不是 .z01 那种必须凑齐才能解）。
+    #    这条只能真去解一遍 —— 断言"文件存在"是断言不到它的。
+    base = os.path.join(ROOT, "backend", "data", "uploads")
+    for line in rows:
+        part, name, size, path = (line.split("|") + ["", "", "", ""])[:4]
+        full = path if os.path.isabs(path) else os.path.join(base, path)
+        if not os.path.exists(full):
+            # 存储是七牛时产物也可能不在本地（D5 说导出包落本地盘，所以这条算异常）
+            print("         ? %s 不在本地磁盘（路径 %s）" % (name, full))
+            continue
+        try:
+            with zipfile.ZipFile(full) as z:
+                names = z.namelist()
+                bad = z.testzip()
+                print("         ✓ %s 可解：%d 个条目，首条 %s%s"
+                      % (name, len(names), names[0] if names else "（空）",
+                         "" if bad is None else "　✗ CRC 错误：%s" % bad))
+        except Exception as e:                                # noqa: BLE001
+            print("         ✗ %s **解不开**：%s" % (name, e))
+    return detail
+
+
 def scenario_export(tid, admin):
     """导出：前置检查（只读）+ 创建（同步打包）。"""
     st_pre = Stat("导出前置检查")
@@ -442,37 +875,80 @@ def scenario_export(tid, admin):
 # ══════════════ 清理 ══════════════
 
 def cleanup():
+    # ⚠️ 删库行之前先把**磁盘上的文件路径**取出来：压测会传几百张图、打出几百 MB 的包，
+    #    只删库行的话这些文件会一直躺在 backend/data/uploads 里
+    #    （"删库不删桶"的本地版本 —— 时间一长本地磁盘就是一本糊涂账）
+    disk = []
+    for tid in TASK_IDS:
+        for stmt in ("select coalesce(std_path, raw_path) from collect_file where task_id=%s;",
+                     "select file_path from collect_export_file where task_id=%s;"):
+            for line in sql(stmt % tid).split("\n"):
+                p = line.strip()
+                if p and p != "null":
+                    disk.append(p)
+
     for tid in TASK_IDS:
         for tbl in ("collect_validation_issue", "collect_audit_record", "collect_notify_record",
-                    "collect_file", "collect_submission", "collect_roster",
+                    "collect_export_file", "collect_file", "collect_submission", "collect_roster",
                     "collect_task_org", "collect_material_def"):
             sql("delete from %s where task_id=%s;" % (tbl, tid))
         sql("delete from collect_task where id=%s;" % tid)
     for uid in UIDS:
         sql("delete from collect_message where user_id=%d;" % uid)
+
+    base = os.path.join(ROOT, "backend", "data", "uploads")
+    removed = 0
+    for p in disk:
+        full = p if os.path.isabs(p) else os.path.join(base, p)
+        # ⚠️ 只删**确实存在**的本地文件：存储是七牛时这些路径是 bucket key，
+        #    本地根本不存在 —— 那时候要清的是桶，不该在本地乱删（也删不到）
+        if os.path.isfile(full):
+            try:
+                os.remove(full)
+                removed += 1
+            except OSError:
+                pass
     for p in (TMP_ROSTER, TMP_IMG):
         if os.path.exists(p):
             os.remove(p)
-    print("  已清理：压测任务 / 名册 / 材料 / 消息 / 临时文件")
+    print("  已清理：压测任务 / 名册 / 材料 / 消息 / 临时文件（本地文件 %d 个）" % removed)
 
 
 def main():
     # ⚠️ `global` 必须出现在**同一函数里第一次使用该名字之前** ——
     #    放晚了会报 "name 'X' is used prior to global declaration"
     global GATE_USERS
+    global BASE
 
-    ap = argparse.ArgumentParser(description="压测（门禁/上传/读/导出）")
+    ap = argparse.ArgumentParser(description="压测（门禁/上传/读/导出/长跑）")
-    ap.add_argument("--only", default="", help="只跑一个场景：gate / read / upload / export")
+    ap.add_argument("--only", default="",
+                    help="只跑一个场景：gate / read / upload / export / export-big / soak")
     ap.add_argument("--conc", type=int, default=0, help="覆盖并发数")
     ap.add_argument("--clean", action="store_true", help="只清理上次残留")
     ap.add_argument("--rows", type=int, default=0,
                     help="名册人数（默认 %d）。⚠️ 谈响应时间必须带上规模" % GATE_USERS)
+    ap.add_argument("--base", default="",
+                    help="后端地址（默认 %s）。要量上传就另起一个 local 存储的实例" % BASE)
     ap.add_argument("--with-upload", action="store_true",
                     help="跑上传压测（⚠️ 会往真实对象存储写文件，默认关闭）")
+    ap.add_argument("--img-px", default="2000x1500",
+                    help="上传夹具的图片尺寸（真实 JPEG，压缩成本与它成正比）")
+    ap.add_argument("--fixture-only", action="store_true",
+                    help="只造夹具并计时（量名册导入耗时用，不做任何压测）")
+    ap.add_argument("--soak", type=int, default=0,
+                    help="长时间稳定性：持续 N 分钟，每分钟采一行 RSS/PG 连接数/延迟")
+    ap.add_argument("--rps", type=float, default=5.0, help="长跑强度（请求/秒，默认 5）")
+    ap.add_argument("--files-per-user", type=int, default=1,
+                    help="导出大包场景：每人几个材料（默认 1）")
+    ap.add_argument("--split-mb", type=int, default=0,
+                    help="导出大包场景：按 N MB 分卷（0 = 不分卷，用服务端默认）")
     args = ap.parse_args()
 
+    if args.base:
+        BASE = args.base.rstrip("/")
     if args.rows > 0:
         GATE_USERS = args.rows
+    px = tuple(int(x) for x in args.img_px.lower().split("x"))
 
     if args.clean:
         # 残留任务按名字前缀找
@@ -497,12 +973,29 @@ def main():
         print("管理员登录失败：" + str(adm))
         return 2
     admin = adm["data"]["token"]
+    # 班主任账号：压测的导出大包场景要真的走一遍"初审 → 终审"，
+    # 而这个账号由 scripts/dev-seed-audit.py 摆出来（演示数据的一部分）
+    _, tea = R.login(BASE, TEACHER_USER, TEACHER_PWD, "demo")
+    if tea.get("code") != 0:
+        print("演示班主任 %s 登录失败（先跑 scripts/dev-seed-audit.py）：%s" % (TEACHER_USER, tea))
+        return 2
+    teacher = tea["data"]["token"]
 
     print("\n── 造夹具 ──")
     t0 = time.perf_counter()
-    tid, tokens = build_fixture(admin)
+    tid, tokens = build_fixture(admin, teacher)
-    print("  任务 %d，名册 %d 人，夹具耗时 %.1fs"
+    print("  任务 %d，名册 %d 人，夹具耗时 %.1fs（含名册导入本身）"
           % (tid, len(tokens), time.perf_counter() - t0))
+    if args.fixture_only:
+        # 量"导入 N 行"这件事：接口耗时在上面那条"夹具计时"里，这里补内存与库内行数
+        print("\n  --fixture-only：夹具已造好，直接清理")
+        pid = app_pid(base_port())
+        print("  后端 RSS：%.0f MB（pid=%s）" % (rss_mb(pid) or -1, pid))
+        print("  库里名册行数：%s（应为 %d）"
+              % (one("select count(*) from collect_roster where task_id=%d and deleted=0;" % tid),
+                 GATE_USERS))
+        cleanup()
+        return 0
 
     stats = {}
     conc = args.conc
@@ -516,15 +1009,22 @@ def main():
             stats["read"] = scenario_read(tid, admin, conc or READ_CONC, READ_ROUNDS)
         if args.only == "upload" or (args.with_upload and args.only == ""):
             print("\n── ③ 材料上传（%d 人同时传，压缩是 CPU 密集）──" % UPLOAD_USERS)
-            stats["upload"] = scenario_upload(tid, tokens, conc or UPLOAD_USERS)
+            stats["upload"] = scenario_upload(tid, tokens, conc or UPLOAD_USERS, px)
         if args.only == "" and not args.with_upload:
             print("\n── ③ 材料上传（**默认跳过**）──")
             print("  ⚠️ 当前 dev 的存储是七牛**真实空间**（见 backend/application-local.yml），")
-            print("     压测上传会打到外部服务并把测试文件灌进真实桶。要跑就加 --with-upload，")
+            print("     压测上传会打到外部服务并把测试文件灌进真实桶。两种跑法：")
-            print("     且最好先把存储切成本地：XT_STORAGE_PROVIDER=local 重启后端。")
+            print("     ① `--base http://localhost:18080` 打另起的 local 实例（推荐，见 docs/09 §9.5）；")
+            print("     ② `XT_STORAGE_PROVIDER=local` 重启当前后端后再加 --with-upload。")
         if args.only in ("", "export"):
             print("\n── ④ 导出（前置检查 + 创建）──")
             scenario_export(tid, admin)
+        if args.only == "export-big":
+            print("\n── ④' 导出大包（造 N 份已通过的单 → 打包 → 验产物）──")
+            scenario_export_big(tid, tokens, admin, teacher, args.files_per_user, args.split_mb)
+        if args.only == "soak" or args.soak > 0:
+            print("\n── ⑤ 长时间稳定性（%d 分钟 × %.1f 请求/秒）──" % (args.soak, args.rps))
+            scenario_soak(tid, admin, args.soak, args.rps)
     finally:
         print("\n── 清理 ──")
         cleanup()