feat: balance api

2025-11-30 23:04:21 +08:00
parent 9006a530d1
commit 9acc4e14fc
21 changed files with 714 additions and 64 deletions
--- a/business/ARCHITECTURE.md
+++ b/business/ARCHITECTURE.md
@@ -39,6 +39,7 @@
 - **SalesOrder**（销售单）：`direction=-1`，`counterparty=customer`，含库存明细；严进严出模式需记录 `consume_detail_ids`。
 - **PaymentOrder**（付款单）：`direction=-1`，`counterparty=supplier`，仅金额字段，不触发库存。
 - **ReceiptOrder**（收款单）：`direction=-1`，`counterparty=customer`，仅金额字段，不触发库存。
+- **SupplierBalance / CustomerBalance**：实时维护供应商应付、客户应收余额，所有审批通过的带金额单据都会写入，供查询接口和报表使用。

 通过 mixin，所有单据都具备：
 - 统一的金额聚合与方向计算（资金/库存可共用 `get_signed_total_amount()`）
--- a/business/migrations/0012_customerbalance_supplierbalance.py
+++ b/business/migrations/0012_customerbalance_supplierbalance.py
@@ -0,0 +1,48 @@
+# Generated by Django 5.2.7 on 2025-11-30 14:17
+
+import django.db.models.deletion
+from decimal import Decimal
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('basic_info', '0016_merchantsetting_type'),
+        ('business', '0011_paymentorder_receiptorder'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CustomerBalance',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
+                ('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
+                ('balance', models.DecimalField(decimal_places=2, default=Decimal('0'), max_digits=15, verbose_name='应收余额')),
+                ('customer', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='balances', to='basic_info.customer', verbose_name='客户')),
+                ('merchant', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='customer_balances', to='basic_info.merchant', verbose_name='所属商户')),
+            ],
+            options={
+                'verbose_name': '客户余额',
+                'verbose_name_plural': '客户余额',
+                'unique_together': {('merchant', 'customer')},
+            },
+        ),
+        migrations.CreateModel(
+            name='SupplierBalance',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
+                ('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
+                ('balance', models.DecimalField(decimal_places=2, default=Decimal('0'), max_digits=15, verbose_name='应付余额')),
+                ('merchant', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='supplier_balances', to='basic_info.merchant', verbose_name='所属商户')),
+                ('supplier', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='balances', to='basic_info.supplier', verbose_name='供应商')),
+            ],
+            options={
+                'verbose_name': '供应商余额',
+                'verbose_name_plural': '供应商余额',
+                'unique_together': {('merchant', 'supplier')},
+            },
+        ),
+    ]
--- a/business/models.py
+++ b/business/models.py
@@ -420,3 +420,45 @@ class ReceiptOrder(OrderDirectionMixin, OrderCounterpartyMixin, ModelBase):

    def get_counterparty_field_name(self) -> str:
        return 'customer'
+
+
+class SupplierBalance(ModelBase):
+    merchant = models.ForeignKey(
+        basic_info_models.Merchant,
+        on_delete=models.PROTECT,
+        related_name='supplier_balances',
+        verbose_name='所属商户',
+    )
+    supplier = models.ForeignKey(
+        basic_info_models.Supplier,
+        on_delete=models.PROTECT,
+        related_name='balances',
+        verbose_name='供应商',
+    )
+    balance = models.DecimalField(max_digits=15, decimal_places=2, default=Decimal('0'), verbose_name='应付余额')
+
+    class Meta:
+        verbose_name = '供应商余额'
+        verbose_name_plural = '供应商余额'
+        unique_together = ('merchant', 'supplier')
+
+
+class CustomerBalance(ModelBase):
+    merchant = models.ForeignKey(
+        basic_info_models.Merchant,
+        on_delete=models.PROTECT,
+        related_name='customer_balances',
+        verbose_name='所属商户',
+    )
+    customer = models.ForeignKey(
+        basic_info_models.Customer,
+        on_delete=models.PROTECT,
+        related_name='balances',
+        verbose_name='客户',
+    )
+    balance = models.DecimalField(max_digits=15, decimal_places=2, default=Decimal('0'), verbose_name='应收余额')
+
+    class Meta:
+        verbose_name = '客户余额'
+        verbose_name_plural = '客户余额'
+        unique_together = ('merchant', 'customer')
--- a/business/services.py
+++ b/business/services.py
@@ -18,6 +18,66 @@ from .tasks import (
    create_purchase_order_stock_entries,
    create_sales_order_stock_entries,
 )
+class BalanceService:
+    @staticmethod
+    def adjust_supplier_balance(
+        *,
+        merchant: basic_info_models.Merchant,
+        supplier: basic_info_models.Supplier,
+        delta: Decimal,
+    ):
+        with transaction.atomic():
+            balance, _ = models.SupplierBalance.objects.select_for_update().get_or_create(
+                merchant=merchant,
+                supplier=supplier,
+                defaults={'balance': Decimal('0')},
+            )
+            balance.balance += delta
+            balance.save(update_fields=['balance', 'updated_at'])
+
+    @staticmethod
+    def adjust_customer_balance(
+        *,
+        merchant: basic_info_models.Merchant,
+        customer: basic_info_models.Customer,
+        delta: Decimal,
+    ):
+        with transaction.atomic():
+            balance, _ = models.CustomerBalance.objects.select_for_update().get_or_create(
+                merchant=merchant,
+                customer=customer,
+                defaults={'balance': Decimal('0')},
+            )
+            balance.balance += delta
+            balance.save(update_fields=['balance', 'updated_at'])
+
+    @staticmethod
+    def get_customer_balance(
+        *,
+        merchant: basic_info_models.Merchant,
+        customer: basic_info_models.Customer,
+    ) -> Decimal:
+        balance = models.CustomerBalance.objects.filter(
+            merchant=merchant,
+            customer=customer,
+        ).first()
+        if balance is None:
+            return Decimal('0')
+        return balance.balance
+
+    @staticmethod
+    def get_supplier_balance(
+        *,
+        merchant: basic_info_models.Merchant,
+        supplier: basic_info_models.Supplier,
+    ) -> Decimal:
+        balance = models.SupplierBalance.objects.filter(
+            merchant=merchant,
+            supplier=supplier,
+        ).first()
+        if balance is None:
+            return Decimal('0')
+        return balance.balance


 logger = logging.getLogger(__name__)
@@ -296,12 +356,37 @@ def review_payment_order(
    if order.status == target_status:
        return order

-    with transaction.atomic():
-        order.status = target_status
-        order.save(update_fields=['status', 'updated_at'])
+    if target_status == models.PaymentOrderStatusEnum.APPROVED:
+        if order.status == models.PaymentOrderStatusEnum.CANCELLED:
+            raise ValueError('作废状态的付款单无法再次审批')
+        with transaction.atomic():
+            locked = models.PaymentOrder.objects.select_related(
+                'merchant', 'supplier'
+            ).select_for_update().get(id=order.id)
+            if locked.status == models.PaymentOrderStatusEnum.APPROVED:
+                return locked
+            if locked.status == models.PaymentOrderStatusEnum.CANCELLED:
+                raise ValueError('作废状态的付款单无法再次审批')
+            locked.status = models.PaymentOrderStatusEnum.APPROVED
+            locked.save(update_fields=['status', 'updated_at'])
+            BalanceService.adjust_supplier_balance(
+                merchant=locked.merchant,
+                supplier=locked.supplier,
+                delta=-locked.amount,
+            )
+        locked.refresh_from_db(fields=['status', 'updated_at'])
+        return locked

-    order.refresh_from_db(fields=['status', 'updated_at'])
-    return order
+    if order.status == models.PaymentOrderStatusEnum.APPROVED:
+        raise ValueError('已审批的付款单无法作废')
+
+    with transaction.atomic():
+        locked = models.PaymentOrder.objects.select_for_update().get(id=order.id)
+        locked.status = models.PaymentOrderStatusEnum.CANCELLED
+        locked.save(update_fields=['status', 'updated_at'])
+
+    locked.refresh_from_db(fields=['status', 'updated_at'])
+    return locked


 def review_receipt_order(
@@ -322,12 +407,37 @@ def review_receipt_order(
    if order.status == target_status:
        return order

-    with transaction.atomic():
-        order.status = target_status
-        order.save(update_fields=['status', 'updated_at'])
+    if target_status == models.ReceiptOrderStatusEnum.APPROVED:
+        if order.status == models.ReceiptOrderStatusEnum.CANCELLED:
+            raise ValueError('作废状态的收款单无法再次审批')
+        with transaction.atomic():
+            locked = models.ReceiptOrder.objects.select_related(
+                'merchant', 'customer'
+            ).select_for_update().get(id=order.id)
+            if locked.status == models.ReceiptOrderStatusEnum.APPROVED:
+                return locked
+            if locked.status == models.ReceiptOrderStatusEnum.CANCELLED:
+                raise ValueError('作废状态的收款单无法再次审批')
+            locked.status = models.ReceiptOrderStatusEnum.APPROVED
+            locked.save(update_fields=['status', 'updated_at'])
+            BalanceService.adjust_customer_balance(
+                merchant=locked.merchant,
+                customer=locked.customer,
+                delta=-locked.amount,
+            )
+        locked.refresh_from_db(fields=['status', 'updated_at'])
+        return locked

-    order.refresh_from_db(fields=['status', 'updated_at'])
-    return order
+    if order.status == models.ReceiptOrderStatusEnum.APPROVED:
+        raise ValueError('已审批的收款单无法作废')
+
+    with transaction.atomic():
+        locked = models.ReceiptOrder.objects.select_for_update().get(id=order.id)
+        locked.status = models.ReceiptOrderStatusEnum.CANCELLED
+        locked.save(update_fields=['status', 'updated_at'])
+
+    locked.refresh_from_db(fields=['status', 'updated_at'])
+    return locked


 def _normalize_order_items(
@@ -449,80 +559,112 @@ def _approve_purchase_order(
    purchase_order: models.PurchaseOrder,
    reviewed_by,
 ) -> models.PurchaseOrder:
-    stock_flow_items = _build_stock_flow_items_from_order(purchase_order)
-
    with transaction.atomic():
-        purchase_order.status = models.PurchaseOrderStatusEnum.APPROVED
-        purchase_order.save(update_fields=['status', 'updated_at'])
+        locked_order = models.PurchaseOrder.objects.select_related(
+            'merchant', 'warehouse', 'supplier'
+        ).prefetch_related('items').select_for_update().get(id=purchase_order.id)
+
+        if locked_order.status == models.PurchaseOrderStatusEnum.APPROVED:
+            return locked_order
+        if locked_order.status == models.PurchaseOrderStatusEnum.CANCELLED:
+            raise ValueError('作废状态的采购单无法再次审批')
+
+        stock_flow_items = _build_stock_flow_items_from_order(locked_order)
+
+        locked_order.status = models.PurchaseOrderStatusEnum.APPROVED
+        locked_order.save(update_fields=['status', 'updated_at'])
+        BalanceService.adjust_supplier_balance(
+            merchant=locked_order.merchant,
+            supplier=locked_order.supplier,
+            delta=locked_order.get_total_amount(),
+        )

    created_by_id = getattr(reviewed_by, 'id', None)
-    if _auto_stock_task_enabled(purchase_order.merchant):
-        logger.info('审批通过采购单 %s，触发入库任务', purchase_order.id)
+    if _auto_stock_task_enabled(locked_order.merchant):
+        logger.info('审批通过采购单 %s，触发入库任务', locked_order.id)
        create_purchase_order_stock_entries.delay(
-            purchase_order_id=purchase_order.id,
-            warehouse_id=purchase_order.warehouse_id,
+            purchase_order_id=locked_order.id,
+            warehouse_id=locked_order.warehouse_id,
            items=stock_flow_items,
            created_by_id=created_by_id,
        )

-    purchase_order.refresh_from_db(fields=['status', 'updated_at'])
-    return purchase_order
+    locked_order.refresh_from_db(fields=['status', 'updated_at'])
+    return locked_order


 def _cancel_purchase_order(purchase_order: models.PurchaseOrder) -> models.PurchaseOrder:
-    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():
-        purchase_order.status = models.PurchaseOrderStatusEnum.CANCELLED
-        purchase_order.save(update_fields=['status', 'updated_at'])
-
-    purchase_order.refresh_from_db(fields=['status', 'updated_at'])
-    return purchase_order
+        locked = models.PurchaseOrder.objects.select_related('merchant').select_for_update().get(id=purchase_order.id)
+        if locked.status == models.PurchaseOrderStatusEnum.APPROVED:
+            raise ValueError('已审批的采购单无法作废')
+        if _order_has_stock_records(
+            merchant_id=locked.merchant_id,
+            source_type=stock_models.StockChangeSourceEnum.PURCHASE,
+            source_id=locked.id,
+        ):
+            raise ValueError('采购单已生成出入库记录，无法作废')
+        locked.status = models.PurchaseOrderStatusEnum.CANCELLED
+        locked.save(update_fields=['status', 'updated_at'])
+    locked.refresh_from_db(fields=['status', 'updated_at'])
+    return locked


 def _approve_sales_order(
    sales_order: models.SalesOrder,
    reviewed_by,
 ) -> models.SalesOrder:
-    stock_flow_items = _build_stock_flow_items_from_order(sales_order)
-
    with transaction.atomic():
-        sales_order.status = models.SalesOrderStatusEnum.APPROVED
-        sales_order.save(update_fields=['status', 'updated_at'])
+        locked_order = models.SalesOrder.objects.select_related(
+            'merchant', 'warehouse', 'customer'
+        ).prefetch_related('items').select_for_update().get(id=sales_order.id)
+
+        if locked_order.status == models.SalesOrderStatusEnum.APPROVED:
+            return locked_order
+        if locked_order.status == models.SalesOrderStatusEnum.CANCELLED:
+            raise ValueError('作废状态的销售单无法再次审批')
+
+        stock_flow_items = _build_stock_flow_items_from_order(locked_order)
+
+        locked_order.status = models.SalesOrderStatusEnum.APPROVED
+        locked_order.save(update_fields=['status', 'updated_at'])
+        BalanceService.adjust_customer_balance(
+            merchant=locked_order.merchant,
+            customer=locked_order.customer,
+            delta=locked_order.get_total_amount(),
+        )

    created_by_id = getattr(reviewed_by, 'id', None)
-    if _auto_stock_task_enabled(sales_order.merchant):
-        logger.info('审批通过销售单 %s，触发出库任务', sales_order.id)
+    if _auto_stock_task_enabled(locked_order.merchant):
+        logger.info('审批通过销售单 %s，触发出库任务', locked_order.id)
        create_sales_order_stock_entries.delay(
-            sales_order_id=sales_order.id,
-            warehouse_id=sales_order.warehouse_id,
+            sales_order_id=locked_order.id,
+            warehouse_id=locked_order.warehouse_id,
            items=stock_flow_items,
            created_by_id=created_by_id,
        )

-    sales_order.refresh_from_db(fields=['status', 'updated_at'])
-    return sales_order
+    locked_order.refresh_from_db(fields=['status', 'updated_at'])
+    return locked_order


 def _cancel_sales_order(sales_order: models.SalesOrder) -> models.SalesOrder:
-    if _order_has_stock_records(
-        merchant_id=sales_order.merchant_id,
-        source_type=stock_models.StockChangeSourceEnum.SALES,
-        source_id=sales_order.id,
-    ):
-        raise ValueError('销售单已生成出入库记录，无法作废')
-
    with transaction.atomic():
-        sales_order.status = models.SalesOrderStatusEnum.CANCELLED
-        sales_order.save(update_fields=['status', 'updated_at'])
+        locked = models.SalesOrder.objects.select_related('merchant').select_for_update().get(id=sales_order.id)
+        if locked.status == models.SalesOrderStatusEnum.APPROVED:
+            raise ValueError('已审批的销售单无法作废')
+        if _order_has_stock_records(
+            merchant_id=locked.merchant_id,
+            source_type=stock_models.StockChangeSourceEnum.SALES,
+            source_id=locked.id,
+        ):
+            raise ValueError('销售单已生成出入库记录，无法作废')

-    sales_order.refresh_from_db(fields=['status', 'updated_at'])
-    return sales_order
+        locked.status = models.SalesOrderStatusEnum.CANCELLED
+        locked.save(update_fields=['status', 'updated_at'])
+
+    locked.refresh_from_db(fields=['status', 'updated_at'])
+    return locked


 def create_purchase_order_stock_entries_sync(
--- a/business/tests.py
+++ b/business/tests.py
@@ -1,7 +1,10 @@
 from typing import List, Dict, Any
+from decimal import Decimal
+from concurrent.futures import ThreadPoolExecutor

 from django.contrib.auth import get_user_model
-from django.test import TestCase, override_settings
+from django.test import TestCase, TransactionTestCase, override_settings
+from django.db import connections
 from django.utils import timezone
 from unittest.mock import patch, MagicMock

@@ -171,6 +174,11 @@ class PurchaseOrderServiceTestCase(TestCase):
            items=[{'product_id': self.product.id, 'quantities': ['10', '5']}],
            created_by_id=self.user.id,
        )
+        balance = business_models.SupplierBalance.objects.get(
+            merchant=self.merchant,
+            supplier=self.supplier,
+        )
+        self.assertEqual(balance.balance, purchase_order.get_total_amount())

    def test_create_purchase_order_without_items_raises(self):
        with self.assertRaises(ValueError):
@@ -324,6 +332,11 @@ class SalesOrderServiceTestCase(TestCase):
            items=[{'product_id': self.product.id, 'quantities': ['6', '4']}],
            created_by_id=self.user.id,
        )
+        balance = business_models.CustomerBalance.objects.get(
+            merchant=self.merchant,
+            customer=self.customer,
+        )
+        self.assertEqual(balance.balance, sales_order.get_total_amount())

    def test_sales_order_cancel_blocked_after_stock_created(self):
        sales_order = services.create_sales_order(
@@ -392,7 +405,6 @@ class SalesOrderServiceTestCase(TestCase):
                created_by=self.user,
            )

-
 class PaymentReceiptServiceTestCase(TestCase):
    def setUp(self):
        (
@@ -426,6 +438,17 @@ class PaymentReceiptServiceTestCase(TestCase):
            reviewed_by=self.operator,
        )
        self.assertEqual(reviewed.status, business_models.PaymentOrderStatusEnum.APPROVED)
+        balance = business_models.SupplierBalance.objects.get(
+            merchant=self.merchant,
+            supplier=self.supplier,
+        )
+        self.assertEqual(balance.balance, Decimal('-120.50'))
+        with self.assertRaises(ValueError):
+            services.review_payment_order(
+                payment_order=order,
+                target_status=business_models.PaymentOrderStatusEnum.CANCELLED,
+                reviewed_by=self.operator,
+            )

    def test_create_receipt_order_and_cancel(self):
        order = services.create_receipt_order(
@@ -472,6 +495,11 @@ class PaymentReceiptServiceTestCase(TestCase):
            reviewed_by=self.operator,
        )
        self.assertEqual(reviewed_again.status, business_models.ReceiptOrderStatusEnum.APPROVED)
+        balance = business_models.CustomerBalance.objects.get(
+            merchant=self.merchant,
+            customer=self.customer,
+        )
+        self.assertEqual(balance.balance, Decimal('-10'))


 class PurchaseOrderStockServiceTestCase(TestCase):
@@ -593,6 +621,58 @@ class SalesOrderStockServiceTestCase(TestCase):
        self.assertEqual(payload['stock_change_record_id'], 987)


+class SalesOrderConcurrencyTestCase(TransactionTestCase):
+    reset_sequences = True
+
+    def setUp(self):
+        (
+            self.merchant,
+            self.customer,
+            self.warehouse_strict,
+            self.warehouse_relaxed,
+            self.warehouse_strict_out,
+            self.product,
+            self.operator,
+        ) = create_sales_fixtures()
+        basic_models.MerchantSetting.objects.filter(
+            merchant=self.merchant,
+            key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
+        ).update(val_bool=False)
+        User = get_user_model()
+        self.user = User.objects.create_user(username='concurrent', password='pass123')
+        self.sales_order = services.create_sales_order(
+            merchant=self.merchant,
+            customer=self.customer,
+            order_date=timezone.now().date(),
+            warehouse=self.warehouse_strict,
+            operator=self.operator,
+            items=[{'product_id': self.product.id, 'numbers': [5], 'price': '12', 'unit': '米'}],
+            created_by=self.user,
+        )
+
+    def test_concurrent_sales_order_approval_updates_balance_once(self):
+        def approve():
+            services.review_sales_order(
+                sales_order_id=self.sales_order.id,
+                target_status=business_models.SalesOrderStatusEnum.APPROVED,
+                reviewed_by=self.user,
+            )
+
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            futures = [executor.submit(approve) for _ in range(2)]
+            for future in futures:
+                future.result()
+
+        balance = business_models.CustomerBalance.objects.get(
+            merchant=self.merchant,
+            customer=self.customer,
+        )
+        self.assertEqual(balance.balance, self.sales_order.get_total_amount())
+
+    def tearDown(self):
+        connections.close_all()
+
+
@override_settings(
    CELERY_TASK_ALWAYS_EAGER=True,
    CELERY_TASK_EAGER_PROPAGATES=True,