From dacb6470ab6c43d99ad4601689eec841b8de8ae8 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Tue, 14 Jul 2026 11:11:32 +0800 Subject: [PATCH] =?UTF-8?q?perf:correct=5Fzt=5Fgxerp=E4=B8=A4=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=E6=94=B9=E4=B8=BA=E7=BA=BF=E7=A8=8B=E6=B1=A0=E5=B9=B6?= =?UTF-8?q?=E5=8F=91,=20=E9=BB=98=E8=AE=A48=E7=BA=BF=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 数据库IO密集, 各批/各大批独立无写冲突; --workers=N可调 Co-Authored-By: Claude Fable 5 --- scripts/correct_zt_gxerp.py | 80 ++++++++++++++++++++++++++----------- 1 file changed, 57 insertions(+), 23 deletions(-) diff --git a/scripts/correct_zt_gxerp.py b/scripts/correct_zt_gxerp.py index cc1d4da5..c796f225 100644 --- a/scripts/correct_zt_gxerp.py +++ b/scripts/correct_zt_gxerp.py @@ -4,8 +4,9 @@ 不重建批次的其他统计数据(那个用 correct_batchst.py)。 用法: - python scripts/correct_zt_gxerp.py [起始日期] [--fresh] - 起始日期默认 2025-06-17; --fresh 清除大批已缓存的白料数, 强制重新回溯 + python scripts/correct_zt_gxerp.py [起始日期] [--fresh] [--workers=N] + 起始日期默认 2025-06-17; --fresh 忽略旧锁重新解析归属并清缓存重算白料(纠错模式); + --workers 并发线程数, 默认 8 """ import os import sys @@ -18,6 +19,8 @@ sys.path.insert(0, BASE_DIR) os.environ.setdefault("DJANGO_SETTINGS_MODULE", "server.settings") django.setup() +import concurrent.futures +import threading from datetime import datetime from django.db.models import Q from apps.wpm.models import BatchSt @@ -25,43 +28,56 @@ from apps.wpm.scripts.batch_gxerp import resolve_zt_big, cal_zt_big import json from apps.utils.tools import MyJSONEncoder +_print_lock = threading.Lock() -def main(since: str = "2025-06-17", fresh: bool = False): + +def main(since: str = "2025-06-17", fresh: bool = False, workers: int = 8): + 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=datetime.strptime(since, "%Y-%m-%d"), version=1 - ).filter(cond).order_by("create_time") - total = qs.count() - print(f"检验批 {total} 个, 开始解析大批归属 (fresh={fresh})", flush=True) + 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) - bigs = [] # 保序去重 - seen = set() - for ind, bs in enumerate(qs.iterator()): + 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: - print(f"[{ind + 1}/{total}] {bs.batch} 归属解析失败: {e}", flush=True) - continue + with _print_lock: + 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)) + + bigs = [] # 保序去重 + seen = set() + for big in results: if big and big not in seen: seen.add(big) bigs.append(big) - if (ind + 1) % 1000 == 0: - 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="直通_良率", + create_time__gte=since_dt, version=1, data__has_key="直通_良率", ).iterator(): if bs.batch in seen: continue @@ -81,7 +97,9 @@ def main(since: str = "2025-06-17", fresh: bool = False): print(f"清理 {cleaned} 个", flush=True) print(f"共 {len(bigs)} 个大批, 开始计算直通良率", flush=True) - for ind, big in enumerate(bigs): + done[0] = 0 + + def cal_one(big): try: if fresh: big_bs = BatchSt.objects.filter(batch=big, version=1).first() @@ -91,13 +109,29 @@ def main(since: str = "2025-06-17", fresh: bool = False): big_bs.data = json.loads(json.dumps(data, cls=MyJSONEncoder)) big_bs.save(update_fields=["data", "update_time"]) res = cal_zt_big(big) or {} - print(f"[{ind + 1}/{len(bigs)}] {big} 良率={res.get('良率')} 白料={res.get('白料数')} " - f"合格={res.get('总合格数')} 子批={res.get('子批数')} 口径={res.get('口径')}", flush=True) + done[0] += 1 + with _print_lock: + print(f"[{done[0]}/{len(bigs)}] {big} 良率={res.get('良率')} 白料={res.get('白料数')} " + f"合格={res.get('总合格数')} 子批={res.get('子批数')} 口径={res.get('口径')}", flush=True) except Exception as e: - print(f"[{ind + 1}/{len(bigs)}] {big} 计算失败: {e}", flush=True) + done[0] += 1 + with _print_lock: + print(f"[{done[0]}/{len(bigs)}] {big} 计算失败: {e}", flush=True) + + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: + list(pool.map(cal_one, 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) + fresh = "--fresh" in sys.argv + workers = 8 + args = [] + for a in sys.argv[1:]: + if a == "--fresh": + 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)