From 8f2d9ba2410c943d090d0b68f3931d14930a2072 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Mon, 13 Jul 2026 22:16:08 +0800 Subject: [PATCH] =?UTF-8?q?feat:=E6=89=B9=E6=AC=A1=E7=BB=9F=E8=AE=A1?= =?UTF-8?q?=E6=96=B0=E5=A2=9E=E5=A4=A7=E6=89=B9=E7=9B=B4=E9=80=9A=E8=89=AF?= =?UTF-8?q?=E7=8E=87(gxerp)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - BatchSt新增zt_batch字段锁定检验小批归属的大批 - batch_gxerp统计完成后自动解析归属并重算大批: 直通_白料数/直通_总合格数/直通_良率/直通_口径 - 白料数沿BatchLog回溯至黑化锚点取进炉领用数,多去向按生产 主线数量占比分摊(不良集中批旁路不占份额),外购兜底入库数量, 老批次沿BatchSt创建来源续接;白料结果缓存,分子超分母隔离 Co-Authored-By: Claude Fable 5 --- apps/wpm/migrations/0133_batchst_zt_batch.py | 18 ++ apps/wpm/models.py | 3 +- apps/wpm/scripts/batch_gxerp.md | 57 +++- apps/wpm/scripts/batch_gxerp.py | 291 ++++++++++++++++++- 4 files changed, 366 insertions(+), 3 deletions(-) create mode 100644 apps/wpm/migrations/0133_batchst_zt_batch.py diff --git a/apps/wpm/migrations/0133_batchst_zt_batch.py b/apps/wpm/migrations/0133_batchst_zt_batch.py new file mode 100644 index 00000000..26567d12 --- /dev/null +++ b/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='直通统计大批号'), + ), + ] diff --git a/apps/wpm/models.py b/apps/wpm/models.py index 21e02f4a..92d732f3 100644 --- a/apps/wpm/models.py +++ b/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) diff --git a/apps/wpm/scripts/batch_gxerp.md b/apps/wpm/scripts/batch_gxerp.md index 4ecc226b..ffea4457 100644 --- a/apps/wpm/scripts/batch_gxerp.md +++ b/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` 末尾自动解析归属并重算大批的 `直通_白料数 / 直通_总合格数 / 直通_良率 / 直通_口径`。 diff --git a/apps/wpm/scripts/batch_gxerp.py b/apps/wpm/scripts/batch_gxerp.py index 522dd0fc..9d01eb6c 100644 --- a/apps/wpm/scripts/batch_gxerp.py +++ b/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 \ No newline at end of file