feat: added filter param for stock record list api. Support purchase order bind stock change record that is already exists

business/services.py

@@ -1220,7 +1220,7 @@ def _approve_purchase_order(
     created_by_id = getattr(reviewed_by, 'id', None)
     if _auto_stock_task_enabled(locked_order.merchant):
         logger.info('审批通过采购单 %s，触发入库任务', locked_order.id)
-    create_purchase_order_stock_entries.delay(
+        create_purchase_order_stock_entries.delay(
             purchase_order_id=locked_order.id,
             warehouse_id=locked_order.warehouse_id,
             items=stock_flow_items,
@@ -1659,6 +1659,65 @@ def _build_stock_flow_items_from_order(order) -> List[Dict[str, Any]]:
     return items_payload
 
 
+def can_bind_source_id(record: stock_models.StockChangeRecord) -> bool:
+    """
+    判断库存记录是否允许绑定 source_id。
+    目前默认允许，后续可按业务需要扩展。
+    """
+    return True
+
+
+def bind_purchase_order_stock_change_record(
+    *,
+    purchase_order_id: int,
+    stock_change_record_id: int,
+    operator=None,
+) -> stock_models.StockChangeRecord:
+    purchase_order = models.PurchaseOrder.objects.select_related(
+        'merchant', 'warehouse'
+    ).get(id=purchase_order_id)
+
+    if _auto_stock_task_enabled(purchase_order.merchant):
+        raise ValueError('自动入库任务已开启，无法手动绑定库存记录')
+
+    if _order_has_stock_records(
+        merchant_id=purchase_order.merchant_id,
+        source_type=stock_models.StockChangeSourceEnum.PURCHASE,
+        source_id=purchase_order.id,
+    ):
+        raise ValueError('采购单已绑定库存记录')
+
+    with transaction.atomic():
+        record = stock_models.StockChangeRecord.objects.select_for_update().get(
+            id=stock_change_record_id
+        )
+
+        if record.source_id is not None:
+            raise ValueError('库存记录已绑定业务单据')
+        if record.merchant_id != purchase_order.merchant_id:
+            raise ValueError('库存记录不属于当前商户')
+        if record.warehouse_id != purchase_order.warehouse_id:
+            raise ValueError('库存记录仓库不匹配')
+        if record.type != stock_models.StockChangeTypeEnum.ADD:
+            raise ValueError('仅允许绑定入库类型库存记录')
+        if record.source_type != stock_models.StockChangeSourceEnum.PURCHASE:
+            raise ValueError('仅允许绑定采购来源的库存记录')
+        if not can_bind_source_id(record):
+            raise ValueError('库存记录当前状态不允许绑定')
+
+        record.source_id = purchase_order.id
+        record.save(update_fields=['source_id', 'updated_at'])
+
+    operator_id = getattr(operator, 'id', None)
+    logger.info(
+        '采购单 %s 绑定库存记录 %s, operator=%s',
+        purchase_order.id,
+        record.id,
+        operator_id,
+    )
+    return record
+
+
 def _auto_stock_task_enabled(merchant: basic_info_models.Merchant) -> bool:
     try:
         setting = MerchantSettingService.get_setting(

business/tests/test_purchase_order.py

@@ -7,6 +7,7 @@ from django.utils import timezone
 from unittest.mock import patch
 
 from stock import models as stock_models
+from basic_info import models as basic_models
 from business import models as business_models, services
 
 from .fixtures import create_basic_fixtures
@@ -108,6 +109,63 @@ class PurchaseOrderServiceTestCase(TestCase):
         self.assertEqual(record.balance_after, balance.balance)
         self.assertEqual(record.direction, business_models.BalanceChangeDirectionEnum.INCREASE)
 
+    def test_review_purchase_order_approval_no_task_when_auto_disabled(self):
+        basic_models.MerchantSetting.objects.filter(
+            merchant=self.merchant,
+            key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
+        ).update(val_bool=False)
+
+        purchase_order = services.create_purchase_order(
+            merchant=self.merchant,
+            supplier=self.supplier,
+            order_date=timezone.now().date(),
+            warehouse=self.warehouse_strict,
+            operator=self.operator,
+            items=self.strict_items,
+            created_by=self.user,
+        )
+
+        with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay:
+            reviewed = services.review_purchase_order(
+                purchase_order=purchase_order,
+                target_status=business_models.PurchaseOrderStatusEnum.APPROVED,
+                reviewed_by=self.user,
+            )
+
+        self.assertEqual(reviewed.status, business_models.PurchaseOrderStatusEnum.APPROVED)
+        mock_delay.assert_not_called()
+
+    def test_bind_purchase_order_stock_change_record(self):
+        basic_models.MerchantSetting.objects.filter(
+            merchant=self.merchant,
+            key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
+        ).update(val_bool=False)
+
+        purchase_order = services.create_purchase_order(
+            merchant=self.merchant,
+            supplier=self.supplier,
+            order_date=timezone.now().date(),
+            warehouse=self.warehouse_strict,
+            operator=self.operator,
+            items=self.strict_items,
+            created_by=self.user,
+        )
+
+        record = stock_models.StockChangeRecord.objects.create(
+            merchant=self.merchant,
+            type=stock_models.StockChangeTypeEnum.ADD,
+            warehouse=self.warehouse_strict,
+            source_type=stock_models.StockChangeSourceEnum.PURCHASE,
+            created_by=self.user,
+        )
+
+        bound = services.bind_purchase_order_stock_change_record(
+            purchase_order_id=purchase_order.id,
+            stock_change_record_id=record.id,
+            operator=self.user,
+        )
+        self.assertEqual(bound.source_id, purchase_order.id)
+
     def test_create_purchase_order_without_items_raises(self):
         with self.assertRaises(ValueError):
             services.create_purchase_order(