From fd4de2bd4b049099ec8c9d685d62e5d573052fd3 Mon Sep 17 00:00:00 2001 From: caoqianming Date: Fri, 7 Aug 2026 09:58:54 +0800 Subject: [PATCH] fix(wpm): match historical numbers to active rule --- apps/wpm/tests/test_number_rule.py | 141 +++++++++++++++++++++++++++-- apps/wpm/views.py | 60 +++++++++--- 2 files changed, 182 insertions(+), 19 deletions(-) diff --git a/apps/wpm/tests/test_number_rule.py b/apps/wpm/tests/test_number_rule.py index ee8e7d5e..0715db3c 100644 --- a/apps/wpm/tests/test_number_rule.py +++ b/apps/wpm/tests/test_number_rule.py @@ -29,9 +29,7 @@ class GenNumberWithRuleFilterTests(SimpleTestCase): 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 + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [] with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset) as mock_filter: MlogbInViewSet.gen_number_with_rule(rule, material, mlog) @@ -47,9 +45,7 @@ class GenNumberWithRuleFilterTests(SimpleTestCase): 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 + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [] material = SimpleNamespace(model=None) mlog = SimpleNamespace( handle_date=date(2026, 8, 4), @@ -62,3 +58,136 @@ class GenNumberWithRuleFilterTests(SimpleTestCase): self.assertFalse( any("handle_date" in key for key in mock_filter.call_args.kwargs) ) + + def test_previous_sequence_width_is_compatible(self): + queryset = MagicMock() + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [ + "2608P0001", + "2608P0003", + "2608P0002", + ] + material = SimpleNamespace(model="P") + mlog = SimpleNamespace( + handle_date=date(2026, 8, 6), + mgroup=SimpleNamespace(process=SimpleNamespace(id=123)), + ) + + with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset): + number = MlogbInViewSet.gen_number_with_rule( + "{c_year2}{c_month:02d}{m_model}{n_count:05d}", + material, + mlog, + ) + + self.assertEqual(number, "2608P00004") + + def test_mixed_sequence_widths_use_numeric_maximum(self): + queryset = MagicMock() + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [ + "2608P9999", + "2608P10000", + "历史异常编号", + ] + material = SimpleNamespace(model="P") + mlog = SimpleNamespace( + handle_date=date(2026, 8, 6), + mgroup=SimpleNamespace(process=SimpleNamespace(id=123)), + ) + + with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset): + number = MlogbInViewSet.gen_number_with_rule( + "{c_year2}{c_month:02d}{m_model}{n_count:05d}", + material, + mlog, + ) + + self.assertEqual(number, "2608P10001") + + def test_unrelated_historical_formats_do_not_participate(self): + queryset = MagicMock() + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [ + "202505508001", + "3p05013", + "05002", + "3pb003", + ] + material = SimpleNamespace(model="P") + mlog = SimpleNamespace( + handle_date=date(2026, 8, 7), + mgroup=SimpleNamespace(process=SimpleNamespace(id=123)), + ) + + with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset): + number = MlogbInViewSet.gen_number_with_rule( + "{m_model}{n_count:04d}", + material, + mlog, + ) + + self.assertEqual(number, "P0001") + + def test_current_rule_uses_only_matching_model_numbers(self): + queryset = MagicMock() + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [ + "P0002", + "P0010", + "B0099", + "3pb100", + ] + material = SimpleNamespace(model="P") + mlog = SimpleNamespace( + handle_date=date(2026, 8, 7), + mgroup=SimpleNamespace(process=SimpleNamespace(id=123)), + ) + + with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset): + number = MlogbInViewSet.gen_number_with_rule( + "{m_model}{n_count:04d}", + material, + mlog, + ) + + self.assertEqual(number, "P0011") + + def test_empty_historical_number_is_ignored(self): + queryset = MagicMock() + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [ + None, + "", + "P0002", + ] + material = SimpleNamespace(model="P") + mlog = SimpleNamespace( + handle_date=date(2026, 8, 7), + mgroup=SimpleNamespace(process=SimpleNamespace(id=123)), + ) + + with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset): + number = MlogbInViewSet.gen_number_with_rule( + "{m_model}{n_count:04d}", + material, + mlog, + ) + + self.assertEqual(number, "P0003") + + def test_repeated_sequence_placeholder_does_not_break_matching(self): + queryset = MagicMock() + queryset.values_list.return_value.distinct.return_value.iterator.return_value = [ + "P02-0002", + "P03-0004", + ] + material = SimpleNamespace(model="P") + mlog = SimpleNamespace( + handle_date=date(2026, 8, 7), + mgroup=SimpleNamespace(process=SimpleNamespace(id=123)), + ) + + with patch("apps.wpmw.models.Wpr.objects.filter", return_value=queryset): + number = MlogbInViewSet.gen_number_with_rule( + "{m_model}{n_count:02d}-{n_count:04d}", + material, + mlog, + ) + + self.assertEqual(number, "P03-0003") diff --git a/apps/wpm/views.py b/apps/wpm/views.py index 1084145f..3c66404c 100644 --- a/apps/wpm/views.py +++ b/apps/wpm/views.py @@ -74,7 +74,6 @@ from django.db.models import Prefetch from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi from django.db import connection -from django.db.models.functions import Substr, Length from apps.qm.models import FtestDefect, FtestItem # Create your views here. @@ -1012,9 +1011,11 @@ 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 + formatter = Formatter() + rule_parts = list(formatter.parse(rule)) rule_fields = { field_name - for _, field_name, _, _ in Formatter().parse(rule) + for _, field_name, _, _ in rule_parts if field_name } handle_date = mlog.handle_date @@ -1049,18 +1050,51 @@ class MlogbInViewSet(BulkCreateModelMixin, BulkUpdateModelMixin, BulkDestroyMode 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) - .annotate(last_seq=Substr("number", Length("number") - (cq_w - 1))) - .order_by("last_seq") - .last() - ) - n_count = 0 - if wpr: + rule_values = { + "c_year": c_year, + "c_year2": c_year2, + "c_month": c_month, + "c_day": c_day, + "m_model": m_model, + } + number_pattern_parts = ["^"] + sequence_group_names = [] + for literal_text, field_name, format_spec, conversion in rule_parts: + number_pattern_parts.append(re.escape(literal_text)) + if not field_name: + continue + if field_name == "n_count": + # 流水号宽度可以变化，规则中的其他部分必须与当前上下文一致。 + group_name = f"n_count_{len(sequence_group_names)}" + sequence_group_names.append(group_name) + number_pattern_parts.append(fr"(?P<{group_name}>[0-9]+)") + continue try: - n_count = int(wpr.number[-cq_w:]) - except Exception as e: - raise ParseError(f"获取该类产品最后编号错误: {str(e)}") + field_value = rule_values[field_name] + if conversion: + field_value = formatter.convert_field(field_value, conversion) + formatted_value = formatter.format_field(field_value, format_spec) + except (KeyError, TypeError, ValueError) as e: + raise ParseError(f"个号生成错误: {e}") + number_pattern_parts.append(re.escape(formatted_value)) + number_pattern_parts.append("$") + number_pattern = re.compile("".join(number_pattern_parts)) + n_count = 0 + # 只从符合当前规则固定部分的历史编号中提取流水号。例如当前规则为 + # P{n_count:04d}时，3pb003等同工序的旧格式编号不能参与续号；同时 + # 流水号使用数字匹配，以兼容04d调整为05d后的历史编号。 + numbers = Wpr.objects.filter(**wpr_filter).values_list("number", flat=True).distinct() + for number in numbers.iterator(): + if not isinstance(number, str): + continue + sequence_match = number_pattern.fullmatch(number) + if sequence_match and sequence_group_names: + sequence_values = { + int(sequence_match.group(group_name)) + for group_name in sequence_group_names + } + if len(sequence_values) == 1: + n_count = max(n_count, sequence_values.pop()) if n_count + gen_count > 10 ** cq_w - 1: raise ParseError(f"流水号超出{cq_w}位上限, 请调整编号规则") try: