From d68ca98f2c2fd0a6cbfe8eedb4775117d6a5a897 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Tue, 14 Jul 2026 11:34:08 +0800 Subject: [PATCH] =?UTF-8?q?perf:=E5=A4=A7=E6=89=B9=E8=AE=A1=E7=AE=97?= =?UTF-8?q?=E6=8F=90=E9=80=9F=E4=B8=8E=E5=9B=9E=E5=88=B7=E8=84=9A=E6=9C=AC?= =?UTF-8?q?=E5=A2=9E=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 分子聚合改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 --- apps/wpm/scripts/batch_gxerp.py | 53 ++++++++++------- scripts/correct_zt_gxerp.py | 101 +++++++++++++++++--------------- 2 files changed, 86 insertions(+), 68 deletions(-) diff --git a/apps/wpm/scripts/batch_gxerp.py b/apps/wpm/scripts/batch_gxerp.py index fb749b37..a4b476cd 100644 --- a/apps/wpm/scripts/batch_gxerp.py +++ b/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): - """重算大批的直通良率并写入其 BatchSt.data""" +def cal_zt_big(big_batch: str, use_white_cache=True): + """重算大批的直通良率并写入其 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: - for e in BatchLog.objects.filter(source=node).select_related("target"): - t = e.target + merge_targets = {} + for e in BatchLog.objects.filter(source_id__in=frontier_ids).select_related("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: - 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: + pairs = src_by_target.get(tid, []) + if pairs and all(s_batch in known or s_zt == big_batch for s_batch, s_zt in pairs): known.add(t.batch) sub_map[t.batch] = t new_nodes.append(t) - frontier = new_nodes - if not frontier: + frontier_ids = [t.id for t in new_nodes] + 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): - sub_batches.append(sub.batch) - count_zt += (sub.data or {}).get("外观检验_直通合格数", 0) or 0 + for b, v in BatchSt.objects.filter(zt_batch=big_batch, version=1).exclude( + batch=big_batch).values_list("batch", "data__外观检验_直通合格数"): + 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(";")) - {"", "谱系异常(合格数大于白料数)"} diff --git a/scripts/correct_zt_gxerp.py b/scripts/correct_zt_gxerp.py index c796f225..d2601868 100644 --- a/scripts/correct_zt_gxerp.py +++ b/scripts/correct_zt_gxerp.py @@ -4,9 +4,9 @@ 不重建批次的其他统计数据(那个用 correct_batchst.py)。 用法: - python scripts/correct_zt_gxerp.py [起始日期] [--fresh] [--workers=N] - 起始日期默认 2025-06-17; --fresh 忽略旧锁重新解析归属并清缓存重算白料(纠错模式); - --workers 并发线程数, 默认 8 + python scripts/correct_zt_gxerp.py [起始日期] [--fresh] [--workers=N] [--skip-resolve] + 起始日期默认 2026-06-01; --fresh 忽略旧锁重新解析归属并强制重算白料(纠错模式); + --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): - 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"{bs.batch} 归属解析失败: {e}", flush=True) - return None - finally: - done[0] += 1 - if done[0] % 1000 == 0: + if skip_resolve: + # 直接用已落库的归属 + seen = set(BatchSt.objects.filter( + create_time__gte=since_dt, version=1, zt_batch__isnull=False, + ).values_list("zt_batch", flat=True).distinct()) + bigs = sorted(seen) + print(f"跳过归属解析, 已落库大批 {len(bigs)} 个 (workers={workers})", flush=True) + else: + 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) + + 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: - results = list(pool.map(resolve_one, items)) + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: + results = list(pool.map(resolve_one, items)) - bigs = [] # 保序去重 - seen = set() - for big in results: - if big and big not in seen: - seen.add(big) - bigs.append(big) + bigs = [] # 保序去重 + seen = set() + for big in results: + if big and big not in seen: + seen.add(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: - 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 {} + res = cal_zt_big(big, use_white_cache=not fresh) 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)