perf:correct_zt_gxerp两阶段改为线程池并发, 默认8线程
数据库IO密集, 各批/各大批独立无写冲突; --workers=N可调 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
d0c38b98f2
commit
dacb6470ab
|
|
@ -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('白料数')} "
|
||||
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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue