zcdsj
/
factory
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(inventory): add effective defect grade filtering

This commit is contained in:
caoqianming 2026-08-06 10:31:37 +08:00
parent ed952d2d3a
commit b365e06318
10 changed files with 290 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
apps/inm/filters.py
View File

@ -1,10 +1,16 @@
from django_filters import rest_framework as filters
from apps.inm.models import MaterialBatch, MIO
from django.db.models import Q, Subquery, OuterRef, F
from apps.qm.defect_grades import effective_defect_grade_q
class MaterialBatchFilter(filters.FilterSet):
count_canmio__gt = filters.NumberFilter(
method='filter_count_canmio__gt', label='可发数量大于')
defect_grade = filters.NumberFilter(
method='filter_defect_grade', label='有效缺陷等级')
def filter_defect_grade(self, queryset, name, value):
return queryset.filter(effective_defect_grade_q(value))
class Meta:
model = MaterialBatch

18
apps/inm/serializers.py
View File

@ -14,6 +14,7 @@ from django.db.models import F, Sum, DecimalField
from server.settings import get_sysconfig
from apps.wpmw.models import Wpr
from decimal import Decimal
from apps.qm.defect_grades import DEFECT_GRADE_NAMES, effective_defect_grade
class WareHourseSerializer(CustomModelSerializer):
@ -49,6 +50,8 @@ class MaterialBatchSerializer(CustomModelSerializer):
source='supplier', read_only=True)
material_ = MaterialSerializer(source='material', read_only=True)
defect_name = serializers.CharField(source="defect.name", read_only=True)
defect_grade = serializers.SerializerMethodField()
defect_grade_name = serializers.SerializerMethodField()
count_mioing = serializers.SerializerMethodField(label='正在出入库数量')
class Meta:
@ -61,6 +64,12 @@ class MaterialBatchSerializer(CustomModelSerializer):
# 保留 decimal 精度(原 IntegerField 会截断在途量, 导致可发量偏大)
return instance.count_mioing_anno if hasattr(instance, 'count_mioing_anno') else instance.count_mioing
def get_defect_grade(self, instance):
return effective_defect_grade(instance)
def get_defect_grade_name(self, instance):
return DEFECT_GRADE_NAMES[self.get_defect_grade(instance)]
def to_representation(self, instance):
ret = super().to_representation(instance)
if 'count' in ret:
@ -86,6 +95,15 @@ class MaterialBatchDetailSerializer(CustomModelSerializer):
source='a_mb', read_only=True, many=True)
supplier_name = serializers.StringRelatedField(
source='supplier', read_only=True)
defect_name = serializers.CharField(source="defect.name", read_only=True)
defect_grade = serializers.SerializerMethodField()
defect_grade_name = serializers.SerializerMethodField()
def get_defect_grade(self, instance):
return effective_defect_grade(instance)
def get_defect_grade_name(self, instance):
return DEFECT_GRADE_NAMES[self.get_defect_grade(instance)]
class Meta:
model = MaterialBatch

70
apps/inm/tests.py
View File

@ -4,10 +4,78 @@ from threading import Barrier
from unittest import skipUnless
from django.db import connection, connections, transaction
from django.test import SimpleTestCase, TransactionTestCase
from django.test import SimpleTestCase, TestCase, TransactionTestCase
from apps.inm.filters import MaterialBatchFilter
from apps.inm.models import MaterialBatch, WareHouse
from apps.inm.serializers import MaterialBatchSerializer
from apps.mtm.models import Material
from apps.qm.models import Defect
class MaterialBatchDefectGradeTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.material = Material.objects.create(name='仓库缺陷等级测试物料')
cls.warehouse = WareHouse.objects.create(
number='GRADE',
name='等级测试仓库',
place='测试地点',
)
cls.defect_b = Defect.objects.create(
name='仓库B类缺陷',
cate=Defect.cate_list[0],
okcate=Defect.DEFECT_OK_B,
)
cls.notok_without_defect = MaterialBatch.objects.create(
material=cls.material,
warehouse=cls.warehouse,
batch='MB-NOTOK-NONE',
count=1,
state=20,
)
cls.normal_with_b_defect = MaterialBatch.objects.create(
material=cls.material,
warehouse=cls.warehouse,
batch='MB-NORMAL-B',
count=1,
state=10,
defect=cls.defect_b,
)
def test_serializer_uses_defect_or_defaults_to_ok_independent_of_state(self):
no_defect_data = MaterialBatchSerializer(
self.notok_without_defect
).data
b_defect_data = MaterialBatchSerializer(
self.normal_with_b_defect
).data
self.assertEqual(no_defect_data['defect_grade'], Defect.DEFECT_OK)
self.assertEqual(no_defect_data['defect_grade_name'], '合格')
self.assertEqual(b_defect_data['defect_grade'], Defect.DEFECT_OK_B)
self.assertEqual(b_defect_data['defect_grade_name'], '合格B类')
def test_effective_grade_filter_is_independent_of_state(self):
ok_items = MaterialBatchFilter(
{'defect_grade': Defect.DEFECT_OK},
queryset=MaterialBatch.objects.all(),
).qs
b_items = MaterialBatchFilter(
{'defect_grade': Defect.DEFECT_OK_B},
queryset=MaterialBatch.objects.all(),
).qs
self.assertQuerySetEqual(
ok_items,
[self.notok_without_defect],
transform=lambda item: item,
)
self.assertQuerySetEqual(
b_items,
[self.normal_with_b_defect],
transform=lambda item: item,
)
class MaterialBatchInventoryKeyTests(SimpleTestCase):

2
apps/inm/views.py
View File

@ -60,7 +60,7 @@ class MaterialBatchViewSet(ListModelMixin, CustomGenericViewSet):
queryset = MaterialBatch.objects.filter(count__gt=0)
serializer_class = MaterialBatchSerializer
retrieve_serializer_class = MaterialBatchDetailSerializer
select_related_fields = ['warehouse', 'material', 'supplier']
select_related_fields = ['warehouse', 'material', 'supplier', 'defect']
filterset_class = MaterialBatchFilter
search_fields = ['material__name', 'material__number',
'material__model', 'material__specification', 'batch']

42
apps/qm/defect_grades.py Normal file
View File

@ -0,0 +1,42 @@
from django.db.models import Q
DEFECT_OK = 10
DEFECT_OK_B = 20
DEFECT_NOTOK = 30
DEFECT_GRADE_CHOICES = (
(DEFECT_OK, "合格"),
(DEFECT_OK_B, "合格B类"),
(DEFECT_NOTOK, "不合格"),
)
DEFECT_GRADE_NAMES = dict(DEFECT_GRADE_CHOICES)
def effective_defect_grade(instance, notok_sign_field=None):
"""Return the inventory grade without coupling it to inventory state."""
defect = getattr(instance, "defect", None)
if defect is not None:
return defect.okcate
if notok_sign_field and getattr(instance, notok_sign_field, None):
return DEFECT_NOTOK
return DEFECT_OK
def effective_defect_grade_q(value, notok_sign_field=None):
"""Build an index-friendly query matching ``effective_defect_grade``."""
explicit_grade = Q(defect__okcate=value)
without_defect = Q(defect__isnull=True)
if not notok_sign_field:
return explicit_grade | without_defect if value == DEFECT_OK else explicit_grade
has_legacy_sign = (
Q(**{f"{notok_sign_field}__isnull": False})
& ~Q(**{notok_sign_field: ""})
)
if value == DEFECT_OK:
return explicit_grade | (without_defect & ~has_legacy_sign)
if value == DEFECT_NOTOK:
return explicit_grade | (without_defect & has_legacy_sign)
return explicit_grade

16
apps/qm/models.py
View File

@ -8,19 +8,25 @@ from django.utils.translation import gettext_lazy as _
from django.db import transaction
from django.db.models import Sum
from rest_framework.exceptions import ParseError
from apps.qm.defect_grades import (
DEFECT_GRADE_CHOICES,
DEFECT_NOTOK as GRADE_NOTOK,
DEFECT_OK as GRADE_OK,
DEFECT_OK_B as GRADE_OK_B,
)
class Defect(CommonAModel):
"""TN:缺陷项"""
DEFECT_OK = 10
DEFECT_OK_B = 20
DEFECT_NOTOK = 30
DEFECT_OK = GRADE_OK
DEFECT_OK_B = GRADE_OK_B
DEFECT_NOTOK = GRADE_NOTOK
cate_list = ["尺寸", "外观", "内质", "性能"]
name = models.CharField(max_length=50, verbose_name="名称")
code = models.CharField(max_length=50, verbose_name="标识", null=True, blank=True)
cate = models.CharField(max_length=50, verbose_name="分类", help_text=str(cate_list))
okcate= models.PositiveSmallIntegerField(verbose_name="不合格分类",
choices=((DEFECT_OK, "合格"), (DEFECT_OK_B, "合格B类"), (DEFECT_NOTOK, "不合格")),
default=DEFECT_NOTOK)
choices=DEFECT_GRADE_CHOICES,
default=GRADE_NOTOK)
note = models.TextField('备注', null=True, blank=True)
def __str__(self):

5
apps/wpm/filters.py
View File

@ -5,6 +5,7 @@ from apps.mtm.models import Route, Material
from django.db.models import Q, Exists, OuterRef
from rest_framework.exceptions import ParseError
from datetime import datetime
from apps.qm.defect_grades import effective_defect_grade_q
class SfLogFilter(filters.FilterSet):
class Meta:
@ -44,6 +45,10 @@ class WMaterialFilter(filters.FilterSet):
mlog_date_start = filters.DateFilter(label="产出开始", method="filter_mlog_date_start")
mlog_date_end = filters.DateFilter(label="产出结束", method="filter_mlog_date_end")
current_merged = filters.BooleanFilter(label="是否本工段新合成的批", method="filter_current_merged")
defect_grade = filters.NumberFilter(label="有效缺陷等级", method="filter_defect_grade")
def filter_defect_grade(self, queryset, name, value):
return queryset.filter(effective_defect_grade_q(value, "notok_sign"))
def filter_mlog_date_start(self, queryset, name, value):
mgroupId = self.data.get("mgroup", None)

5
apps/wpm/serializers.py
View File

@ -24,6 +24,7 @@ from apps.wpmw.models import Wpr
from apps.qm.serializers import FtestProcessSerializer, FtestProcessListSerializer
import logging
from apps.qm.models import Defect
from apps.qm.defect_grades import DEFECT_GRADE_NAMES, effective_defect_grade
from apps.utils.snowflake import idWorker
from decimal import Decimal
from apps.em.models import Equipment
@ -199,10 +200,10 @@ class WMaterialSerializer(CustomModelSerializer):
return getattr(NotOkOption, obj.notok_sign, NotOkOption.qt).label if obj.notok_sign else None
def get_defect_grade(self, obj):
return obj.defect.okcate if obj.defect else None
return effective_defect_grade(obj, "notok_sign")
def get_defect_grade_name(self, obj):
return obj.defect.get_okcate_display() if obj.defect else None
return DEFECT_GRADE_NAMES[self.get_defect_grade(obj)]
def get_count_working(self, obj):
# 列表接口 queryset 已注解(单次聚合); 嵌套等无注解场景回退模型属性

51
apps/wpm/tests.py
View File

@ -138,13 +138,30 @@ class WMaterialDefectGradeTests(TestCase):
count=1,
state=WMaterial.WM_OK,
)
cls.repair_without_defect = WMaterial.objects.create(
material=cls.material,
batch="REPAIR-NONE",
count=1,
state=WMaterial.WM_REPAIR,
)
cls.notok_with_legacy_sign = WMaterial.objects.create(
material=cls.material,
batch="NOTOK-LEGACY",
count=1,
state=WMaterial.WM_NOTOK,
notok_sign="zw",
)
def test_serializer_exposes_nullable_defect_grade_without_using_state(self):
def test_serializer_exposes_effective_defect_grade_without_using_state(self):
normal_notok_data = WMaterialSerializer(
self.normal_with_notok_defect
).data
notok_b_data = WMaterialSerializer(self.notok_with_b_defect).data
no_defect_data = WMaterialSerializer(self.normal_without_defect).data
repair_no_defect_data = WMaterialSerializer(
self.repair_without_defect
).data
legacy_data = WMaterialSerializer(self.notok_with_legacy_sign).data
self.assertEqual(
normal_notok_data["defect_grade"], Defect.DEFECT_NOTOK
@ -154,8 +171,14 @@ class WMaterialDefectGradeTests(TestCase):
notok_b_data["defect_grade"], Defect.DEFECT_OK_B
)
self.assertEqual(notok_b_data["defect_grade_name"], "合格B类")
self.assertIsNone(no_defect_data["defect_grade"])
self.assertIsNone(no_defect_data["defect_grade_name"])
self.assertEqual(no_defect_data["defect_grade"], Defect.DEFECT_OK)
self.assertEqual(no_defect_data["defect_grade_name"], "合格")
self.assertEqual(
repair_no_defect_data["defect_grade"], Defect.DEFECT_OK
)
self.assertEqual(repair_no_defect_data["defect_grade_name"], "合格")
self.assertEqual(legacy_data["defect_grade"], Defect.DEFECT_NOTOK)
self.assertEqual(legacy_data["defect_grade_name"], "不合格")
def test_filtering_state_and_defect_grade_are_independent(self):
normal_notok = WMaterialFilter(
@ -184,6 +207,28 @@ class WMaterialDefectGradeTests(TestCase):
transform=lambda item: item,
)
def test_effective_grade_filter_includes_defaults_and_legacy_signs(self):
ok_items = WMaterialFilter(
{"defect_grade": Defect.DEFECT_OK},
queryset=WMaterial.objects.all(),
).qs
notok_items = WMaterialFilter(
{"defect_grade": Defect.DEFECT_NOTOK},
queryset=WMaterial.objects.all(),
).qs
self.assertCountEqual(
ok_items.values_list("id", flat=True),
[self.normal_without_defect.id, self.repair_without_defect.id],
)
self.assertCountEqual(
notok_items.values_list("id", flat=True),
[
self.normal_with_notok_defect.id,
self.notok_with_legacy_sign.id,
],
)
class MlogbwViewSetTests(SimpleTestCase):
@patch("apps.wpm.views.MlogViewSet.lock_and_check_can_update")

86
apps/wpm/tests/test_defect_grade.py Normal file
View File

@ -0,0 +1,86 @@
from django.test import TestCase
from apps.mtm.models import Material
from apps.qm.models import Defect
from apps.wpm.filters import WMaterialFilter
from apps.wpm.models import WMaterial
from apps.wpm.serializers import WMaterialSerializer
class WMaterialDefectGradeTests(TestCase):
@classmethod
def setUpTestData(cls):
cls.material = Material.objects.create(name="缺陷等级测试物料")
cls.defect_b = Defect.objects.create(
name="B类缺陷",
cate=Defect.cate_list[0],
okcate=Defect.DEFECT_OK_B,
)
cls.defect_notok = Defect.objects.create(
name="不合格缺陷",
cate=Defect.cate_list[0],
okcate=Defect.DEFECT_NOTOK,
)
cls.repair_without_defect = WMaterial.objects.create(
material=cls.material,
batch="REPAIR-NONE",
count=1,
state=WMaterial.WM_REPAIR,
)
cls.normal_with_notok_defect = WMaterial.objects.create(
material=cls.material,
batch="NORMAL-NOTOK",
count=1,
state=WMaterial.WM_OK,
defect=cls.defect_notok,
)
cls.notok_with_b_defect = WMaterial.objects.create(
material=cls.material,
batch="NOTOK-B",
count=1,
state=WMaterial.WM_NOTOK,
defect=cls.defect_b,
)
cls.notok_with_legacy_sign = WMaterial.objects.create(
material=cls.material,
batch="NOTOK-LEGACY",
count=1,
state=WMaterial.WM_NOTOK,
notok_sign="zw",
)
def test_serializer_uses_effective_grade_independent_of_state(self):
repair_data = WMaterialSerializer(self.repair_without_defect).data
notok_data = WMaterialSerializer(
self.normal_with_notok_defect
).data
b_data = WMaterialSerializer(self.notok_with_b_defect).data
legacy_data = WMaterialSerializer(self.notok_with_legacy_sign).data
self.assertEqual(repair_data["defect_grade"], Defect.DEFECT_OK)
self.assertEqual(repair_data["defect_grade_name"], "合格")
self.assertEqual(notok_data["defect_grade"], Defect.DEFECT_NOTOK)
self.assertEqual(b_data["defect_grade"], Defect.DEFECT_OK_B)
self.assertEqual(legacy_data["defect_grade"], Defect.DEFECT_NOTOK)
def test_effective_grade_filter_matches_serializer_rules(self):
ok_items = WMaterialFilter(
{"defect_grade": Defect.DEFECT_OK},
queryset=WMaterial.objects.all(),
).qs
notok_items = WMaterialFilter(
{"defect_grade": Defect.DEFECT_NOTOK},
queryset=WMaterial.objects.all(),
).qs
self.assertCountEqual(
ok_items.values_list("id", flat=True),
[self.repair_without_defect.id],
)
self.assertCountEqual(
notok_items.values_list("id", flat=True),
[
self.normal_with_notok_defect.id,
self.notok_with_legacy_sign.id,
],
)