perf:大批计算提速与回刷脚本增强

- 分子聚合改values_list按JSON键取值, 不再拉子批整份data
- 子批发现改按层批量查询(每层2次), 替代逐节点循环查询
- cal_zt_big加use_white_cache参数, fresh回刷不再额外读写大批data
- 脚本内关闭DEBUG(避免SQL日志内存累积拖慢长任务)
- 新增--skip-resolve跳过归属解析直接算大批; 默认起始2026-06-01

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/wpm/scripts/batch_gxerp.py

@@ -440,8 +440,11 @@ def _zt_white_legacy(batchst: BatchSt, depth, path, memo):
     return p_white * ratio, p_kinds | {"谱系兜底"}, p_clean
 
 
-def cal_zt_big(big_batch: str):
+def cal_zt_big(big_batch: str, use_white_cache=True):
-    """重算大批的直通良率并写入其 BatchSt.data"""
+    """重算大批的直通良率并写入其 BatchSt.data
+
+    use_white_cache=False 时忽略已缓存的白料数强制重新回溯(供回刷纠错)。
+    """
     try:
         big_bs = BatchSt.objects.get(batch=big_batch, version=1)
     except BatchSt.DoesNotExist:
@@ -454,27 +457,36 @@ def cal_zt_big(big_batch: str):
     # 从大批侧发现子批: 拆批目标 + 完全由本大批子批合并出的复检批(如 大批-A-1), 支持多级
     known = {big_batch}
     sub_map = {}
-    frontier = [big_bs]
+    frontier_ids = [big_bs.id]
     for _ in range(3):
         new_nodes = []
-        for node in frontier:
+        merge_targets = {}
-            for e in BatchLog.objects.filter(source=node).select_related("target"):
+        for e in BatchLog.objects.filter(source_id__in=frontier_ids).select_related("target"):
-                t = e.target
+            t = e.target
+            if t.batch in known:
+                continue
+            if e.relation_type == "split" and e.handover_id:
+                known.add(t.batch)
+                sub_map[t.batch] = t
+                new_nodes.append(t)
+            elif e.relation_type == "merge":
+                merge_targets[t.id] = t
+        if merge_targets:
+            src_by_target = {}
+            for tid, s_batch, s_zt in BatchLog.objects.filter(
+                    target_id__in=merge_targets.keys(), relation_type="merge"
+            ).values_list("target_id", "source__batch", "source__zt_batch"):
+                src_by_target.setdefault(tid, []).append((s_batch, s_zt))
+            for tid, t in merge_targets.items():
                 if t.batch in known:
                     continue
-                if e.relation_type == "split" and e.handover_id:
+                pairs = src_by_target.get(tid, [])
-                    ok = True
+                if pairs and all(s_batch in known or s_zt == big_batch for s_batch, s_zt in pairs):
-                elif e.relation_type == "merge":
-                    m_srcs = BatchLog.objects.filter(target=t, relation_type="merge").select_related("source")
-                    ok = all(s.source.batch in known or s.source.zt_batch == big_batch for s in m_srcs)
-                else:
-                    ok = False
-                if ok:
                     known.add(t.batch)
                     sub_map[t.batch] = t
                     new_nodes.append(t)
-        frontier = new_nodes
+        frontier_ids = [t.id for t in new_nodes]
-        if not frontier:
+        if not frontier_ids:
             break
     # 锁定归属
     for b, t in sub_map.items():
@@ -488,16 +500,17 @@ def cal_zt_big(big_batch: str):
     # 分子: 所有归属本大批的批次(检验小批/复检批)外观检验直通合格数之和
     count_zt = 0
     sub_batches = []
-    for sub in BatchSt.objects.filter(zt_batch=big_batch, version=1).exclude(batch=big_batch):
+    for b, v in BatchSt.objects.filter(zt_batch=big_batch, version=1).exclude(
-        sub_batches.append(sub.batch)
+            batch=big_batch).values_list("batch", "data__外观检验_直通合格数"):
-        count_zt += (sub.data or {}).get("外观检验_直通合格数", 0) or 0
+        sub_batches.append(b)
+        count_zt += v or 0
     # 大批未拆批直接检验时自身也计入
     count_zt += (big_bs.data or {}).get("外观检验_直通合格数", 0) or 0
 
     # 分母: 白料数(大批成型后基本不变, 已算过的直接复用, 含算不出的负缓存;
     # 大批自身重算时 main 会重建 data, 缓存自然失效并重新回溯)
     data = big_bs.data or {}
-    if "直通_白料数" in data:
+    if use_white_cache and "直通_白料数" in data:
         white = data.get("直通_白料数")
         white = Decimal(str(white)) if white else None
         kinds = set((data.get("直通_口径") or "").split(";")) - {"", "谱系异常(合格数大于白料数)"}

scripts/correct_zt_gxerp.py

@@ -4,9 +4,9 @@
 不重建批次的其他统计数据(那个用 correct_batchst.py)。
 
 用法:
-    python scripts/correct_zt_gxerp.py [起始日期] [--fresh] [--workers=N]
+    python scripts/correct_zt_gxerp.py [起始日期] [--fresh] [--workers=N] [--skip-resolve]
-    起始日期默认 2025-06-17; --fresh 忽略旧锁重新解析归属并清缓存重算白料(纠错模式);
+    起始日期默认 2026-06-01; --fresh 忽略旧锁重新解析归属并强制重算白料(纠错模式);
-    --workers 并发线程数, 默认 8
+    --workers 并发线程数, 默认 8; --skip-resolve 跳过归属解析, 直接用已落库的 zt_batch 算大批
 """
 import os
 import sys
@@ -19,6 +19,9 @@ sys.path.insert(0, BASE_DIR)
 os.environ.setdefault("DJANGO_SETTINGS_MODULE", "server.settings")
 django.setup()
 
+from django.conf import settings
+settings.DEBUG = False  # DEBUG 下 Django 会把每条 SQL 记入内存, 长脚本越跑越慢
+
 import concurrent.futures
 import threading
 from datetime import datetime
@@ -31,48 +34,56 @@ from apps.utils.tools import MyJSONEncoder
 _print_lock = threading.Lock()
 
 
-def main(since: str = "2025-06-17", fresh: bool = False, workers: int = 8):
+def main(since: str = "2026-06-01", fresh: bool = False, workers: int = 8, skip_resolve: bool = False):
     since_dt = datetime.strptime(since, "%Y-%m-%d")
-    cond = Q(data__has_key="尺寸检验_日期") | Q(data__has_key="外观检验_日期")
-    if fresh:
-        # 纠错模式需覆盖曾被(错误)锁定/计算过的批
-        cond = cond | Q(zt_batch__isnull=False) | Q(data__has_key="直通_良率")
-    qs = BatchSt.objects.filter(create_time__gte=since_dt, version=1).filter(cond).order_by("create_time")
-    items = list(qs)
-    total = len(items)
-    print(f"检验批 {total} 个, 开始解析大批归属 (fresh={fresh}, workers={workers})", flush=True)
-
     done = [0]
 
-    def resolve_one(bs):
+    if skip_resolve:
-        try:
+        # 直接用已落库的归属
-            # fresh 模式忽略旧锁, 从谱系重新解析(纠正多级拆批错锁)
+        seen = set(BatchSt.objects.filter(
-            big = resolve_zt_big(bs, bs.data or {}, use_lock=not fresh)
+            create_time__gte=since_dt, version=1, zt_batch__isnull=False,
-            if big and bs.zt_batch != big:
+        ).values_list("zt_batch", flat=True).distinct())
-                bs.zt_batch = big
+        bigs = sorted(seen)
-                bs.save(update_fields=["zt_batch", "update_time"])
+        print(f"跳过归属解析, 已落库大批 {len(bigs)} 个 (workers={workers})", flush=True)
-            return big
+    else:
-        except Exception as e:
+        cond = Q(data__has_key="尺寸检验_日期") | Q(data__has_key="外观检验_日期")
-            with _print_lock:
+        if fresh:
-                print(f"{bs.batch} 归属解析失败: {e}", flush=True)
+            # 纠错模式需覆盖曾被(错误)锁定/计算过的批
-            return None
+            cond = cond | Q(zt_batch__isnull=False) | Q(data__has_key="直通_良率")
-        finally:
+        qs = BatchSt.objects.filter(create_time__gte=since_dt, version=1).filter(cond).order_by("create_time")
-            done[0] += 1
+        items = list(qs)
-            if done[0] % 1000 == 0:
+        total = len(items)
+        print(f"检验批 {total} 个, 开始解析大批归属 (fresh={fresh}, workers={workers})", flush=True)
+
+        def resolve_one(bs):
+            try:
+                # fresh 模式忽略旧锁, 从谱系重新解析(纠正多级拆批错锁)
+                big = resolve_zt_big(bs, bs.data or {}, use_lock=not fresh)
+                if big and bs.zt_batch != big:
+                    bs.zt_batch = big
+                    bs.save(update_fields=["zt_batch", "update_time"])
+                return big
+            except Exception as e:
                 with _print_lock:
-                    print(f"归属解析 {done[0]}/{total}", flush=True)
+                    print(f"{bs.batch} 归属解析失败: {e}", flush=True)
+                return None
+            finally:
+                done[0] += 1
+                if done[0] % 1000 == 0:
+                    with _print_lock:
+                        print(f"归属解析 {done[0]}/{total}", flush=True)
 
-    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
+        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
-        results = list(pool.map(resolve_one, items))
+            results = list(pool.map(resolve_one, items))
 
-    bigs = []  # 保序去重
+        bigs = []  # 保序去重
-    seen = set()
+        seen = set()
-    for big in results:
+        for big in results:
-        if big and big not in seen:
+            if big and big not in seen:
-            seen.add(big)
+                seen.add(big)
-            bigs.append(big)
+                bigs.append(big)
 
-    if fresh:
+    if fresh and not skip_resolve:
         # 清理: 曾被错当成大批计算过直通字段、如今无任何子批指向的批,
         # 剥离其直通键并解除自指锁定, 交由后续大批计算重新归属
         cleaned = 0
@@ -101,14 +112,7 @@ def main(since: str = "2025-06-17", fresh: bool = False, workers: int = 8):
 
     def cal_one(big):
         try:
-            if fresh:
+            res = cal_zt_big(big, use_white_cache=not fresh) or {}
-                big_bs = BatchSt.objects.filter(batch=big, version=1).first()
-                if big_bs and "直通_白料数" in (big_bs.data or {}):
-                    data = big_bs.data
-                    data.pop("直通_白料数", None)
-                    big_bs.data = json.loads(json.dumps(data, cls=MyJSONEncoder))
-                    big_bs.save(update_fields=["data", "update_time"])
-            res = cal_zt_big(big) or {}
             done[0] += 1
             with _print_lock:
                 print(f"[{done[0]}/{len(bigs)}] {big} 良率={res.get('良率')} 白料={res.get('白料数')} "
@@ -125,13 +129,14 @@ def main(since: str = "2025-06-17", fresh: bool = False, workers: int = 8):
 
 if __name__ == "__main__":
     fresh = "--fresh" in sys.argv
+    skip_resolve = "--skip-resolve" in sys.argv
     workers = 8
     args = []
     for a in sys.argv[1:]:
-        if a == "--fresh":
+        if a in ("--fresh", "--skip-resolve"):
             continue
         if a.startswith("--workers="):
             workers = int(a.split("=", 1)[1])
             continue
         args.append(a)
-    main(since=args[0] if args else "2025-06-17", fresh=fresh, workers=workers)
+    main(since=args[0] if args else "2026-06-01", fresh=fresh, workers=workers, skip_resolve=skip_resolve)