330 lines
12 KiB
Python
330 lines
12 KiB
Python
from rest_framework.mixins import ListModelMixin
|
|
from rest_framework.decorators import action
|
|
from rest_framework.exceptions import ParseError
|
|
from rest_framework.views import APIView
|
|
from rest_framework.serializers import Serializer
|
|
from apps.qm.models import QuaStat, NotOkOption, Ftest, Ptest, FtestWork, FtestItem
|
|
from apps.qm.models import TestItem, Defect, Qct, QctTestItem, QctMat, QctDefect
|
|
from apps.qm.serializers import QuaStatSerializer, TestItemSerializer, QuaStatUpdateSerializer, FtestSerializer, PtestSerializer, \
|
|
FtestWorkCreateUpdateSerializer, FtestWorkSerializer, DefectSerializer, QctSerializer, QctTestItemSerializer, QctDefectSerializer, QctMatSerializer, \
|
|
QctDetailSerializer, UpdateRatePassTSerializer, QctGetSerializer
|
|
from apps.qm.tasks import cal_quastat_sflog
|
|
from rest_framework.response import Response
|
|
from apps.utils.mixins import BulkUpdateModelMixin
|
|
import datetime
|
|
|
|
from apps.utils.viewsets import CustomGenericViewSet, CustomModelViewSet
|
|
from apps.wpm.models import SfLog
|
|
from apps.qm.filters import QuaStatFilter, TestItemFilter, FtestWorkFilter, QctFilter, FtestFilter
|
|
from django.db import transaction
|
|
from apps.qm.services import ftestwork_submit
|
|
from apps.wpm.services_2 import ana_batch_thread
|
|
from apps.wf.models import State
|
|
# Create your views here.
|
|
|
|
class DefectViewSet(CustomModelViewSet):
|
|
"""
|
|
缺陷项
|
|
|
|
缺陷项
|
|
"""
|
|
queryset = Defect.objects.all()
|
|
serializer_class = DefectSerializer
|
|
filterset_fields = ["cate", "okcate"]
|
|
search_fields = ["name", "code"]
|
|
|
|
def perform_destroy(self, instance):
|
|
QctDefect.objects.filter(defect=instance).delete()
|
|
instance.delete()
|
|
|
|
class QctViewSet(CustomModelViewSet):
|
|
"""
|
|
检测模板
|
|
|
|
检测模板
|
|
"""
|
|
queryset = Qct.objects.all()
|
|
serializer_class = QctSerializer
|
|
retrieve_serializer_class = QctDetailSerializer
|
|
filterset_class = QctFilter
|
|
search_fields = ["name", "number"]
|
|
ordering = ["number", "-update_time"]
|
|
ordering_fields = ["number", "create_time", "update_time"]
|
|
|
|
@action(methods=['post'], detail=False, perms_map={'post': '*'}, serializer_class=QctGetSerializer)
|
|
@transaction.atomic
|
|
def get_qct(self, request, *args, **kwargs):
|
|
"""
|
|
获取检验模板
|
|
|
|
获取检验模板
|
|
"""
|
|
sr = QctGetSerializer(data=request.data)
|
|
sr.is_valid(raise_exception=True)
|
|
vdata = sr.validated_data
|
|
qct = Qct.get(vdata["material"], vdata["tag"], vdata.get("type", None))
|
|
if qct:
|
|
return Response(QctDetailSerializer(instance=qct).data)
|
|
return Response()
|
|
|
|
|
|
class QctTestItemViewSet(CustomModelViewSet):
|
|
"""检测模板项
|
|
|
|
检测模板项
|
|
"""
|
|
perms_map = {"get": "*", "post": "qct.update", "put": "qct.update", "delete": "qct.update"}
|
|
queryset = QctTestItem.objects.all()
|
|
serializer_class = QctTestItemSerializer
|
|
select_related_fields = ["qct", "testitem"]
|
|
filterset_fields = ["qct", "testitem"]
|
|
ordering = ["qct", "sort"]
|
|
|
|
class QctDefectViewSet(CustomModelViewSet):
|
|
"""检测缺陷项
|
|
|
|
检测缺陷项
|
|
"""
|
|
perms_map = {"get": "*", "post": "qct.update", "put": "qct.update", "delete": "qct.update"}
|
|
queryset = QctDefect.objects.all()
|
|
serializer_class = QctDefectSerializer
|
|
select_related_fields = ["qct", "defect"]
|
|
filterset_fields = ["qct", "defect"]
|
|
ordering = ["qct", "sort"]
|
|
|
|
def perform_create(self, serializer):
|
|
ins: QctDefect = serializer.save()
|
|
if ins.is_default:
|
|
QctDefect.objects.filter(qct=ins.qct).exclude(id=ins.id).update(is_default=False)
|
|
|
|
def perform_update(self, serializer):
|
|
ins: QctDefect = serializer.save()
|
|
if ins.is_default:
|
|
QctDefect.objects.filter(qct=ins.qct).exclude(id=ins.id).update(is_default=False)
|
|
|
|
class QctMatViewSet(CustomModelViewSet):
|
|
"""检测物料
|
|
|
|
检测物料
|
|
"""
|
|
perms_map = {"get": "*", "post": "qct.update", "put": "qct.update", "delete": "qct.update"}
|
|
queryset = QctMat.objects.all()
|
|
serializer_class = QctMatSerializer
|
|
filterset_fields = ["qct", "material"]
|
|
|
|
class NotOkOptionView(APIView):
|
|
perms_map = {'get': '*'}
|
|
|
|
def get(self, request):
|
|
"""
|
|
获取不合格项
|
|
|
|
获取不合格项
|
|
"""
|
|
res1 = [{'value': i.value, 'label': i.label, 'extra_info': i.get_extra_info(i.value)} for i in NotOkOption]
|
|
res2 = {i.value: i.label for i in NotOkOption}
|
|
return Response({"res_list": res1, "res_dict": res2})
|
|
|
|
class TestItemViewSet(CustomModelViewSet):
|
|
"""
|
|
list:质检项目
|
|
|
|
质检项目
|
|
"""
|
|
queryset = TestItem.objects.all()
|
|
serializer_class = TestItemSerializer
|
|
filterset_class = TestItemFilter
|
|
search_fields = ['tags', 'name', 'number', 'mcate_tags']
|
|
ordering = ['id']
|
|
|
|
def add_info_for_list(self, data):
|
|
affects_list = [i['affects'] for i in data]
|
|
affectIds = []
|
|
for item in affects_list:
|
|
affectIds.extend(item)
|
|
affects = TestItem.objects.filter(id__in=affectIds).values_list('id', 'name')
|
|
affects_dict = dict(affects)
|
|
for item in data:
|
|
affects = item["affects"]
|
|
item["affects_name"] = ";".join([affects_dict.get(x, '未知') for x in affects])
|
|
return data
|
|
|
|
def perform_destroy(self, instance):
|
|
QctTestItem.objects.filter(testitem=instance).delete()
|
|
instance.delete()
|
|
|
|
|
|
class QuaStatViewSet(ListModelMixin, BulkUpdateModelMixin, CustomGenericViewSet):
|
|
"""
|
|
list:质量数据统计
|
|
|
|
质量数据统计
|
|
"""
|
|
perms_map = {'get': '*', 'put': 'quastat.update'}
|
|
queryset = QuaStat.objects.all()
|
|
serializer_class = QuaStatSerializer
|
|
update_serializer_class = QuaStatUpdateSerializer
|
|
filterset_class = QuaStatFilter
|
|
select_related_fields = ['belong_dept', 'material', 'testitem']
|
|
ordering = ['belong_dept__sort', 'material__sort', 'testitem__sort']
|
|
|
|
def after_bulk_update(self, objs):
|
|
now = datetime.datetime.now()
|
|
sflogIds = []
|
|
for i in objs:
|
|
sflogIds.append(i['sflog'])
|
|
sflogIds = list(set(sflogIds))
|
|
SfLog.objects.filter(id__in=sflogIds).update(
|
|
last_test_time=now) # 更新质检记录时间
|
|
for sflogId in sflogIds:
|
|
cal_quastat_sflog.delay(sflogId)
|
|
|
|
# 根据物料ID和检测项目ID rate_pass_t
|
|
@action(detail=False, methods=['post'], url_path='update_rate_pass_t', perms_map={'post': 'quastat.submit'})
|
|
def update_rate_pass_t(self, request, *args, **kwargs):
|
|
# material_id = request.data.get("material_id")
|
|
# testitem_id = request.data.get("testitem_id")
|
|
# new_rate_pass_t = request.data.get("rate_pass_t")
|
|
# # update_start = datetime.datetime.strptime(request.data.get("update_start_time"), "%Y-%m-%d")
|
|
# # update_end = datetime.datetime.strptime(request.data.get("update_end_time"), "%Y-%m-%d")
|
|
# month_s = request.data.get("month_s")
|
|
serializer = UpdateRatePassTSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
material_id = serializer.validated_data.get("material_id")
|
|
testitem_id = serializer.validated_data.get("testitem_id")
|
|
new_rate_pass_t = serializer.validated_data.get("rate_pass_t")
|
|
month_s = serializer.validated_data.get("month_s")
|
|
if not (material_id and testitem_id and new_rate_pass_t is not None):
|
|
return Response(
|
|
{"detail": "Missing required fields: material_id, testitem_id, rate_pass_t"},
|
|
status=400
|
|
)
|
|
# 先根据查询物料查询 sflogids 然后再过滤 work_data ,返回在时间段内的sflogids,再更新quaStat
|
|
qua_data = QuaStat.objects.filter(
|
|
material_id=material_id,
|
|
testitem_id=testitem_id,
|
|
sflog__work_date__month=month_s,
|
|
|
|
)
|
|
sflog_ids = qua_data.values_list('sflog', flat=True).distinct()
|
|
update_count = qua_data.update(rate_pass_t=new_rate_pass_t)
|
|
for i in sflog_ids:
|
|
cal_quastat_sflog.delay(i)
|
|
return Response({"rate_pass_t": new_rate_pass_t}, status=200)
|
|
|
|
|
|
class FtestViewSet(CustomModelViewSet):
|
|
"""
|
|
list:首件/成品检验
|
|
|
|
首件/成品检验
|
|
"""
|
|
queryset = Ftest.objects.all()
|
|
serializer_class = FtestSerializer
|
|
select_related_fields = ['test_user', 'check_user', 'ftest_work']
|
|
filterset_class = FtestFilter
|
|
|
|
def count_sampling(self, ftest_work:FtestWork):
|
|
qs = Ftest.objects.filter(ftest_work=ftest_work)
|
|
all_count = qs.count()
|
|
ok_count = qs.filter(is_ok=True).count()
|
|
ftest_work.count_sampling = all_count
|
|
ftest_work.count_sampling_ok = ok_count
|
|
if ftest_work.type2 == FtestWork.TYPE2_ALL: # 如果是全检
|
|
ftest_work.count_ok = ok_count
|
|
ftest_work.count_notok = all_count - ok_count
|
|
ftest_work.save()
|
|
|
|
def perform_create(self, serializer):
|
|
ins: Ftest = serializer.save()
|
|
if ins.ftest_work:
|
|
self.count_sampling(ins.ftest_work)
|
|
|
|
def perform_update(self, serializer):
|
|
ins: Ftest = serializer.save()
|
|
if ins.ftest_work:
|
|
self.count_sampling(ins.ftest_work)
|
|
|
|
def perform_destroy(self, instance):
|
|
ftest_work = instance.ftest_work
|
|
instance.delete()
|
|
if ftest_work:
|
|
self.count_sampling(ftest_work)
|
|
|
|
|
|
class PtestViewSet(CustomModelViewSet):
|
|
"""
|
|
list:性能测试
|
|
|
|
性能测试
|
|
"""
|
|
queryset = Ptest.objects.all()
|
|
serializer_class = PtestSerializer
|
|
select_related_fields = ['testitem']
|
|
filterset_fields = ['testitem', 'test_date', 'sample_number']
|
|
search_fields = ['testitem__name', 'sample_number']
|
|
|
|
|
|
class FtestWorkViewSet(CustomModelViewSet):
|
|
"""
|
|
list: 检验工作
|
|
|
|
检验工作
|
|
"""
|
|
queryset = FtestWork.objects.all()
|
|
serializer_class = FtestWorkSerializer
|
|
create_serializer_class = FtestWorkCreateUpdateSerializer
|
|
update_serializer_class = FtestWorkCreateUpdateSerializer
|
|
select_related_fields = ['material', 'mb', 'mb__material']
|
|
filterset_class = FtestWorkFilter
|
|
|
|
@transaction.atomic
|
|
def update(self, request, *args, **kwargs):
|
|
ins:FtestWork = self.get_object()
|
|
partial = kwargs.pop('partial', False)
|
|
if ins.submit_time is not None:
|
|
raise ParseError('已提交无法修改')
|
|
if ins.ticket and ins.ticket.state.type != State.STATE_TYPE_START:
|
|
raise ParseError('审批单已进行，无法修改')
|
|
serializer = self.get_serializer(ins, data=request.data, partial=partial)
|
|
serializer.is_valid(raise_exception=True)
|
|
self.perform_update(serializer)
|
|
# 触发批次统计分析
|
|
ana_batch_thread(xbatchs=[ins.batch])
|
|
return Response(serializer.data)
|
|
|
|
@transaction.atomic
|
|
def destroy(self, request, *args, **kwargs):
|
|
ins:FtestWork = self.get_object()
|
|
if ins.submit_time is not None:
|
|
raise ParseError('已提交无法删除')
|
|
if ins.ticket:
|
|
raise ParseError('存在审批, 无法删除')
|
|
self.perform_destroy(ins)
|
|
# 触发批次统计分析
|
|
ana_batch_thread(xbatchs=[ins.batch])
|
|
return Response(status=204)
|
|
|
|
def perform_create(self, serializer):
|
|
ins = serializer.save()
|
|
# 触发批次统计分析
|
|
ana_batch_thread(xbatchs=[ins.batch])
|
|
|
|
|
|
@action(methods=['post'], detail=True, perms_map={'post': 'ftestwork.submit'}, serializer_class=Serializer)
|
|
@transaction.atomic
|
|
def submit(self, request, *args, **kwargs):
|
|
"""提交检验工作
|
|
|
|
提交检验工作
|
|
"""
|
|
ins:FtestWork = self.get_object()
|
|
if ins.ticket:
|
|
raise ParseError('该检验工作存在审批!')
|
|
if ins.wm is None and ins.mb is None:
|
|
raise ParseError('该检验工作未关联库存')
|
|
if ins.submit_time is None:
|
|
ftestwork_submit(ins, request.user)
|
|
else:
|
|
raise ParseError('该检验工作已提交')
|
|
return Response() |