forked from erp-dev/erp
106 lines
3.0 KiB
Python
106 lines
3.0 KiB
Python
"""
|
||
Shipment 模块业务逻辑层
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from typing import List
|
||
|
||
from django.db import transaction
|
||
from django.db.models import QuerySet
|
||
|
||
from shipment.models import SalesItem, Shipment
|
||
|
||
|
||
def get_sales_items_by_printing_order(
|
||
printing_order_id: int,
|
||
include_already_has_shipment: bool = False,
|
||
) -> QuerySet[SalesItem]:
|
||
"""
|
||
通过生产订单ID查询对应的销售品
|
||
|
||
Args:
|
||
printing_order_id: 生产订单ID
|
||
include_already_has_shipment: 是否包含已关联出货单的销售品，默认为 False
|
||
|
||
Returns:
|
||
SalesItem 查询集
|
||
"""
|
||
from printing.models import PrintingJob
|
||
|
||
# 1. 获取该生产订单下所有 PrintingJob 的 ID
|
||
job_ids = PrintingJob.objects.filter(
|
||
printing_order_id=printing_order_id
|
||
).values_list('id', flat=True)
|
||
|
||
# 2. 查询 SalesItem，过滤 printing_job_id 在这些 job_ids 中
|
||
queryset = SalesItem.objects.filter(printing_job_id__in=list(job_ids))
|
||
|
||
# 3. 根据参数决定是否过滤已出货的销售品
|
||
if not include_already_has_shipment:
|
||
queryset = queryset.filter(shipment__isnull=True)
|
||
|
||
return queryset.select_related('shipment').order_by('id')
|
||
|
||
|
||
@transaction.atomic
|
||
def create_shipment(
|
||
customer_id: int,
|
||
shipment_date,
|
||
sales_item_ids: List[int],
|
||
created_by,
|
||
remark: str = '',
|
||
) -> Shipment:
|
||
"""
|
||
创建出货单并关联销售品
|
||
|
||
Args:
|
||
customer_id: 客户ID
|
||
shipment_date: 出货日期
|
||
sales_item_ids: 要关联的销售品ID列表
|
||
created_by: 创建人
|
||
remark: 备注
|
||
|
||
Returns:
|
||
创建的 Shipment 实例
|
||
|
||
Raises:
|
||
ValueError: 如果销售品不存在或已被关联到其他出货单
|
||
"""
|
||
from basic_info.models import Customer
|
||
|
||
# 验证客户存在
|
||
try:
|
||
customer = Customer.objects.get(id=customer_id)
|
||
except Customer.DoesNotExist:
|
||
raise ValueError(f'客户 {customer_id} 不存在')
|
||
|
||
# 验证销售品
|
||
if sales_item_ids:
|
||
# 查询销售品
|
||
sales_items = SalesItem.objects.filter(id__in=sales_item_ids)
|
||
found_ids = set(sales_items.values_list('id', flat=True))
|
||
missing_ids = set(sales_item_ids) - found_ids
|
||
|
||
if missing_ids:
|
||
raise ValueError(f'以下销售品不存在: {list(missing_ids)}')
|
||
|
||
# 检查是否有已关联出货单的销售品
|
||
already_shipped = sales_items.filter(shipment__isnull=False)
|
||
if already_shipped.exists():
|
||
shipped_ids = list(already_shipped.values_list('id', flat=True))
|
||
raise ValueError(f'以下销售品已关联到其他出货单: {shipped_ids}')
|
||
|
||
# 创建出货单
|
||
shipment = Shipment.objects.create(
|
||
customer=customer,
|
||
shipment_date=shipment_date,
|
||
remark=remark,
|
||
created_by=created_by,
|
||
)
|
||
|
||
# 关联销售品
|
||
if sales_item_ids:
|
||
SalesItem.objects.filter(id__in=sales_item_ids).update(shipment=shipment)
|
||
|
||
return shipment
|