feat(wpm): recommend equipment for production logs

apps/wpm/serializers.py

@@ -41,6 +41,18 @@ WM_STATE_NAMES = {
 }
 
 
+class MlogEquipmentOptionSerializer(serializers.ModelSerializer):
+    mgroup_name = serializers.CharField(source="mgroup.name", read_only=True)
+    full_name = serializers.SerializerMethodField()
+
+    def get_full_name(self, obj):
+        return f"{obj.number}|{obj.name}|{obj.model}"
+
+    class Meta:
+        model = Equipment
+        fields = ["id", "name", "number", "model", "mgroup_name", "full_name"]
+
+
 class OtherLogSerializer(CustomModelSerializer):
     class Meta:
         model = OtherLog

apps/wpm/services.py

@@ -1,4 +1,5 @@
 import datetime
+from collections import defaultdict
 
 from django.core.cache import cache
 from django.db.models import Sum
@@ -27,6 +28,44 @@ from django.db.models import F
 
 myLogger = logging.getLogger('log')
 
+RECENT_EQUIPMENT_LOG_LIMIT = 50
+
+
+def get_recent_mgroup_equipment_ids(
+    mgroup_id, log_limit=RECENT_EQUIPMENT_LOG_LIMIT
+):
+    """按日志时间倒序返回工段最近使用过的设备 ID，空值和重复值忽略。"""
+    recent_logs = list(
+        Mlog.objects.filter(mgroup_id=mgroup_id)
+        .order_by("-create_time", "-id")
+        .values_list("id", "equipment_id", "equipment_2_id")[:log_limit]
+    )
+    if not recent_logs:
+        return []
+
+    log_ids = [log_id for log_id, _, _ in recent_logs]
+    multiple_equipment_ids = defaultdict(list)
+    for log_id, equipment_id in (
+        Mlog.equipments.through.objects.filter(mlog_id__in=log_ids)
+        .order_by("id")
+        .values_list("mlog_id", "equipment_id")
+    ):
+        multiple_equipment_ids[log_id].append(equipment_id)
+
+    result = []
+    seen = set()
+    for log_id, equipment_id, equipment_2_id in recent_logs:
+        candidate_ids = [
+            equipment_id,
+            equipment_2_id,
+            *multiple_equipment_ids[log_id],
+        ]
+        for candidate_id in candidate_ids:
+            if candidate_id and candidate_id not in seen:
+                seen.add(candidate_id)
+                result.append(candidate_id)
+    return result
+
 def inherit_zt_batch(source: BatchSt, target: BatchSt):
     """拆批/报工改号时目标批继承来源批的直通统计大批归属(纯继承, 不做判定)
 

apps/wpm/tests/test_equipment_log_options.py

@@ -0,0 +1,52 @@
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from apps.wpm.services import get_recent_mgroup_equipment_ids
+
+
+class RecentMgroupEquipmentTests(SimpleTestCase):
+    @patch("apps.wpm.services.Mlog")
+    def test_collects_all_equipment_fields_in_log_order_without_duplicates(
+        self, mlog
+    ):
+        log_queryset = MagicMock()
+        mlog.objects.filter.return_value = log_queryset
+        log_queryset.order_by.return_value.values_list.return_value.__getitem__.return_value = [
+            ("log-new", "equipment-a", None),
+            ("log-middle", "equipment-b", "equipment-a"),
+            ("log-empty", None, None),
+        ]
+        through_queryset = MagicMock()
+        mlog.equipments.through.objects.filter.return_value = through_queryset
+        through_queryset.order_by.return_value.values_list.return_value = [
+            ("log-new", "equipment-c"),
+            ("log-middle", "equipment-c"),
+            ("log-middle", "equipment-d"),
+        ]
+
+        result = get_recent_mgroup_equipment_ids("mgroup-1", log_limit=50)
+
+        self.assertEqual(
+            result,
+            ["equipment-a", "equipment-c", "equipment-b", "equipment-d"],
+        )
+        log_queryset.order_by.return_value.values_list.return_value.__getitem__.assert_called_once_with(
+            slice(None, 50, None)
+        )
+
+    @patch("apps.wpm.services.Mlog")
+    def test_returns_empty_when_recent_logs_have_no_equipment(self, mlog):
+        log_queryset = MagicMock()
+        mlog.objects.filter.return_value = log_queryset
+        log_queryset.order_by.return_value.values_list.return_value.__getitem__.return_value = [
+            ("log-1", None, None),
+            ("log-2", None, None),
+        ]
+        through_queryset = MagicMock()
+        mlog.equipments.through.objects.filter.return_value = through_queryset
+        through_queryset.order_by.return_value.values_list.return_value = []
+
+        result = get_recent_mgroup_equipment_ids("mgroup-1")
+
+        self.assertEqual(result, [])

apps/wpm/views.py

@@ -7,7 +7,7 @@ from rest_framework.decorators import action
 from rest_framework.exceptions import ParseError
 from rest_framework.response import Response
 from rest_framework.serializers import Serializer
-from django.db.models import Sum
+from django.db.models import Case, IntegerField, Sum, When
 from django.utils import timezone
 from apps.system.models import User
 
@@ -54,13 +54,23 @@ from .serializers import (
     MlogUserSerializer,
     BatchLogSerializer,
     MlogQuickSerializer,
+    MlogEquipmentOptionSerializer,
     MlogbwStartTestSerializer,
     HandoverListSerializer,
     BatchChangeSerializer,
     MlogbOutPatchUpdateSerializer
 )
-from .services import mlog_submit, handover_submit, mlog_revert, get_batch_dag, handover_revert
-from apps.wpm.services import mlog_submit_validate, generate_new_batch
+from .services import (
+    RECENT_EQUIPMENT_LOG_LIMIT,
+    generate_new_batch,
+    get_batch_dag,
+    get_recent_mgroup_equipment_ids,
+    handover_revert,
+    handover_submit,
+    mlog_revert,
+    mlog_submit,
+    mlog_submit_validate,
+)
 from apps.wf.models import State, Ticket
 from apps.wpmw.models import Wpr
 from apps.qm.models import Qct, Ftest, TestItem
@@ -332,6 +342,92 @@ class MlogViewSet(CustomModelViewSet):
     ]
     ordering_fields = ["create_time", "update_time"]
 
+    @swagger_auto_schema(
+        manual_parameters=[
+            openapi.Parameter(
+                name="mgroup",
+                in_=openapi.IN_QUERY,
+                description="日志所属工段",
+                type=openapi.TYPE_STRING,
+                required=True,
+            ),
+            openapi.Parameter(
+                name="search",
+                in_=openapi.IN_QUERY,
+                description="按设备名称或编号搜索全部生产设备",
+                type=openapi.TYPE_STRING,
+                required=False,
+            ),
+        ]
+    )
+    @action(
+        methods=["get"],
+        detail=False,
+        perms_map={"get": "*"},
+        serializer_class=MlogEquipmentOptionSerializer,
+    )
+    def equipment_options(self, request, *args, **kwargs):
+        """返回本工段设备、最近 50 条日志用过的设备或搜索结果。"""
+        mgroup_id = request.query_params.get("mgroup")
+        if not mgroup_id:
+            raise ParseError("请传入mgroup参数")
+
+        search = request.query_params.get("search", "").strip()
+        owned_ids = list(
+            Equipment.objects.filter(
+                type=Equipment.EQUIP_TYPE_PRO,
+                mgroup_id=mgroup_id,
+            )
+            .order_by("name", "number")
+            .values_list("id", flat=True)
+        )
+        owned_id_set = set(owned_ids)
+
+        if search:
+            queryset = (
+                Equipment.objects.filter(type=Equipment.EQUIP_TYPE_PRO)
+                .filter(Q(name__icontains=search) | Q(number__icontains=search))
+                .order_by("name", "number")
+            )
+            option_group = "搜索结果"
+        else:
+            recent_ids = get_recent_mgroup_equipment_ids(
+                mgroup_id, RECENT_EQUIPMENT_LOG_LIMIT
+            )
+            option_ids = list(dict.fromkeys([*owned_ids, *recent_ids]))
+            if option_ids:
+                order = Case(
+                    *[
+                        When(id=equipment_id, then=position)
+                        for position, equipment_id in enumerate(option_ids)
+                    ],
+                    output_field=IntegerField(),
+                )
+                queryset = Equipment.objects.filter(
+                    id__in=option_ids,
+                    type=Equipment.EQUIP_TYPE_PRO,
+                ).order_by(order)
+            else:
+                queryset = Equipment.objects.none()
+            option_group = None
+
+        queryset = queryset.select_related("mgroup")
+        page = self.paginate_queryset(queryset)
+        equipment_list = page if page is not None else queryset
+        data = MlogEquipmentOptionSerializer(
+            equipment_list,
+            many=True,
+            context=self.get_serializer_context(),
+        ).data
+        for item in data:
+            item["option_group"] = option_group or (
+                "本工段设备" if item["id"] in owned_id_set else "近期使用"
+            )
+
+        if page is not None:
+            return self.get_paginated_response(data)
+        return Response(data)
+
     def add_info_for_item(self, data):
         if data.get("oinfo_json", {}):
             czx_dict = dict(TestItem.objects.filter(id__in=data.get("oinfo_json", {}).keys()).values_list("id", "name"))