Prevent duplicate warehouse inventory creation

apps/inm/models.py

@@ -1,4 +1,6 @@
-from django.db import models
+import json
+
+from django.db import connection, models
 from apps.utils.models import BaseModel, CommonBModel, CommonBDModel, CommonADModel
 from apps.pum.models import Supplier, PuOrder
 from apps.sam.models import Customer, Order
@@ -38,6 +40,113 @@ class MaterialBatch(BaseModel):
     material_ofrom = models.ForeignKey(Material, verbose_name='原料物料', on_delete=models.SET_NULL, null=True, blank=True, related_name='mb_mofrom')
     defect = models.ForeignKey('qm.defect', verbose_name='缺陷', on_delete=models.PROTECT, null=True, blank=True)
 
+    INVENTORY_KEY_FIELDS = (
+        'material',
+        'batch',
+        'warehouse',
+        'state',
+        'defect',
+    )
+
+    @classmethod
+    def _normalize_inventory_lookup(cls, **kwargs):
+        """生成唯一、完整的仓库库存业务键。"""
+        unknown_fields = set(kwargs) - set(cls.INVENTORY_KEY_FIELDS)
+        if unknown_fields:
+            fields = ', '.join(sorted(unknown_fields))
+            raise TypeError(f'不支持的仓库库存定位字段: {fields}')
+
+        required_fields = ('material', 'batch', 'warehouse')
+        missing_fields = [
+            field for field in required_fields
+            if kwargs.get(field) is None
+        ]
+        if missing_fields:
+            fields = ', '.join(missing_fields)
+            raise ValueError(f'仓库库存业务键必须包含: {fields}')
+
+        lookup = {
+            field: kwargs.get(field)
+            for field in cls.INVENTORY_KEY_FIELDS
+        }
+        if lookup['state'] is None:
+            lookup['state'] = cls._meta.get_field('state').get_default()
+        return lookup
+
+    @classmethod
+    def _inventory_advisory_lock_payload(cls, lookup):
+        lock_values = {}
+        for name in cls.INVENTORY_KEY_FIELDS:
+            field = cls._meta.get_field(name)
+            value = lookup[name]
+            if field.is_relation and value is not None:
+                value = getattr(value, 'pk', value)
+            lock_values[field.attname] = value
+        return json.dumps(
+            {'model': cls._meta.label_lower, 'lookup': lock_values},
+            sort_keys=True,
+            ensure_ascii=False,
+            default=str,
+            separators=(',', ':'),
+        )
+
+    @classmethod
+    def locked_get_or_create_inventory(cls, defaults=None, **kwargs):
+        """
+        在事务中按完整业务键获取或创建仓库库存。
+
+        已存在记录使用行锁；首次创建使用 PostgreSQL 事务级 advisory lock，
+        并在取得锁后重新查询，避免并发创建重复库存。
+        """
+        if not connection.in_atomic_block:
+            raise RuntimeError(
+                'locked_get_or_create_inventory 必须在事务中调用'
+            )
+        if connection.vendor != 'postgresql':
+            raise RuntimeError(
+                'locked_get_or_create_inventory 仅支持 PostgreSQL'
+            )
+
+        defaults = defaults or {}
+        lookup = cls._normalize_inventory_lookup(**kwargs)
+        create_defaults = {
+            key: value
+            for key, value in defaults.items()
+            if key not in cls.INVENTORY_KEY_FIELDS
+        }
+
+        rows = list(
+            cls.objects.select_for_update().filter(**lookup)[:2]
+        )
+        if len(rows) > 1:
+            raise RuntimeError(
+                f'{cls.__name__} 数据异常：库存业务键 {lookup} 命中多条'
+            )
+        if rows:
+            return rows[0], False
+
+        lock_payload = cls._inventory_advisory_lock_payload(lookup)
+        with connection.cursor() as cursor:
+            cursor.execute(
+                'SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))',
+                [lock_payload],
+            )
+
+        rows = list(
+            cls.objects.select_for_update().filter(**lookup)[:2]
+        )
+        if len(rows) > 1:
+            raise RuntimeError(
+                f'{cls.__name__} 数据异常：库存业务键 {lookup} 命中多条'
+            )
+        if rows:
+            return rows[0], False
+
+        return cls.objects.create(
+            **lookup,
+            **create_defaults,
+        ), True
+
 
     @property
     def count_mioing(self): 

apps/inm/services.py

@@ -223,7 +223,7 @@ def do_in(item: MIOItem):
         
         # 增加mb
         if not is_zhj:
-            mb, _ = MaterialBatch.objects.get_or_create(
+            mb, _ = MaterialBatch.locked_get_or_create_inventory(
                     material=xmaterial,
                     warehouse=item.warehouse,
                     batch=xbatch,
@@ -258,7 +258,7 @@ def do_in(item: MIOItem):
 
 
     if is_zhj: # 组合件单独处理并且不做追踪单个处理
-        mb, is_created = MaterialBatch.objects.get_or_create(
+        mb, is_created = MaterialBatch.locked_get_or_create_inventory(
                     material=item.material, 
                     warehouse=item.warehouse, 
                     batch=item.batch,
@@ -415,7 +415,7 @@ class InmService:
             state = WMaterial.WM_OK
             if defect and defect.okcate in [Defect.DEFECT_NOTOK]:
                 state = WMaterial.WM_NOTOK
-            mb, _ = MaterialBatch.objects.get_or_create(
+            mb, _ = MaterialBatch.locked_get_or_create_inventory(
                 material=material, 
                 warehouse=warehouse, 
                 batch=batch, 

apps/inm/services_daoru.py

@@ -55,7 +55,7 @@ def daoru_mb(path: str):
                 process=process,
                 defaults={"type": type, "name": name, "specification": specification, "model": model, "process": process, "number": ranstr(6), "id": idWorker.get_id()},
             )
-            MaterialBatch.objects.get_or_create(
+            MaterialBatch.locked_get_or_create_inventory(
                 material=material, batch=batch, warehouse=warehouse, defaults={"material": material, "batch": batch, "warehouse": warehouse, "count": count, "id": idWorker.get_id()}
             )
             i = i + 1
@@ -174,4 +174,4 @@ def daoru_mioitems(path:str, mio:MIO):
                                 unit_price=material.unit_price, id=idWorker.get_id()))
         ind = ind + 1
     
-    MIOItem.objects.bulk_create(mioitems)
+    MIOItem.objects.bulk_create(mioitems)

apps/inm/tests.py

@@ -1,3 +1,115 @@
-from django.test import TestCase
+from concurrent.futures import ThreadPoolExecutor
+from decimal import Decimal
+from threading import Barrier
+from unittest import skipUnless
 
-# Create your tests here.
+from django.db import connection, connections, transaction
+from django.test import SimpleTestCase, TransactionTestCase
+
+from apps.inm.models import MaterialBatch, WareHouse
+from apps.mtm.models import Material
+
+
+class MaterialBatchInventoryKeyTests(SimpleTestCase):
+    def setUp(self):
+        self.material = Material(id='100', name='测试物料')
+        self.warehouse = WareHouse(
+            id='200',
+            number='TEST',
+            name='测试仓库',
+            place='测试地点',
+        )
+
+    def test_inventory_key_normalizes_omitted_optional_fields(self):
+        omitted = MaterialBatch._normalize_inventory_lookup(
+            material=self.material,
+            batch='BATCH-001',
+            warehouse=self.warehouse,
+        )
+        explicit = MaterialBatch._normalize_inventory_lookup(
+            material=self.material,
+            batch='BATCH-001',
+            warehouse=self.warehouse,
+            state=10,
+            defect=None,
+        )
+
+        self.assertEqual(omitted, explicit)
+        self.assertEqual(
+            MaterialBatch._inventory_advisory_lock_payload(omitted),
+            MaterialBatch._inventory_advisory_lock_payload(explicit),
+        )
+
+    def test_inventory_key_requires_material_batch_and_warehouse(self):
+        required_values = {
+            'material': self.material,
+            'batch': 'BATCH-001',
+            'warehouse': self.warehouse,
+        }
+
+        for field in required_values:
+            with self.subTest(field=field):
+                lookup = required_values.copy()
+                lookup[field] = None
+                with self.assertRaisesRegex(ValueError, field):
+                    MaterialBatch._normalize_inventory_lookup(**lookup)
+
+    def test_inventory_key_rejects_unknown_fields(self):
+        with self.assertRaisesRegex(TypeError, 'supplier'):
+            MaterialBatch._normalize_inventory_lookup(
+                material=self.material,
+                batch='BATCH-001',
+                warehouse=self.warehouse,
+                supplier=None,
+            )
+
+
+@skipUnless(
+    connection.vendor == 'postgresql',
+    'advisory lock concurrency test requires PostgreSQL',
+)
+class MaterialBatchConcurrencyTests(TransactionTestCase):
+    def test_concurrent_first_create_uses_one_inventory_record(self):
+        material = Material.objects.create(name='仓库并发测试物料')
+        warehouse = WareHouse.objects.create(
+            number='CONCURRENT',
+            name='并发测试仓库',
+            place='测试地点',
+        )
+        barrier = Barrier(2)
+
+        def create_inventory():
+            connections.close_all()
+            try:
+                barrier.wait()
+                with transaction.atomic():
+                    mb, created = (
+                        MaterialBatch.locked_get_or_create_inventory(
+                            material=material,
+                            batch='CONCURRENT-001',
+                            warehouse=warehouse,
+                            defaults={'count': Decimal('0')},
+                        )
+                    )
+                    mb.count += Decimal('1')
+                    mb.save(update_fields=['count'])
+                    return created
+            finally:
+                connections.close_all()
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            created_results = list(executor.map(
+                lambda _: create_inventory(),
+                range(2),
+            ))
+
+        queryset = MaterialBatch.objects.filter(
+            material=material,
+            batch='CONCURRENT-001',
+            warehouse=warehouse,
+            state=10,
+            defect=None,
+        )
+        self.assertEqual(queryset.count(), 1)
+        self.assertEqual(queryset.get().count, Decimal('2'))
+        self.assertCountEqual(created_results, [True, False])

apps/qm/services.py

@@ -146,7 +146,7 @@ def ftestwork_submit(ins:FtestWork, user: User):
                 mbstate = WMaterial.WM_OK
                 if item.defect.okcate == Defect.DEFECT_NOTOK:
                     mbstate = WMaterial.WM_NOTOK
-                mbx, new_create = MaterialBatch.objects.get_or_create(
+                mbx, new_create = MaterialBatch.locked_get_or_create_inventory(
                     material=mb.material, 
                     warehouse=mb.warehouse, 
                     batch=mb.batch,