fix:归属解析支持多级拆批爬链与未拆批直检大批

- 孙批沿拆批链爬到顶层大批, 修复中间子批被错锁为大批
- 复检批合批来源统一按爬链顶层判定, 修复"来源大批不一致"误报
- 来源均为炉料链且批号为大批格式时自身即大批(未拆批直接检验);
  不良集中池等非大批格式不归属
- correct脚本--fresh忽略旧锁重新解析, 并清理错误的大批标记与直通键

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

apps/wpm/scripts/batch_gxerp.py

@@ -229,33 +229,60 @@ def _zt_split_parents(batchst: BatchSt):
     return {e.source.batch for e in qs}
 
 
-def resolve_zt_big(batchst: BatchSt, data: dict):
+def _zt_climb_split_top(batchst: BatchSt, max_hops=4):
-    """确定检验小批所属的大批号, 无法确定返回 None"""
+    """沿拆批链向上爬到顶层批(多级拆批的孙批归属顶层大批)
-    if batchst.zt_batch:
+
+    无拆批父级返回 None; 多个拆批父级(异常)停在当前层。
+    """
+    node = batchst
+    top = None
+    for _ in range(max_hops):
+        ps = _zt_split_parents(node)
+        if len(ps) != 1:
+            break
+        top = ps.pop()
+        parent = BatchSt.objects.filter(batch=top, version=1).first()
+        if parent is None:
+            break
+        node = parent
+    return top
+
+
+def resolve_zt_big(batchst: BatchSt, data: dict, use_lock=True):
+    """确定检验小批所属的大批号, 无法确定返回 None
+
+    use_lock=False 时忽略已锁定的 zt_batch, 重新从谱系解析(供回刷纠错)。
+    """
+    if use_lock and batchst.zt_batch:
         return batchst.zt_batch
     if not any(k in data for k in ("尺寸检验_count_real", "外观检验_count_real")):
         return None
-    parents = _zt_split_parents(batchst)
+    top = _zt_climb_split_top(batchst)
-    if len(parents) == 1:
+    if top:
-        return parents.pop()
+        return top
-    if len(parents) > 1:
+    # 合批复检批(如 大批-A-1): 所有可归属的合批来源须属于同一大批
-        myLogger.error(f"直通统计-{batchst.batch}存在多个拆批来源{parents}, 无法归属大批")
-        return None
-    # 合批复检批(如 大批-A-1): 所有合批来源须归属同一大批
     merge_srcs = [e.source for e in BatchLog.objects.filter(target=batchst, relation_type="merge").select_related("source")]
     if merge_srcs:
         bigs = set()
         for s in merge_srcs:
-            if s.zt_batch:
+            if s.zt_batch and s.zt_batch != s.batch:
                 bigs.add(s.zt_batch)
             else:
-                ps = _zt_split_parents(s)
+                t = _zt_climb_split_top(s)
-                bigs.add(ps.pop() if len(ps) == 1 else None)
+                if t:
-        if len(bigs) == 1 and None not in bigs:
+                    bigs.add(t)
+                # 无拆批谱系的来源(炉料链/改号链)不参与判定
+        if len(bigs) == 1:
             return bigs.pop()
+        if len(bigs) > 1:
             myLogger.error(f"直通统计-{batchst.batch}合批来源大批不一致{bigs}, 无法归属")
             return None
-    # 无任何拆合批上游: 自身即大批(未拆批直接检验)
+        # 来源均为炉料链: 合批改名诞生的大批未拆批直接检验, 自身即大批;
+        # 非大批格式(如不良集中池)不归属
+        if ZT_BIG_BATCH_RE.match(batchst.batch):
+            return batchst.batch
+        return None
+    # 无任何拆合批上游: 自身即大批(未拆批直接检验/外购直检)
     return batchst.batch
 
 

scripts/correct_zt_gxerp.py

@@ -26,31 +26,59 @@ import json
 from apps.utils.tools import MyJSONEncoder
 
 
-def main(since: str = "2026-06-01", fresh: bool = False):
+def main(since: str = "2025-06-17", fresh: bool = False):
+    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=datetime.strptime(since, "%Y-%m-%d"), version=1
-    ).filter(
+    ).filter(cond).order_by("create_time")
-        Q(data__has_key="尺寸检验_日期") | Q(data__has_key="外观检验_日期")
-    ).order_by("create_time")
     total = qs.count()
-    print(f"检验批 {total} 个, 开始解析大批归属")
+    print(f"检验批 {total} 个, 开始解析大批归属 (fresh={fresh})", flush=True)
 
     bigs = []  # 保序去重
     seen = set()
     for ind, bs in enumerate(qs.iterator()):
         try:
-            big = resolve_zt_big(bs, bs.data or {})
+            # 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"])
         except Exception as e:
-            print(f"[{ind + 1}/{total}] {bs.batch} 归属解析失败: {e}")
+            print(f"[{ind + 1}/{total}] {bs.batch} 归属解析失败: {e}", flush=True)
             continue
         if big and big not in seen:
             seen.add(big)
             bigs.append(big)
         if (ind + 1) % 1000 == 0:
-            print(f"归属解析 {ind + 1}/{total}")
+            print(f"归属解析 {ind + 1}/{total}", flush=True)
+
+    if fresh:
+        # 清理: 曾被错当成大批计算过直通字段、如今无任何子批指向的批,
+        # 剥离其直通键并解除自指锁定, 交由后续大批计算重新归属
+        cleaned = 0
+        for bs in BatchSt.objects.filter(
+            create_time__gte=datetime.strptime(since, "%Y-%m-%d"), version=1,
+            data__has_key="直通_良率",
+        ).iterator():
+            if bs.batch in seen:
+                continue
+            if BatchSt.objects.filter(zt_batch=bs.batch, version=1).exclude(id=bs.id).exists():
+                continue
+            data = bs.data
+            for k in ["直通_子批次", "直通_总合格数", "直通_白料数", "直通_良率", "直通_口径"]:
+                data.pop(k, None)
+            bs.data = json.loads(json.dumps(data, cls=MyJSONEncoder))
+            if bs.zt_batch == bs.batch:
+                bs.zt_batch = None
+                bs.save(update_fields=["data", "zt_batch", "update_time"])
+            else:
+                bs.save(update_fields=["data", "update_time"])
+            cleaned += 1
+            print(f"清理错误大批标记: {bs.batch}", flush=True)
+        print(f"清理 {cleaned} 个", flush=True)
 
     print(f"共 {len(bigs)} 个大批, 开始计算直通良率", flush=True)
     for ind, big in enumerate(bigs):