fix(wpm): derive number date filters from rule

apps/wpm/tests/test_number_rule.py

@@ -0,0 +1,64 @@
+from datetime import date
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from django.test import SimpleTestCase
+
+from apps.wpm.views import MlogbInViewSet
+
+
+class GenNumberWithRuleFilterTests(SimpleTestCase):
+    def test_date_filters_follow_rule_placeholders(self):
+        cases = [
+            ("SN-{n_count:04d}", {}),
+            ("{c_year}-{n_count:04d}", {"year": 2026}),
+            ("{c_year2}-{n_count:04d}", {"year": 2026}),
+            ("{c_month:02d}-{n_count:04d}", {"month": 8}),
+            ("{c_day:02d}-{n_count:04d}", {"day": 4}),
+            (
+                "{c_year}{c_month:02d}{c_day:02d}-{n_count:04d}",
+                {"year": 2026, "month": 8, "day": 4},
+            ),
+        ]
+        material = SimpleNamespace(model=None)
+        mlog = SimpleNamespace(
+            handle_date=date(2026, 8, 4),
+            mgroup=SimpleNamespace(process=SimpleNamespace(id=123)),
+        )
+
+        for rule, expected_dates in cases:
+            with self.subTest(rule=rule):
+                queryset = MagicMock()
+                queryset.annotate.return_value = queryset
+                queryset.order_by.return_value = queryset
+                queryset.last.return_value = None
+
+                with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset) as mock_filter:
+                    MlogbInViewSet.gen_number_with_rule(rule, material, mlog)
+
+                filters = mock_filter.call_args.kwargs
+                date_prefix = "wpr_mlogbw__mlogb__mlog__handle_date__"
+                actual_dates = {
+                    key.removeprefix(date_prefix): value
+                    for key, value in filters.items()
+                    if key.startswith(date_prefix)
+                }
+                self.assertEqual(actual_dates, expected_dates)
+
+    def test_escaped_date_placeholder_text_does_not_add_filter(self):
+        queryset = MagicMock()
+        queryset.annotate.return_value = queryset
+        queryset.order_by.return_value = queryset
+        queryset.last.return_value = None
+        material = SimpleNamespace(model=None)
+        mlog = SimpleNamespace(
+            handle_date=date(2026, 8, 4),
+            mgroup=SimpleNamespace(process=SimpleNamespace(id=123)),
+        )
+
+        with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset) as mock_filter:
+            MlogbInViewSet.gen_number_with_rule("{{c_year}}-{n_count:04d}", material, mlog)
+
+        self.assertFalse(
+            any("handle_date" in key for key in mock_filter.call_args.kwargs)
+        )

apps/wpm/views.py

@@ -1,5 +1,6 @@
 import math
 import re
+from string import Formatter
 
 from django.db import transaction
 from rest_framework.decorators import action
@@ -1010,13 +1011,18 @@ class MlogbInViewSet(BulkCreateModelMixin, BulkUpdateModelMixin, BulkDestroyMode
     def gen_number_with_rule(cls, rule, material_out: Material, mlog: Mlog, gen_count=1):
         from apps.wpmw.models import Wpr
 
+        rule_fields = {
+            field_name
+            for _, field_name, _, _ in Formatter().parse(rule)
+            if field_name
+        }
         handle_date = mlog.handle_date
         c_year = handle_date.year
         c_year2 = str(c_year)[-2:]
         c_month = handle_date.month
         c_day = handle_date.day
         m_model = material_out.model
-        if 'm_model' in rule:
+        if "m_model" in rule_fields:
             if m_model is None:
                 raise ParseError("生成编号出错:产品型号不能为空")
             elif m_model and m_model.islower():
@@ -1029,16 +1035,18 @@ class MlogbInViewSet(BulkCreateModelMixin, BulkUpdateModelMixin, BulkDestroyMode
         if connection.vendor == "postgresql" and connection.in_atomic_block:
             with connection.cursor() as cursor:
                 cursor.execute("SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", [f"wpr_number_rule:{process.id}"])
-        # 按生产日志查询, 流水号归零周期跟随规则中最细的日期占位符
+        # 只按规则中实际使用的日期占位符筛选历史编号
         wpr_filter = {
             "wpr_mlogbw__mlogb__material_out__isnull": False,
             "wpr_mlogbw__mlogb__mlog__mgroup__process": process,
             "wpr_mlogbw__mlogb__mlog__is_fix": False,
             "wpr_mlogbw__mlogb__mlog__submit_time__isnull": False,
-            "wpr_mlogbw__mlogb__mlog__handle_date__year": c_year,
-            "wpr_mlogbw__mlogb__mlog__handle_date__month": c_month,
         }
-        if "c_day" in rule:
+        if rule_fields & {"c_year", "c_year2"}:
+            wpr_filter["wpr_mlogbw__mlogb__mlog__handle_date__year"] = c_year
+        if "c_month" in rule_fields:
+            wpr_filter["wpr_mlogbw__mlogb__mlog__handle_date__month"] = c_month
+        if "c_day" in rule_fields:
             wpr_filter["wpr_mlogbw__mlogb__mlog__handle_date__day"] = c_day
         wpr = (
             Wpr.objects.filter(**wpr_filter)