90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
from apps.wpm.models import Mlog, BatchSt, Handover
|
|
from apps.mtm.models import Mgroup
|
|
from apps.system.models import Dept
|
|
from apps.inm.models import MIOItem
|
|
from apps.qm.models import FtestWork
|
|
from datetime import datetime
|
|
from django.conf import settings
|
|
import json
|
|
from apps.utils.tools import MyJSONEncoder
|
|
import decimal
|
|
from apps.utils.thread import MyThread
|
|
import logging
|
|
import time
|
|
from zoneinfo import ZoneInfo
|
|
import importlib
|
|
|
|
myLogger = logging.getLogger('log')
|
|
tz_shanghai = ZoneInfo("Asia/Shanghai")
|
|
|
|
# 批次统计分析
|
|
|
|
def ana_batch_thread(xbatchs: list, mgroup=None):
|
|
MyThread(target=ana_batch, args=(xbatchs, mgroup)).start()
|
|
|
|
def ana_wpr_thread(wprIds: list, mgroup):
|
|
MyThread(target=ana_wpr, args=(wprIds, mgroup)).start()
|
|
|
|
def ana_batch(xbatchs: list, mgroup):
|
|
"""
|
|
批次统计分析
|
|
"""
|
|
time.sleep(10)
|
|
xbatchs = list(set(xbatchs))
|
|
BASE_PROJECT_CODE = getattr(settings, "BASE_PROJECT_CODE", None)
|
|
try:
|
|
m = importlib.import_module(f"apps.wpm.scripts.batch_{BASE_PROJECT_CODE}")
|
|
except ModuleNotFoundError:
|
|
# myLogger.error(f"ana_batch {BASE_PROJECT_CODE} not found")
|
|
return
|
|
f = getattr(m, "main")
|
|
for xbatch in xbatchs:
|
|
f(xbatch, mgroup)
|
|
|
|
def ana_wpr(wprIds: list, mgroup):
|
|
"""
|
|
单个统计分析
|
|
"""
|
|
time.sleep(10)
|
|
wprIds = list(set(wprIds))
|
|
BASE_PROJECT_CODE = getattr(settings, "BASE_PROJECT_CODE", None)
|
|
try:
|
|
m = importlib.import_module(f"apps.wpm.scripts.wpr_{BASE_PROJECT_CODE}")
|
|
except ModuleNotFoundError:
|
|
return
|
|
f = getattr(m, "main")
|
|
for wprId in wprIds:
|
|
f(wprId, mgroup)
|
|
|
|
|
|
def get_f_l_date(data):
|
|
first_date = None
|
|
last_date = None
|
|
for k, v in data.items():
|
|
if k.endswith("_日期"):
|
|
if v:
|
|
if isinstance(v, list):
|
|
myLogger.error(f"get_f_l_date {k} {v}")
|
|
continue
|
|
dates = []
|
|
for s in v.split(";"):
|
|
s = s.strip()
|
|
if not s:
|
|
continue
|
|
try:
|
|
dates.append(datetime.strptime(s, "%Y-%m-%d").date())
|
|
except ValueError:
|
|
myLogger.error(f"get_f_l_date invalid date {k} {s}")
|
|
if not dates:
|
|
continue
|
|
cur_min = min(dates)
|
|
cur_max = max(dates)
|
|
if first_date is None or cur_min < first_date:
|
|
first_date = cur_min
|
|
if last_date is None or cur_max > last_date:
|
|
last_date = cur_max
|
|
return {"first_date": first_date.strftime("%Y-%m-%d") if first_date else None,
|
|
"last_date": last_date.strftime("%Y-%m-%d") if last_date else None,
|
|
"first_time": datetime.combine(first_date, datetime.min.time()).replace(tzinfo=tz_shanghai) if first_date else None,
|
|
"last_time": datetime.combine(last_date, datetime.max.time().replace(microsecond=0)).replace(tzinfo=tz_shanghai) if last_date else None}
|