feat:批次统计新增大批直通良率(gxerp)

- BatchSt新增zt_batch字段锁定检验小批归属的大批
- batch_gxerp统计完成后自动解析归属并重算大批:
  直通_白料数/直通_总合格数/直通_良率/直通_口径
- 白料数沿BatchLog回溯至黑化锚点取进炉领用数,多去向按生产
  主线数量占比分摊(不良集中批旁路不占份额),外购兜底入库数量,
  老批次沿BatchSt创建来源续接;白料结果缓存,分子超分母隔离

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

apps/wpm/migrations/0133_batchst_zt_batch.py

@@ -0,0 +1,18 @@
+# Generated by Django 4.2.27 on 2026-07-13 10:36
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('wpm', '0132_mlogbw_tooling_alter_mlogbw_equip'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='batchst',
+            name='zt_batch',
+            field=models.TextField(blank=True, db_index=True, null=True, verbose_name='直通统计大批号'),
+        ),
+    ]

apps/wpm/models.py

@@ -787,7 +787,8 @@ class BatchSt(BaseModel):
     TN: 产品批次统计
     """
     batch = models.TextField("批次号", db_index=True)
-    version = models.IntegerField("版本号", default=1, db_index=True) 
+    version = models.IntegerField("版本号", default=1, db_index=True)
+    zt_batch = models.TextField("直通统计大批号", null=True, blank=True, db_index=True)  # 锁定该批归属的大批(拆出检验小批的批), 由批次统计脚本维护
     first_time = models.DateTimeField("首次操作时间", null=True, blank=True)
     last_time = models.DateTimeField("最后操作时间", null=True, blank=True)
     data = models.JSONField("数据", default=dict, blank=True)

apps/wpm/scripts/batch_gxerp.md

@@ -209,8 +209,63 @@ FtestWork where type2 = TYPE2_SOME
 
 1. **`mgroup_obj` 入参未使用**：当前实现忽略该参数，对所有 `Mgroup` 全量遍历。
 
-## 十、修订记录
+---
+
+## 十、大批直通良率（`main` 末尾自动触发）
+
+### 定义
+
+- **大批** = 拆出检验小批的那个批（如 `2605-BJ-0023`，退火后合批改名产生；外购半成品则为采购入库批）。由 `BatchLog` 拆合批关系确定，不依赖批次号命名和工艺路线。
+- **直通良率** = 大批下所有子批次 `外观检验_直通合格数` 之和 ÷ 大批白料数（进炉前数量）× 100。
+
+### 归属锁定（`BatchSt.zt_batch`）
+
+批次统计完成后（`handle_zt`）：
+
+1. 当前批 `data` 含 `尺寸检验_count_real` / `外观检验_count_real` 时解析归属（`resolve_zt_big`）：
+   - 唯一拆批来源（handover split）→ 该来源即大批；
+   - 合批复检批（如 `大批-A-1`）→ 所有合批来源归属同一大批时继承；不一致则记日志不归属；
+   - 无任何拆合批上游 → 自身即大批（未拆批直接检验、外购直检）。
+2. 归属写入 `zt_batch`（锁定，已锁定的直接复用）；随后重算大批（`cal_zt_big`）。
+3. 大批自身重算时（有子批 `zt_batch` 指向它）也会触发重算。
+
+### 分子
+
+`BatchSt.filter(zt_batch=大批)` 各批 `外观检验_直通合格数` 之和（+大批自身若有）。子批发现从大批侧沿拆批边/同大批合批边展开（`cal_zt_big`，支持 `-A-1` 复检批、多级拆分）。检验前合并的子批无检验数据，天然不重复计数。
+
+### 分母（白料数，`_zt_white_count` 回溯）
+
+从大批沿 `BatchLog` 向上回溯（限深 `ZT_MAX_DEPTH=12`，按路径防环，跳过合并回自身的整理性边）：
+
+- **锚点**：批次存在锚点工段（`ZT_ANCHOR_MGROUPS=["黑化"]`）产出报工时停止，白料数 = 该批锚点工段 `mlogb_from.count_use`（进炉前领用数）。多次黑化的链取离大批最近的一次。
+- **分摊**（`_zt_source_out_ratio`）：来源批有多条**生产主线**流出边（报工改号/拆批/汇入大批格式 `ZT_BIG_BATCH_RE` 的合批）时，按各去向数量占比分摊；汇入不良集中批（如 `黑检片-00`）的旁路不占白料份额，使炉后不良计入生产主线分母。单一去向全额传递。
+- **边数量**：拆批取 `Handoverb(batch=目标批)`，合批取 `Handoverb(batch=来源批)`，报工改号取目标产出行的 `mlogb_from.count_use`；无明细时兜底整单数量。
+- **外购兜底**：无上游谱系且由入库创建 → 白料数 = 入库数量（`MIOItem.count`），口径记 `入库数量`。
+- **老数据兜底**（`_zt_white_legacy`）：BatchLog 上线前的批次无边，沿 `BatchSt.handover/mlog` 创建来源续上回溯，口径记 `谱系兜底`。
+
+### 缓存与隔离
+
+- 白料数算出后缓存在大批 `data`（含算不出的负缓存）；子批触发的重算直接复用、只重聚合分子。大批自身重算时 `data` 重建、缓存失效并重新回溯。上游补改报工后需大批重算或全量回刷才会刷新。
+- 总合格数 > 白料数（如白料池循环整理导致回溯不全）→ `直通_良率` 置空并在口径标注 `谱系异常`，记错误日志（2026-06 抽样 120 个大批：116 正常 / 2 异常隔离 / 2 谱系断头不可得）。
+
+### 输出字段（写入大批 `data`）
+
+| Key | 含义 |
+| --- | --- |
+| `直通_子批次` | 归属本大批的子批号 `;` 拼接 |
+| `直通_总合格数` | 分子 |
+| `直通_白料数` | 分母（不可得为 null） |
+| `直通_良率` | 百分比（不可得/异常为 null） |
+| `直通_口径` | 白料来源口径：`黑化领用` / `入库数量` / `谱系兜底` 及异常标注 |
+
+### 部署
+
+1. `python manage.py migrate wpm`（新增 `BatchSt.zt_batch`，迁移 `0133_batchst_zt_batch`）。
+2. 历史回刷：`python scripts/correct_batchst.py`（对有检验数据的批重跑 `main`，自动完成归属与大批计算）。
+
+## 十一、修订记录
 
 - 修复 `小日期 / 大日期` 命名与含义反向，`小日期` 取最早日期、`大日期` 取最晚日期；同步修正 `batch_bxerp.py`、`batch_gzerp.py` 中所有同类字段。
 - `get_f_l_date` 改为先 `datetime.strptime` 解析为 `date` 对象再比较，移除对字符串字典序的隐式依赖；非法日期片段记录日志后跳过。
 - 所有合格率/直通率计算的兜底 `except` 同时捕获 `decimal.InvalidOperation` 与 `ZeroDivisionError`，以兼容 `Decimal` 与原生数值的除零场景；`batch_bxerp.py` 与 `batch_gzerp.py` 中仅捕获 `decimal.InvalidOperation` 的合格率分支也同步扩展。
+- 新增大批直通良率统计（见第十节）：`BatchSt` 加 `zt_batch` 归属锁定字段，`main` 末尾自动解析归属并重算大批的 `直通_白料数 / 直通_总合格数 / 直通_良率 / 直通_口径`。

apps/wpm/scripts/batch_gxerp.py

@@ -1,9 +1,12 @@
-from apps.wpm.models import BatchSt
+from apps.wpm.models import BatchSt, BatchLog, Handoverb
 import logging
+import re
 from apps.qm.models import Defect, FtestWork, FtestworkDefect
 from apps.wpm.models import Mlogb, MlogbDefect, Mlog
 from apps.mtm.models import Mgroup
+from apps.inm.models import MIOItem
 import decimal
+from decimal import Decimal
 from django.db.models import Sum
 from datetime import datetime
 from apps.wpm.services_2 import get_f_l_date
@@ -11,6 +14,16 @@ import json
 from apps.utils.tools import MyJSONEncoder
 myLogger = logging.getLogger("log")
 
+# ==================== 大批直通良率统计 ====================
+# 大批 = 拆出检验小批的那个批(由拆批交接关系确定, 不依赖批次号命名/工艺路线)
+# 直通良率 = 大批下所有子批次外观检验直通合格数之和 / 大批白料数
+# 白料数(入料数) = 沿拆合批关系(BatchLog)从大批向上回溯, 找到锚点工段产出批,
+#   取其锚点工段领用数(即进炉前数量); 一条炉料链拆给多个大批时按生产主线流出数量占比分摊
+#   (汇入不良集中批的旁路不占份额), 占比截断为1; 无锚点谱系(如外购半成品)时兜底用入库数量。
+ZT_ANCHOR_MGROUPS = ["黑化"]  # 入料数锚点工段
+ZT_MAX_DEPTH = 12  # 白料回溯最大层数
+ZT_BIG_BATCH_RE = re.compile(r"^\d{4}-[0-9A-Za-z.]+-\d+")  # 大批批号格式(合批改名产生), 用于区分生产分流与不良汇集
+
 def main(batch: str, mgroup_obj):
     try:
         batchst = BatchSt.objects.get(batch=batch, version=1)
@@ -190,5 +203,281 @@ def main(batch: str, mgroup_obj):
         batchst.last_time = res["last_time"]
     batchst.save()
 
+    try:
+        handle_zt(batchst, data)
+    except Exception:
+        myLogger.exception(f"直通统计-{batch}计算失败")
+
+
+def handle_zt(batchst: BatchSt, data: dict):
+    """确定当前批归属的大批并重算该大批的直通良率"""
+    big_batch = resolve_zt_big(batchst, data)
+    if big_batch and batchst.zt_batch != big_batch:
+        batchst.zt_batch = big_batch
+        batchst.save(update_fields=["zt_batch"])
+    if big_batch is None and BatchSt.objects.filter(zt_batch=batchst.batch, version=1).exclude(id=batchst.id).exists():
+        # 当前批自身是大批(已有子批归属于它), 其数据变化(如白料数)也需重算
+        big_batch = batchst.batch
+    if big_batch:
+        cal_zt_big(big_batch)
+
+
+def _zt_split_parents(batchst: BatchSt):
+    """经拆批交接产生该批的来源批次号集合"""
+    qs = BatchLog.objects.filter(target=batchst, relation_type="split", handover__isnull=False).select_related("source")
+    return {e.source.batch for e in qs}
+
+
+def resolve_zt_big(batchst: BatchSt, data: dict):
+    """确定检验小批所属的大批号, 无法确定返回 None"""
+    if 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)
+    if len(parents) == 1:
+        return parents.pop()
+    if len(parents) > 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:
+                bigs.add(s.zt_batch)
+            else:
+                ps = _zt_split_parents(s)
+                bigs.add(ps.pop() if len(ps) == 1 else None)
+        if len(bigs) == 1 and None not in bigs:
+            return bigs.pop()
+        myLogger.error(f"直通统计-{batchst.batch}合批来源大批不一致{bigs}, 无法归属")
+        return None
+    # 无任何拆合批上游: 自身即大批(未拆批直接检验)
+    return batchst.batch
+
+
+def _zt_edge_qty(edge: BatchLog, target_batch: str):
+    """该条拆合批边流入 target 的数量"""
+    if edge.handover_id:
+        # 拆批明细行记目标批, 合批明细行记来源批
+        hb_batch = target_batch if edge.relation_type == "split" else edge.source.batch
+        qty = Handoverb.objects.filter(handover_id=edge.handover_id, batch=hb_batch).aggregate(t=Sum("count"))["t"]
+        if qty:
+            return qty
+        return edge.handover.count  # 无明细行时兜底整单数量
+    if edge.mlog_id:
+        # 报工改号: 目标批产出行对应的领用数
+        row = Mlogb.objects.filter(mlog_id=edge.mlog_id, batch=target_batch,
+                                   material_out__isnull=False).select_related("mlogb_from").first()
+        if row:
+            return row.mlogb_from.count_use if row.mlogb_from else row.count_real
+    return None
+
+
+def _zt_source_out_ratio(edge: BatchLog, qty):
+    """qty 占来源批生产主线流出量的比例(白料分摊系数)
+
+    生产主线流出 = 报工改号边 / 拆批边 / 汇入大批格式批号的合批边(即当前回溯所走的这类边);
+    汇入不良集中批(如 黑检片-00)等旁路的合批边不占白料份额 —— 炉后不良对应的白料
+    应计入生产主线大批的分母, 直通良率才能体现炉段损耗。
+    合并回自身的整理性合批边(source==target)不算流出。
+    只有一条生产主线流出边时全额传递(ratio=1); 多条(如炉料链拆给多个大批)按数量占比分摊。
+    """
+    out_edges = [e for e in BatchLog.objects.filter(source=edge.source).select_related("target", "handover")
+                 if e.target.batch != e.source.batch]
+    prod_edges = [e for e in out_edges
+                  if e.mlog_id or e.relation_type == "split" or ZT_BIG_BATCH_RE.match(e.target.batch) or e.id == edge.id]
+    if len(prod_edges) <= 1:
+        return Decimal(1)
+    total = Decimal(0)
+    for e in prod_edges:
+        e_qty = _zt_edge_qty(e, e.target.batch)
+        if e_qty:
+            total += Decimal(e_qty)
+        else:
+            myLogger.error(f"直通统计-{e.source.batch}->{e.target.batch}流出边数量缺失, 分摊占比可能偏大")
+    if not total:
+        return Decimal(1)
+    return min(Decimal(qty) / total, Decimal(1))
+
+
+def _zt_white_count(batchst: BatchSt, depth=0, path=frozenset()):
+    """回溯计算该批对应的白料数(进炉前数量)
+
+    返回 (白料数, 口径set); 不可得时白料数为 None。
+    分摊: 来源批白料 p_white 按 qty 占其全部流出量的比例分给当前批。
+    """
+    if depth > ZT_MAX_DEPTH:
+        return None, {"超出回溯深度"}
+    path = path | {batchst.batch}
+
+    # 锚点: 该批有锚点工段产出报工, 白料数=锚点工段领用数
+    anchor_qs = Mlogb.objects.filter(
+        batch=batchst.batch, material_out__isnull=False, need_inout=True,
+        mlog__mgroup__name__in=ZT_ANCHOR_MGROUPS, mlog__submit_time__isnull=False,
+        mlog__is_fix=False).select_related("mlogb_from", "mlog__mgroup")
+    white = Decimal(0)
+    kinds = set()
+    found_anchor = False
+    for row in anchor_qs:
+        found_anchor = True
+        white += row.mlogb_from.count_use if row.mlogb_from else row.count_real
+        kinds.add(f"{row.mlog.mgroup.name}领用")
+    if found_anchor:
+        return white, kinds
+
+    edges = [e for e in BatchLog.objects.filter(target=batchst).select_related("source", "handover")
+             if e.source.batch not in path]  # 跳过自环/已访问节点(同名重复改号、合回上游等)
+    if not edges:
+        # 无上游谱系: 入库创建的批(如外购半成品)兜底用入库数量
+        mioitem = batchst.mioitem
+        if mioitem is None and batchst.mio_id:
+            mioitem = MIOItem.objects.filter(mio_id=batchst.mio_id, batch=batchst.batch).first()
+        if mioitem and mioitem.count:
+            return mioitem.count, {"入库数量"}
+        # BatchLog 上线前的老批次没有边, 但 BatchSt 记录了创建它的交接/报工, 由此续上回溯
+        legacy = _zt_white_legacy(batchst, depth, path)
+        if legacy is not None:
+            return legacy
+        return None, {f"{batchst.batch}无上游谱系且无入库数量"}
+
+    white_total = Decimal(0)
+    for edge in edges:
+        qty = _zt_edge_qty(edge, batchst.batch)
+        if not qty:
+            return None, {f"{edge.source.batch}->{batchst.batch}边数量缺失"}
+        p_white, p_kinds = _zt_white_count(edge.source, depth + 1, path)
+        if p_white is None:
+            return None, p_kinds
+        white_total += Decimal(p_white) * _zt_source_out_ratio(edge, qty)
+        kinds |= p_kinds
+    return white_total, kinds
+
+
+def _zt_white_legacy(batchst: BatchSt, depth, path):
+    """BatchLog 上线前的老批次没有拆合批边, 根据 BatchSt 记录的创建来源续上回溯
+
+    可解时返回 (白料数, 口径set), 否则返回 None。
+    """
+    src_batch = None
+    qty = None
+    ratio = Decimal(1)
+    if batchst.handover_id and batchst.handover.wm_id:
+        # 拆批交接创建: 来源为交接的车间库存批, 按同单各行数量占比分摊
+        src_batch = batchst.handover.wm.batch
+        own = Handoverb.objects.filter(handover_id=batchst.handover_id, batch=batchst.batch).aggregate(t=Sum("count"))["t"]
+        total = Handoverb.objects.filter(handover_id=batchst.handover_id).aggregate(t=Sum("count"))["t"]
+        qty = own or batchst.handover.count
+        if own and total and total > own:
+            ratio = Decimal(own) / Decimal(total)
+    elif batchst.mlog_id:
+        # 报工改号创建: 来源为报工的投入批
+        row = Mlogb.objects.filter(mlog_id=batchst.mlog_id, batch=batchst.batch,
+                                   material_out__isnull=False).select_related("mlogb_from").first()
+        if row and row.mlogb_from:
+            src_batch = row.mlogb_from.batch
+            qty = row.mlogb_from.count_use
+    if not src_batch or src_batch in path or not qty:
+        return None
+    try:
+        src_bs = BatchSt.objects.get(batch=src_batch, version=1)
+    except BatchSt.DoesNotExist:
+        return None
+    p_white, p_kinds = _zt_white_count(src_bs, depth + 1, path)
+    if p_white is None:
+        return None
+    return p_white * ratio, p_kinds | {"谱系兜底"}
+
+
+def cal_zt_big(big_batch: str):
+    """重算大批的直通良率并写入其 BatchSt.data"""
+    try:
+        big_bs = BatchSt.objects.get(batch=big_batch, version=1)
+    except BatchSt.DoesNotExist:
+        myLogger.error(f"直通统计-大批{big_batch}不存在")
+        return
+    if big_bs.zt_batch != big_batch:
+        big_bs.zt_batch = big_batch
+        big_bs.save(update_fields=["zt_batch"])
+
+    # 从大批侧发现子批: 拆批目标 + 完全由本大批子批合并出的复检批(如 大批-A-1), 支持多级
+    known = {big_batch}
+    sub_map = {}
+    frontier = [big_bs]
+    for _ in range(3):
+        new_nodes = []
+        for node in frontier:
+            for e in BatchLog.objects.filter(source=node).select_related("target"):
+                t = e.target
+                if t.batch in known:
+                    continue
+                if e.relation_type == "split" and e.handover_id:
+                    ok = True
+                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
+        if not frontier:
+            break
+    # 锁定归属
+    for b, t in sub_map.items():
+        if t.zt_batch and t.zt_batch != big_batch:
+            myLogger.error(f"直通统计-{b}已归属{t.zt_batch}, 与{big_batch}冲突, 跳过")
+            continue
+        if t.zt_batch != big_batch:
+            t.zt_batch = big_batch
+            t.save(update_fields=["zt_batch"])
+
+    # 分子: 所有归属本大批的批次(检验小批/复检批)外观检验直通合格数之和
+    count_zt = 0
+    sub_batches = []
+    for sub in BatchSt.objects.filter(zt_batch=big_batch, version=1).exclude(batch=big_batch):
+        sub_batches.append(sub.batch)
+        count_zt += (sub.data or {}).get("外观检验_直通合格数", 0) or 0
+    # 大批未拆批直接检验时自身也计入
+    count_zt += (big_bs.data or {}).get("外观检验_直通合格数", 0) or 0
+
+    # 分母: 白料数(大批成型后基本不变, 已算过的直接复用, 含算不出的负缓存;
+    # 大批自身重算时 main 会重建 data, 缓存自然失效并重新回溯)
+    data = big_bs.data or {}
+    if "直通_白料数" in data:
+        white = data.get("直通_白料数")
+        white = Decimal(str(white)) if white else None
+        kinds = set((data.get("直通_口径") or "").split(";")) - {"", "谱系异常(合格数大于白料数)"}
+    else:
+        white, kinds = _zt_white_count(big_bs)
+        if not white:
+            myLogger.error(f"直通统计-{big_batch}白料数不可得: {kinds}")
+
+    data["直通_子批次"] = ";".join(sorted(sub_batches))
+    data["直通_总合格数"] = count_zt
+    data["直通_口径"] = ";".join(sorted(kinds))
+    if white and Decimal(str(count_zt)) > white:
+        # 分子超过分母说明谱系/分摊异常(如白料池循环整理导致回溯不全), 隔离不出良率
+        myLogger.error(f"直通统计-{big_batch}总合格数{count_zt}大于白料数{white}, 谱系或分摊异常")
+        data["直通_白料数"] = round(white, 1)
+        data["直通_良率"] = None
+        data["直通_口径"] = ";".join(sorted(kinds | {"谱系异常(合格数大于白料数)"}))
+    elif white:
+        data["直通_白料数"] = round(white, 1)
+        try:
+            data["直通_良率"] = round((Decimal(str(count_zt)) / white) * 100, 2)
+        except (decimal.InvalidOperation, ZeroDivisionError):
+            data["直通_良率"] = 0
+    else:
+        data["直通_白料数"] = None
+        data["直通_良率"] = None
+    big_bs.data = json.loads(json.dumps(data, cls=MyJSONEncoder))
+    big_bs.save(update_fields=["data"])
+
+
 if __name__ == '__main__':
     pass