feat:新增直通良率专用回刷脚本correct_zt_gxerp

只解析zt_batch归属并重算大批直通_*字段,每个大批仅算一次,
不重建批次其他统计;--fresh可强制重新回溯白料数

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

apps/wpm/scripts/batch_gxerp.md

@@ -261,7 +261,7 @@ FtestWork where type2 = TYPE2_SOME
 ### 部署
 
 1. `python manage.py migrate wpm`（新增 `BatchSt.zt_batch`，迁移 `0133_batchst_zt_batch`）。
-2. 历史回刷：`python scripts/correct_batchst.py`（对有检验数据的批重跑 `main`，自动完成归属与大批计算）。
+2. 历史回刷：`python scripts/correct_zt_gxerp.py [起始日期] [--fresh]`——只解析归属并重算各大批 `直通_*` 字段（每个大批仅算一次），不重建批次其他统计；`--fresh` 强制清缓存重新回溯白料。全量重建统计仍用 `scripts/correct_batchst.py`（其 `main` 末尾也会触发直通计算）。
 
 ## 十一、修订记录
 

scripts/correct_zt_gxerp.py

@@ -0,0 +1,75 @@
+"""大批直通良率专用回刷脚本(gxerp)
+
+只解析检验批的大批归属(BatchSt.zt_batch)并重算各大批的 直通_* 统计,
+不重建批次的其他统计数据(那个用 correct_batchst.py)。
+
+用法:
+    python scripts/correct_zt_gxerp.py [起始日期] [--fresh]
+    起始日期默认 2025-06-17; --fresh 清除大批已缓存的白料数, 强制重新回溯
+"""
+import os
+import sys
+import django
+
+CUR_DIR = os.path.dirname(os.path.abspath(__file__))
+BASE_DIR = os.path.dirname(CUR_DIR)
+sys.path.insert(0, BASE_DIR)
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "server.settings")
+django.setup()
+
+from datetime import datetime
+from django.db.models import Q
+from apps.wpm.models import BatchSt
+from apps.wpm.scripts.batch_gxerp import resolve_zt_big, cal_zt_big
+import json
+from apps.utils.tools import MyJSONEncoder
+
+
+def main(since: str = "2025-06-17", fresh: bool = False):
+    qs = BatchSt.objects.filter(
+        create_time__gte=datetime.strptime(since, "%Y-%m-%d"), version=1
+    ).filter(
+        Q(data__has_key="尺寸检验_日期") | Q(data__has_key="外观检验_日期")
+    ).order_by("create_time")
+    total = qs.count()
+    print(f"检验批 {total} 个, 开始解析大批归属")
+
+    bigs = []  # 保序去重
+    seen = set()
+    for ind, bs in enumerate(qs.iterator()):
+        try:
+            big = resolve_zt_big(bs, bs.data or {})
+            if big and bs.zt_batch != big:
+                bs.zt_batch = big
+                bs.save(update_fields=["zt_batch"])
+        except Exception as e:
+            print(f"[{ind + 1}/{total}] {bs.batch} 归属解析失败: {e}")
+            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"共 {len(bigs)} 个大批, 开始计算直通良率")
+    for ind, big in enumerate(bigs):
+        try:
+            if fresh:
+                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"])
+            cal_zt_big(big)
+        except Exception as e:
+            print(f"{big} 计算失败: {e}")
+        if (ind + 1) % 200 == 0:
+            print(f"大批 {ind + 1}/{len(bigs)}")
+    print("done")
+
+
+if __name__ == "__main__":
+    args = [a for a in sys.argv[1:] if a != "--fresh"]
+    main(since=args[0] if args else "2025-06-17", fresh="--fresh" in sys.argv)