76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""大批直通良率专用回刷脚本(gxerp)
|
|
|
|
只解析检验批的大批归属(BatchSt.zt_batch)并重算各大批的 直通_* 统计,
|
|
不重建批次的其他统计数据(那个用 correct_batchst.py)。
|
|
|
|
用法:
|
|
python scripts/correct_zt_gxerp.py [起始日期] [--fresh]
|
|
起始日期默认 2025-06-17; --fresh 清除大批已缓存的白料数, 强制重新回溯
|
|
"""
|
|
import os
|
|
import sys
|
|
import django
|
|
|
|
CUR_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
BASE_DIR = os.path.dirname(CUR_DIR)
|
|
sys.path.insert(0, BASE_DIR)
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "server.settings")
|
|
django.setup()
|
|
|
|
from datetime import datetime
|
|
from django.db.models import Q
|
|
from apps.wpm.models import BatchSt
|
|
from apps.wpm.scripts.batch_gxerp import resolve_zt_big, cal_zt_big
|
|
import json
|
|
from apps.utils.tools import MyJSONEncoder
|
|
|
|
|
|
def main(since: str = "2025-06-17", fresh: bool = False):
|
|
qs = BatchSt.objects.filter(
|
|
create_time__gte=datetime.strptime(since, "%Y-%m-%d"), version=1
|
|
).filter(
|
|
Q(data__has_key="尺寸检验_日期") | Q(data__has_key="外观检验_日期")
|
|
).order_by("create_time")
|
|
total = qs.count()
|
|
print(f"检验批 {total} 个, 开始解析大批归属")
|
|
|
|
bigs = [] # 保序去重
|
|
seen = set()
|
|
for ind, bs in enumerate(qs.iterator()):
|
|
try:
|
|
big = resolve_zt_big(bs, bs.data or {})
|
|
if big and bs.zt_batch != big:
|
|
bs.zt_batch = big
|
|
bs.save(update_fields=["zt_batch"])
|
|
except Exception as e:
|
|
print(f"[{ind + 1}/{total}] {bs.batch} 归属解析失败: {e}")
|
|
continue
|
|
if big and big not in seen:
|
|
seen.add(big)
|
|
bigs.append(big)
|
|
if (ind + 1) % 1000 == 0:
|
|
print(f"归属解析 {ind + 1}/{total}")
|
|
|
|
print(f"共 {len(bigs)} 个大批, 开始计算直通良率")
|
|
for ind, big in enumerate(bigs):
|
|
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"])
|
|
cal_zt_big(big)
|
|
except Exception as e:
|
|
print(f"{big} 计算失败: {e}")
|
|
if (ind + 1) % 200 == 0:
|
|
print(f"大批 {ind + 1}/{len(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)
|