from __future__ import annotations import logging from datetime import date, datetime from decimal import Decimal, InvalidOperation from typing import Any, Dict, List, Tuple from django.contrib.auth import get_user_model from django.db import transaction from basic_info import models as basic_info_models from stock import models as stock_models from stock.services import StockFlowService from basic_info.services import MerchantSettingService from . import models from .tasks import ( create_purchase_order_stock_entries, create_sales_order_stock_entries, create_purchase_return_order_stock_entries, create_sales_return_order_stock_entries, ) class BalanceService: @staticmethod def adjust_supplier_balance( *, merchant: basic_info_models.Merchant, supplier: basic_info_models.Supplier, delta: Decimal, source_type: models.BalanceChangeSourceEnum, source_id: int, request_id: str | None = None, remarks: str | None = '', extra_meta: Dict[str, Any] | None = None, ): BalanceService._adjust_balance( merchant=merchant, counterparty=supplier, balance_model=models.SupplierBalance, balance_field='supplier', delta=delta, target_type=models.BalanceChangeTargetEnum.SUPPLIER, source_type=source_type, source_id=source_id, request_id=request_id, remarks=remarks, extra_meta=extra_meta, ) @staticmethod def adjust_customer_balance( *, merchant: basic_info_models.Merchant, customer: basic_info_models.Customer, delta: Decimal, source_type: models.BalanceChangeSourceEnum, source_id: int, request_id: str | None = None, remarks: str | None = '', extra_meta: Dict[str, Any] | None = None, ): BalanceService._adjust_balance( merchant=merchant, counterparty=customer, balance_model=models.CustomerBalance, balance_field='customer', delta=delta, target_type=models.BalanceChangeTargetEnum.CUSTOMER, source_type=source_type, source_id=source_id, request_id=request_id, remarks=remarks, extra_meta=extra_meta, ) @staticmethod def _adjust_balance( *, merchant: basic_info_models.Merchant, counterparty, balance_model, balance_field: str, delta: Decimal, target_type: models.BalanceChangeTargetEnum, source_type: models.BalanceChangeSourceEnum, source_id: int, request_id: str | None, remarks: str | None, extra_meta: Dict[str, Any] | None, ): meta_payload = extra_meta or {} remarks_value = remarks or '' with transaction.atomic(): balance, _ = balance_model.objects.select_for_update().get_or_create( merchant=merchant, defaults={'balance': Decimal('0')}, **{balance_field: counterparty}, ) before = balance.balance balance.balance += delta balance.save(update_fields=['balance', 'updated_at']) record_kwargs = { 'merchant': merchant, 'target_type': target_type, 'source_type': source_type, 'source_id': source_id, 'delta': delta, 'direction': ( models.BalanceChangeDirectionEnum.INCREASE if delta >= 0 else models.BalanceChangeDirectionEnum.DECREASE ), 'balance_before': before, 'balance_after': balance.balance, 'request_id': request_id, 'remarks': remarks_value, 'extra_meta': meta_payload, } if target_type == models.BalanceChangeTargetEnum.SUPPLIER: record_kwargs['supplier'] = counterparty else: record_kwargs['customer'] = counterparty models.BalanceChangeRecord.objects.create(**record_kwargs) @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__) def _normalize_order_date(value) -> date: if isinstance(value, date): return value if isinstance(value, datetime): return value.date() if isinstance(value, str): try: return date.fromisoformat(value) except ValueError as exc: raise ValueError('order_date 格式不正确，应为 YYYY-MM-DD') from exc raise ValueError('order_date 格式不正确') def create_purchase_order( *, merchant: basic_info_models.Merchant, supplier: basic_info_models.Supplier, order_date, warehouse: basic_info_models.WareHouse, operator: basic_info_models.Employee, items: List[Dict[str, Any]], remarks: str | None = '', created_by=None, ) -> models.PurchaseOrder: """ 创建采购订单并触发异步创建入库单任务。 Args: merchant: 采购单所属商户 supplier: 供应商 order_date: 订单日期 (date) warehouse: 入库仓库实例 operator: 经办人 items: 产品明细，字段会根据仓库模式校验 remarks: 备注 created_by: 创建者用户（可选，用于 stock 记录中的 created_by） """ if not items: raise ValueError('items 不能为空') normalized_date = _normalize_order_date(order_date) purchase_items, stock_flow_items = _normalize_order_items( merchant=merchant, warehouse=warehouse, items=items, is_outgoing=False, ) with transaction.atomic(): purchase_order = models.PurchaseOrder.objects.create( merchant=merchant, supplier=supplier, purchase_date=normalized_date, operator=operator, warehouse=warehouse, remarks=remarks, ) bulk_objects = [ models.PurchaseOrderItem( purchase_order=purchase_order, product=item_data['product'], price=item_data['price'], color=item_data.get('color'), quantity=item_data['quantity'], unit=item_data['unit'], spec=item_data.get('spec'), empty_diff_percent=item_data['empty_diff_percent'], quantity_of_rolls=item_data.get('quantity_of_rolls'), num_of_rolls=item_data['num_of_rolls'], batch_number=item_data.get('batch_number'), remarks=item_data.get('remarks'), ) for item_data in purchase_items ] models.PurchaseOrderItem.objects.bulk_create(bulk_objects) purchase_order.refresh_from_db() return purchase_order def create_sales_order( *, merchant: basic_info_models.Merchant, customer: basic_info_models.Customer, order_date, warehouse: basic_info_models.WareHouse, operator: basic_info_models.Employee, items: List[Dict[str, Any]], remarks: str | None = '', created_by=None, ) -> models.SalesOrder: """ 创建销售订单，后续审批通过后会触发出库任务。 """ if not items: raise ValueError('items 不能为空') normalized_date = _normalize_order_date(order_date) sales_items, stock_flow_items = _normalize_order_items( merchant=merchant, warehouse=warehouse, items=items, is_outgoing=True, ) with transaction.atomic(): sales_order = models.SalesOrder.objects.create( merchant=merchant, customer=customer, sales_date=normalized_date, operator=operator, warehouse=warehouse, remarks=remarks, ) bulk_objects = [ models.SalesOrderItem( sales_order=sales_order, product=item_data['product'], price=item_data['price'], color=item_data.get('color'), quantity=item_data['quantity'], unit=item_data['unit'], spec=item_data.get('spec'), empty_diff_percent=item_data['empty_diff_percent'], quantity_of_rolls=item_data.get('quantity_of_rolls'), num_of_rolls=item_data['num_of_rolls'], consume_detail_ids=item_data.get('consume_detail_ids'), batch_number=item_data.get('batch_number'), remarks=item_data.get('remarks'), ) for item_data in sales_items ] models.SalesOrderItem.objects.bulk_create(bulk_objects) sales_order.refresh_from_db() return sales_order def create_purchase_return_order( *, merchant: basic_info_models.Merchant, supplier: basic_info_models.Supplier, return_date, warehouse: basic_info_models.WareHouse, operator: basic_info_models.Employee, items: List[Dict[str, Any]], remarks: str | None = '', purchase_order: models.PurchaseOrder | None = None, purchase_order_id: int | None = None, created_by=None, ) -> models.PurchaseReturnOrder: """ 创建采购退货单，出库并冲减供应商欠款。 """ if not items: raise ValueError('items 不能为空') normalized_date = _normalize_order_date(return_date) normalized_items, _ = _normalize_order_items( merchant=merchant, warehouse=warehouse, items=items, is_outgoing=True, ) resolved_purchase_order = purchase_order if resolved_purchase_order is None and purchase_order_id: resolved_purchase_order = models.PurchaseOrder.objects.filter( id=purchase_order_id, merchant=merchant, ).first() if resolved_purchase_order is None: raise ValueError('关联的采购单不存在或不属于当前商户') if resolved_purchase_order and resolved_purchase_order.supplier_id != supplier.id: raise ValueError('采购退货单的供应商必须与原采购单一致') with transaction.atomic(): return_order = models.PurchaseReturnOrder.objects.create( merchant=merchant, supplier=supplier, purchase_order=resolved_purchase_order, return_date=normalized_date, operator=operator, warehouse=warehouse, remarks=remarks, ) bulk_objects = [ models.PurchaseReturnOrderItem( purchase_return_order=return_order, product=item_data['product'], price=item_data['price'], color=item_data.get('color'), quantity=item_data['quantity'], unit=item_data['unit'], spec=item_data.get('spec'), empty_diff_percent=item_data['empty_diff_percent'], quantity_of_rolls=item_data.get('quantity_of_rolls'), num_of_rolls=item_data['num_of_rolls'], consume_detail_ids=item_data.get('consume_detail_ids'), batch_number=item_data.get('batch_number'), remarks=item_data.get('remarks'), ) for item_data in normalized_items ] models.PurchaseReturnOrderItem.objects.bulk_create(bulk_objects) return_order.refresh_from_db() return return_order def create_sales_return_order( *, merchant: basic_info_models.Merchant, customer: basic_info_models.Customer, return_date, warehouse: basic_info_models.WareHouse, operator: basic_info_models.Employee, items: List[Dict[str, Any]], remarks: str | None = '', sales_order: models.SalesOrder | None = None, sales_order_id: int | None = None, created_by=None, ) -> models.SalesReturnOrder: """ 创建销售退货单，入库并冲减客户欠款。 """ if not items: raise ValueError('items 不能为空') normalized_date = _normalize_order_date(return_date) normalized_items, _ = _normalize_order_items( merchant=merchant, warehouse=warehouse, items=items, is_outgoing=False, ) resolved_sales_order = sales_order if resolved_sales_order is None and sales_order_id: resolved_sales_order = models.SalesOrder.objects.filter( id=sales_order_id, merchant=merchant, ).first() if resolved_sales_order is None: raise ValueError('关联的销售单不存在或不属于当前商户') if resolved_sales_order and resolved_sales_order.customer_id != customer.id: raise ValueError('销售退货单的客户必须与原销售单一致') with transaction.atomic(): return_order = models.SalesReturnOrder.objects.create( merchant=merchant, customer=customer, sales_order=resolved_sales_order, return_date=normalized_date, operator=operator, warehouse=warehouse, remarks=remarks, ) bulk_objects = [ models.SalesReturnOrderItem( sales_return_order=return_order, product=item_data['product'], price=item_data['price'], color=item_data.get('color'), quantity=item_data['quantity'], unit=item_data['unit'], spec=item_data.get('spec'), empty_diff_percent=item_data['empty_diff_percent'], quantity_of_rolls=item_data.get('quantity_of_rolls'), num_of_rolls=item_data['num_of_rolls'], consume_detail_ids=item_data.get('consume_detail_ids'), batch_number=item_data.get('batch_number'), remarks=item_data.get('remarks'), ) for item_data in normalized_items ] models.SalesReturnOrderItem.objects.bulk_create(bulk_objects) return_order.refresh_from_db() return return_order def create_payment_order( *, merchant: basic_info_models.Merchant, supplier: basic_info_models.Supplier, payment_date, amount, operator: basic_info_models.Employee, remarks: str | None = '', ) -> models.PaymentOrder: """ 创建付款单（资金流出）。 """ normalized_date = _normalize_order_date(payment_date) normalized_amount = _ensure_positive_amount(amount, 'amount') with transaction.atomic(): payment_order = models.PaymentOrder.objects.create( merchant=merchant, supplier=supplier, payment_date=normalized_date, amount=normalized_amount, operator=operator, remarks=remarks, ) payment_order.refresh_from_db() return payment_order def create_receipt_order( *, merchant: basic_info_models.Merchant, customer: basic_info_models.Customer, receipt_date, amount, operator: basic_info_models.Employee, remarks: str | None = '', ) -> models.ReceiptOrder: """ 创建收款单（资金流入）。 """ normalized_date = _normalize_order_date(receipt_date) normalized_amount = _ensure_positive_amount(amount, 'amount') with transaction.atomic(): receipt_order = models.ReceiptOrder.objects.create( merchant=merchant, customer=customer, receipt_date=normalized_date, amount=normalized_amount, operator=operator, remarks=remarks, ) receipt_order.refresh_from_db() return receipt_order def review_purchase_order( *, purchase_order: models.PurchaseOrder | None = None, purchase_order_id: int | None = None, target_status: models.PurchaseOrderStatusEnum, reviewed_by=None, ) -> models.PurchaseOrder: """ 审批或作废采购单。 当目标状态为 APPROVED 且开启自动入库任务时，将触发入库 Celery 任务； 当目标状态为 CANCELLED 时，会在更新状态前确认未生成任何出入库单。 """ order = _resolve_purchase_order_instance(purchase_order, purchase_order_id) if target_status not in { models.PurchaseOrderStatusEnum.APPROVED, models.PurchaseOrderStatusEnum.CANCELLED, }: raise ValueError('target_status 只能是 APPROVED 或 CANCELLED') if order.status == target_status: return order if target_status == models.PurchaseOrderStatusEnum.APPROVED: if order.status == models.PurchaseOrderStatusEnum.CANCELLED: raise ValueError('作废状态的采购单无法再次审批') return _approve_purchase_order(order, reviewed_by) return _cancel_purchase_order(order) def review_sales_order( *, sales_order: models.SalesOrder | None = None, sales_order_id: int | None = None, target_status: models.SalesOrderStatusEnum, reviewed_by=None, ) -> models.SalesOrder: """ 审批或作废销售单。 """ order = _resolve_sales_order_instance(sales_order, sales_order_id) if target_status not in { models.SalesOrderStatusEnum.APPROVED, models.SalesOrderStatusEnum.CANCELLED, }: raise ValueError('target_status 只能是 APPROVED 或 CANCELLED') if order.status == target_status: return order if target_status == models.SalesOrderStatusEnum.APPROVED: if order.status == models.SalesOrderStatusEnum.CANCELLED: raise ValueError('作废状态的销售单无法再次审批') return _approve_sales_order(order, reviewed_by) return _cancel_sales_order(order) def review_purchase_return_order( *, purchase_return_order: models.PurchaseReturnOrder | None = None, purchase_return_order_id: int | None = None, target_status: models.PurchaseReturnStatusEnum, reviewed_by=None, ) -> models.PurchaseReturnOrder: order = _resolve_purchase_return_order_instance(purchase_return_order, purchase_return_order_id) if target_status not in { models.PurchaseReturnStatusEnum.APPROVED, models.PurchaseReturnStatusEnum.CANCELLED, }: raise ValueError('target_status 只能是 APPROVED 或 CANCELLED') if order.status == target_status: return order if target_status == models.PurchaseReturnStatusEnum.APPROVED: if order.status == models.PurchaseReturnStatusEnum.CANCELLED: raise ValueError('作废状态的采购退货单无法再次审批') return _approve_purchase_return_order(order, reviewed_by) return _cancel_purchase_return_order(order) def review_sales_return_order( *, sales_return_order: models.SalesReturnOrder | None = None, sales_return_order_id: int | None = None, target_status: models.SalesReturnStatusEnum, reviewed_by=None, ) -> models.SalesReturnOrder: order = _resolve_sales_return_order_instance(sales_return_order, sales_return_order_id) if target_status not in { models.SalesReturnStatusEnum.APPROVED, models.SalesReturnStatusEnum.CANCELLED, }: raise ValueError('target_status 只能是 APPROVED 或 CANCELLED') if order.status == target_status: return order if target_status == models.SalesReturnStatusEnum.APPROVED: if order.status == models.SalesReturnStatusEnum.CANCELLED: raise ValueError('作废状态的销售退货单无法再次审批') return _approve_sales_return_order(order, reviewed_by) return _cancel_sales_return_order(order) def review_payment_order( *, payment_order: models.PaymentOrder | None = None, payment_order_id: int | None = None, target_status: models.PaymentOrderStatusEnum, reviewed_by=None, ) -> models.PaymentOrder: order = _resolve_payment_order_instance(payment_order, payment_order_id) if target_status not in { models.PaymentOrderStatusEnum.APPROVED, models.PaymentOrderStatusEnum.CANCELLED, }: raise ValueError('target_status 只能是 APPROVED 或 CANCELLED') if order.status == target_status: return order 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, source_type=models.BalanceChangeSourceEnum.PAYMENT_ORDER, source_id=locked.id, ) locked.refresh_from_db(fields=['status', 'updated_at']) return locked 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( *, receipt_order: models.ReceiptOrder | None = None, receipt_order_id: int | None = None, target_status: models.ReceiptOrderStatusEnum, reviewed_by=None, ) -> models.ReceiptOrder: order = _resolve_receipt_order_instance(receipt_order, receipt_order_id) if target_status not in { models.ReceiptOrderStatusEnum.APPROVED, models.ReceiptOrderStatusEnum.CANCELLED, }: raise ValueError('target_status 只能是 APPROVED 或 CANCELLED') if order.status == target_status: return order 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, source_type=models.BalanceChangeSourceEnum.RECEIPT_ORDER, source_id=locked.id, ) locked.refresh_from_db(fields=['status', 'updated_at']) return locked 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( *, merchant: basic_info_models.Merchant, warehouse: basic_info_models.WareHouse, items: List[Dict[str, Any]], is_outgoing: bool, ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: """ 根据仓库模式校验订单明细，并返回： - normalized_items: 用于创建订单明细模型 - stock_flow_items: 传递给 StockFlowService 的 items 结构 is_outgoing=True 表示出库（销售等），需要额外校验 consume_detail_ids。 """ normalized_items: List[Dict[str, Any]] = [] stock_flow_items: List[Dict[str, Any]] = [] warehouse_mode = warehouse.mode for index, raw_item in enumerate(items): product_id = raw_item.get('product_id') if not product_id: raise ValueError(f'items[{index}].product_id 不能为空') try: product = basic_info_models.Product.objects.get(id=product_id, merchant=merchant) except basic_info_models.Product.DoesNotExist as exc: raise ValueError(f'产品 {product_id} 不存在或不属于当前商户') from exc price = _to_decimal(raw_item.get('price', '0'), f'items[{index}].price') empty_diff_percent = _to_decimal(raw_item.get('empty_diff_percent', '0'), f'items[{index}].empty_diff_percent') color = raw_item.get('color') batch_number = raw_item.get('batch_number') remarks = raw_item.get('remarks') spec = raw_item.get('spec') unit = raw_item.get('unit') or product.get_unit_display() or '米' quantity = 0 num_of_rolls = 0 quantity_of_rolls = None consume_detail_ids_str = None if warehouse_mode == basic_info_models.WareHouseModeEnum.UNRESTRICTED: if 'numbers' in raw_item and raw_item['numbers']: raise ValueError(f'仓库为宽进/宽出模式，items[{index}] 不应提供 numbers') quantity = _to_positive_int(raw_item.get('quantity'), f'items[{index}].quantity') num_of_rolls = _to_positive_int(raw_item.get('num_of_rolls'), f'items[{index}].num_of_rolls') stock_flow_items.append({ 'product_id': product.id, 'value': str(quantity), 'num_of_rolls': num_of_rolls, }) elif warehouse_mode == basic_info_models.WareHouseModeEnum.RESTRICT_IN: numbers = raw_item.get('numbers') if not numbers or not isinstance(numbers, list): raise ValueError(f'仓库为严进宽出模式，items[{index}] 需要提供 numbers 数组') normalized_numbers = [ str(_to_positive_int(value, f'items[{index}].numbers[{pos}]')) for pos, value in enumerate(numbers) ] num_of_rolls = len(normalized_numbers) quantity = sum(int(val) for val in normalized_numbers) quantity_of_rolls = ','.join(normalized_numbers) stock_flow_items.append({ 'product_id': product.id, 'quantities': normalized_numbers, }) else: # WareHouseModeEnum.RESTRICT_IN_OUT if is_outgoing: consume_ids = raw_item.get('consume_detail_ids') if not consume_ids or not isinstance(consume_ids, list): raise ValueError(f'仓库为严进严出模式，items[{index}] 需要提供 consume_detail_ids 数组') normalized_ids = [ _to_positive_int(value, f'items[{index}].consume_detail_ids[{pos}]') for pos, value in enumerate(consume_ids) ] consume_detail_ids_str = ','.join(str(value) for value in normalized_ids) quantity = _to_positive_int(raw_item.get('quantity'), f'items[{index}].quantity') num_of_rolls = len(normalized_ids) stock_flow_items.append({ 'product_id': product.id, 'consume_detail_ids': normalized_ids, }) else: numbers = raw_item.get('numbers') if not numbers or not isinstance(numbers, list): raise ValueError(f'仓库为严进严出模式，items[{index}] 需要提供 numbers 数组') normalized_numbers = [ str(_to_positive_int(value, f'items[{index}].numbers[{pos}]')) for pos, value in enumerate(numbers) ] num_of_rolls = len(normalized_numbers) quantity = sum(int(val) for val in normalized_numbers) quantity_of_rolls = ','.join(normalized_numbers) stock_flow_items.append({ 'product_id': product.id, 'quantities': normalized_numbers, }) normalized_items.append({ 'product': product, 'price': price, 'color': color, 'quantity': quantity, 'unit': unit, 'empty_diff_percent': empty_diff_percent, 'quantity_of_rolls': quantity_of_rolls, 'num_of_rolls': num_of_rolls, 'batch_number': batch_number, 'remarks': remarks, 'spec': spec, 'consume_detail_ids': consume_detail_ids_str, }) return normalized_items, stock_flow_items def _approve_purchase_order( purchase_order: models.PurchaseOrder, reviewed_by, ) -> models.PurchaseOrder: with transaction.atomic(): 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(), source_type=models.BalanceChangeSourceEnum.PURCHASE_ORDER, source_id=locked_order.id, ) 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( purchase_order_id=locked_order.id, warehouse_id=locked_order.warehouse_id, items=stock_flow_items, created_by_id=created_by_id, ) locked_order.refresh_from_db(fields=['status', 'updated_at']) return locked_order def _cancel_purchase_order(purchase_order: models.PurchaseOrder) -> models.PurchaseOrder: with transaction.atomic(): 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: with transaction.atomic(): 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(), source_type=models.BalanceChangeSourceEnum.SALES_ORDER, source_id=locked_order.id, ) created_by_id = getattr(reviewed_by, 'id', None) if _auto_stock_task_enabled(locked_order.merchant): logger.info('审批通过销售单 %s，触发出库任务', locked_order.id) create_sales_order_stock_entries.delay( sales_order_id=locked_order.id, warehouse_id=locked_order.warehouse_id, items=stock_flow_items, created_by_id=created_by_id, ) locked_order.refresh_from_db(fields=['status', 'updated_at']) return locked_order def _cancel_sales_order(sales_order: models.SalesOrder) -> models.SalesOrder: with transaction.atomic(): 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('销售单已生成出入库记录，无法作废') locked.status = models.SalesOrderStatusEnum.CANCELLED locked.save(update_fields=['status', 'updated_at']) locked.refresh_from_db(fields=['status', 'updated_at']) return locked def _approve_purchase_return_order( purchase_return_order: models.PurchaseReturnOrder, reviewed_by, ) -> models.PurchaseReturnOrder: with transaction.atomic(): locked_order = models.PurchaseReturnOrder.objects.select_related( 'merchant', 'warehouse', 'supplier' ).prefetch_related('items').select_for_update().get(id=purchase_return_order.id) if locked_order.status == models.PurchaseReturnStatusEnum.APPROVED: return locked_order if locked_order.status == models.PurchaseReturnStatusEnum.CANCELLED: raise ValueError('作废状态的采购退货单无法再次审批') stock_flow_items = _build_stock_flow_items_from_order(locked_order) locked_order.status = models.PurchaseReturnStatusEnum.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(), source_type=models.BalanceChangeSourceEnum.PURCHASE_RETURN_ORDER, source_id=locked_order.id, ) created_by_id = getattr(reviewed_by, 'id', None) if _auto_stock_task_enabled(locked_order.merchant): logger.info('审批通过采购退货单 %s，触发出库任务', locked_order.id) create_purchase_return_order_stock_entries.delay( purchase_return_order_id=locked_order.id, warehouse_id=locked_order.warehouse_id, items=stock_flow_items, created_by_id=created_by_id, ) locked_order.refresh_from_db(fields=['status', 'updated_at']) return locked_order def _cancel_purchase_return_order( purchase_return_order: models.PurchaseReturnOrder, ) -> models.PurchaseReturnOrder: with transaction.atomic(): locked = models.PurchaseReturnOrder.objects.select_related('merchant').select_for_update().get( id=purchase_return_order.id ) if locked.status == models.PurchaseReturnStatusEnum.APPROVED: raise ValueError('已审批的采购退货单无法作废') if _order_has_stock_records( merchant_id=locked.merchant_id, source_type=stock_models.StockChangeSourceEnum.PURCHASE_RETURN, source_id=locked.id, ): raise ValueError('采购退货单已生成出入库记录，无法作废') locked.status = models.PurchaseReturnStatusEnum.CANCELLED locked.save(update_fields=['status', 'updated_at']) locked.refresh_from_db(fields=['status', 'updated_at']) return locked def _approve_sales_return_order( sales_return_order: models.SalesReturnOrder, reviewed_by, ) -> models.SalesReturnOrder: with transaction.atomic(): locked_order = models.SalesReturnOrder.objects.select_related( 'merchant', 'warehouse', 'customer' ).prefetch_related('items').select_for_update().get(id=sales_return_order.id) if locked_order.status == models.SalesReturnStatusEnum.APPROVED: return locked_order if locked_order.status == models.SalesReturnStatusEnum.CANCELLED: raise ValueError('作废状态的销售退货单无法再次审批') stock_flow_items = _build_stock_flow_items_from_order(locked_order) locked_order.status = models.SalesReturnStatusEnum.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(), source_type=models.BalanceChangeSourceEnum.SALES_RETURN_ORDER, source_id=locked_order.id, ) created_by_id = getattr(reviewed_by, 'id', None) if _auto_stock_task_enabled(locked_order.merchant): logger.info('审批通过销售退货单 %s，触发入库任务', locked_order.id) create_sales_return_order_stock_entries.delay( sales_return_order_id=locked_order.id, warehouse_id=locked_order.warehouse_id, items=stock_flow_items, created_by_id=created_by_id, ) locked_order.refresh_from_db(fields=['status', 'updated_at']) return locked_order def _cancel_sales_return_order( sales_return_order: models.SalesReturnOrder, ) -> models.SalesReturnOrder: with transaction.atomic(): locked = models.SalesReturnOrder.objects.select_related('merchant').select_for_update().get( id=sales_return_order.id ) if locked.status == models.SalesReturnStatusEnum.APPROVED: raise ValueError('已审批的销售退货单无法作废') if _order_has_stock_records( merchant_id=locked.merchant_id, source_type=stock_models.StockChangeSourceEnum.SALES_RETURN, source_id=locked.id, ): raise ValueError('销售退货单已生成出入库记录，无法作废') locked.status = models.SalesReturnStatusEnum.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( *, purchase_order_id: int, warehouse_id: int, items: List[Dict[str, Any]], created_by_id: int | None = None, ) -> Dict[str, Any]: """ 根据采购单生成入库记录。 """ try: purchase_order = models.PurchaseOrder.objects.select_related('merchant').get(id=purchase_order_id) except models.PurchaseOrder.DoesNotExist: logger.error('PurchaseOrder %s 不存在，无法创建入库单', purchase_order_id) return {'error': 'purchase_order_not_found', 'purchase_order_id': purchase_order_id} merchant = purchase_order.merchant created_by = None if created_by_id: UserModel = get_user_model() created_by = UserModel.objects.filter(id=created_by_id).first() service = StockFlowService(merchant=merchant, created_by=created_by) record, details, created_count = service.stock_in( warehouse_id=warehouse_id, source_type=stock_models.StockChangeSourceEnum.PURCHASE, source_id=purchase_order.id, items=items, ) payload = { 'purchase_order_id': purchase_order.id, 'stock_change_record_id': getattr(record, 'id', None), 'created_details_count': created_count, } logger.info('采购单 %s 入库任务完成: %s', purchase_order.id, payload) return payload def create_sales_order_stock_entries_sync( *, sales_order_id: int, warehouse_id: int, items: List[Dict[str, Any]], created_by_id: int | None = None, ) -> Dict[str, Any]: """ 根据销售单生成出库记录。 """ try: sales_order = models.SalesOrder.objects.select_related('merchant').get(id=sales_order_id) except models.SalesOrder.DoesNotExist: logger.error('SalesOrder %s 不存在，无法创建出库单', sales_order_id) return {'error': 'sales_order_not_found', 'sales_order_id': sales_order_id} merchant = sales_order.merchant created_by = None if created_by_id: UserModel = get_user_model() created_by = UserModel.objects.filter(id=created_by_id).first() service = StockFlowService(merchant=merchant, created_by=created_by) record, details, created_count = service.stock_out( warehouse_id=warehouse_id, source_type=stock_models.StockChangeSourceEnum.SALES, source_id=sales_order.id, items=items, ) payload = { 'sales_order_id': sales_order.id, 'stock_change_record_id': getattr(record, 'id', None), 'created_details_count': created_count, } logger.info('销售单 %s 出库任务完成: %s', sales_order.id, payload) return payload def create_purchase_return_order_stock_entries_sync( *, purchase_return_order_id: int, warehouse_id: int, items: List[Dict[str, Any]], created_by_id: int | None = None, ) -> Dict[str, Any]: """ 根据采购退货单生成出库记录。 """ try: purchase_return_order = models.PurchaseReturnOrder.objects.select_related('merchant').get( id=purchase_return_order_id ) except models.PurchaseReturnOrder.DoesNotExist: logger.error('PurchaseReturnOrder %s 不存在，无法创建出库单', purchase_return_order_id) return {'error': 'purchase_return_order_not_found', 'purchase_return_order_id': purchase_return_order_id} merchant = purchase_return_order.merchant created_by = None if created_by_id: UserModel = get_user_model() created_by = UserModel.objects.filter(id=created_by_id).first() service = StockFlowService(merchant=merchant, created_by=created_by) record, details, created_count = service.stock_out( warehouse_id=warehouse_id, source_type=stock_models.StockChangeSourceEnum.PURCHASE_RETURN, source_id=purchase_return_order.id, items=items, ) payload = { 'purchase_return_order_id': purchase_return_order.id, 'stock_change_record_id': getattr(record, 'id', None), 'created_details_count': created_count, } logger.info('采购退货单 %s 出库任务完成: %s', purchase_return_order.id, payload) return payload def create_sales_return_order_stock_entries_sync( *, sales_return_order_id: int, warehouse_id: int, items: List[Dict[str, Any]], created_by_id: int | None = None, ) -> Dict[str, Any]: """ 根据销售退货单生成入库记录。 """ try: sales_return_order = models.SalesReturnOrder.objects.select_related('merchant').get( id=sales_return_order_id ) except models.SalesReturnOrder.DoesNotExist: logger.error('SalesReturnOrder %s 不存在，无法创建入库单', sales_return_order_id) return {'error': 'sales_return_order_not_found', 'sales_return_order_id': sales_return_order_id} merchant = sales_return_order.merchant created_by = None if created_by_id: UserModel = get_user_model() created_by = UserModel.objects.filter(id=created_by_id).first() service = StockFlowService(merchant=merchant, created_by=created_by) record, details, created_count = service.stock_in( warehouse_id=warehouse_id, source_type=stock_models.StockChangeSourceEnum.SALES_RETURN, source_id=sales_return_order.id, items=items, ) payload = { 'sales_return_order_id': sales_return_order.id, 'stock_change_record_id': getattr(record, 'id', None), 'created_details_count': created_count, } logger.info('销售退货单 %s 入库任务完成: %s', sales_return_order.id, payload) return payload def _build_stock_flow_items_from_order(order) -> List[Dict[str, Any]]: """ 根据订单明细还原 StockFlowService 所需的 items 结构。 """ warehouse_mode = order.warehouse.mode items_payload: List[Dict[str, Any]] = [] order_items = order.items.all() if not order_items: raise ValueError('订单没有任何明细，无法生成库存记录') if warehouse_mode == basic_info_models.WareHouseModeEnum.UNRESTRICTED: for item in order_items: items_payload.append({ 'product_id': item.product_id, 'value': str(item.quantity), 'num_of_rolls': item.num_of_rolls or 1, }) return items_payload # 严进 / 严进严出模式 for item in order_items: consume_ids_raw = getattr(item, 'consume_detail_ids', None) if consume_ids_raw: normalized_ids = [value.strip() for value in consume_ids_raw.split(',') if value.strip()] if not normalized_ids: raise ValueError('严进严出订单缺少 consume_detail_ids 数据，无法生成库存记录') items_payload.append({ 'product_id': item.product_id, 'consume_detail_ids': [int(value) for value in normalized_ids], }) continue raw_numbers = (item.quantity_of_rolls or '').split(',') normalized_numbers = [value.strip() for value in raw_numbers if value.strip()] if not normalized_numbers: raise ValueError('严进仓订单缺少 numbers 数据，无法生成库存记录') items_payload.append({ 'product_id': item.product_id, 'quantities': normalized_numbers, }) return items_payload def _auto_stock_task_enabled(merchant: basic_info_models.Merchant) -> bool: try: setting = MerchantSettingService.get_setting( merchant, basic_info_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS, ) except basic_info_models.MerchantSetting.DoesNotExist: logger.warning('商户 %s 未配置 auto_create_stock_change_tasks，默认关闭', merchant.id) return False return setting.value is True def _order_has_stock_records( *, merchant_id: int, source_type: stock_models.StockChangeSourceEnum, source_id: int, ) -> bool: return stock_models.StockChangeRecord.objects.filter( merchant_id=merchant_id, source_type=source_type, source_id=source_id, ).exists() def _resolve_purchase_order_instance( purchase_order: models.PurchaseOrder | None, purchase_order_id: int | None, ) -> models.PurchaseOrder: if purchase_order is None and purchase_order_id is None: raise ValueError('必须提供 purchase_order 或 purchase_order_id') if purchase_order is not None: purchase_order_id = purchase_order.id return models.PurchaseOrder.objects.select_related('merchant', 'warehouse').prefetch_related('items').get( id=purchase_order_id ) def _resolve_sales_order_instance( sales_order: models.SalesOrder | None, sales_order_id: int | None, ) -> models.SalesOrder: if sales_order is None and sales_order_id is None: raise ValueError('必须提供 sales_order 或 sales_order_id') if sales_order is not None: sales_order_id = sales_order.id return models.SalesOrder.objects.select_related('merchant', 'warehouse').prefetch_related('items').get( id=sales_order_id ) def _resolve_purchase_return_order_instance( purchase_return_order: models.PurchaseReturnOrder | None, purchase_return_order_id: int | None, ) -> models.PurchaseReturnOrder: if purchase_return_order is None and purchase_return_order_id is None: raise ValueError('必须提供 purchase_return_order 或 purchase_return_order_id') if purchase_return_order is not None: purchase_return_order_id = purchase_return_order.id return models.PurchaseReturnOrder.objects.select_related('merchant', 'warehouse').prefetch_related('items').get( id=purchase_return_order_id ) def _resolve_sales_return_order_instance( sales_return_order: models.SalesReturnOrder | None, sales_return_order_id: int | None, ) -> models.SalesReturnOrder: if sales_return_order is None and sales_return_order_id is None: raise ValueError('必须提供 sales_return_order 或 sales_return_order_id') if sales_return_order is not None: sales_return_order_id = sales_return_order.id return models.SalesReturnOrder.objects.select_related('merchant', 'warehouse').prefetch_related('items').get( id=sales_return_order_id ) def _resolve_payment_order_instance( payment_order: models.PaymentOrder | None, payment_order_id: int | None, ) -> models.PaymentOrder: if payment_order is None and payment_order_id is None: raise ValueError('必须提供 payment_order 或 payment_order_id') if payment_order is not None: payment_order_id = payment_order.id return models.PaymentOrder.objects.select_related('merchant', 'supplier').get(id=payment_order_id) def _resolve_receipt_order_instance( receipt_order: models.ReceiptOrder | None, receipt_order_id: int | None, ) -> models.ReceiptOrder: if receipt_order is None and receipt_order_id is None: raise ValueError('必须提供 receipt_order 或 receipt_order_id') if receipt_order is not None: receipt_order_id = receipt_order.id return models.ReceiptOrder.objects.select_related('merchant', 'customer').get(id=receipt_order_id) def _to_decimal(value, field_name: str) -> Decimal: try: return Decimal(str(value)) except (InvalidOperation, TypeError) as exc: raise ValueError(f'{field_name} 必须是合法数值') from exc def _ensure_positive_amount(value, field_name: str) -> Decimal: amount = _to_decimal(value, field_name) if amount <= 0: raise ValueError(f'{field_name} 必须大于 0') return amount def _to_positive_int(value, field_name: str) -> int: if value is None: raise ValueError(f'{field_name} 不能为空') decimal_value = _to_decimal(value, field_name) if decimal_value <= 0: raise ValueError(f'{field_name} 必须大于 0') if decimal_value != decimal_value.to_integral_value(): raise ValueError(f'{field_name} 必须为整数') return int(decimal_value)