forked from erp-dev/erp
179 lines
5.5 KiB
Python
179 lines
5.5 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 ExternalFinishedProduct, 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} 不存在')
|
||
|
||
# merchant 隔离：必须能解析出当前用户 merchant
|
||
emp = getattr(created_by, 'employee', None)
|
||
merchant = getattr(emp, 'merchant', None) if emp else None
|
||
if not merchant:
|
||
# superuser 也必须绑定 merchant（避免产生无法隔离的数据）
|
||
raise ValueError('用户未关联商户，无法创建出货单')
|
||
if customer.merchant_id != merchant.id:
|
||
raise ValueError('无权限为该客户创建出货单')
|
||
|
||
# 验证销售品
|
||
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(
|
||
merchant=merchant,
|
||
customer=customer,
|
||
shipment_date=shipment_date,
|
||
remark=remark,
|
||
created_by=created_by,
|
||
)
|
||
|
||
# 关联销售品
|
||
if sales_item_ids:
|
||
updated = SalesItem.objects.filter(id__in=sales_item_ids, merchant=merchant).update(shipment=shipment)
|
||
if updated != len(sales_item_ids):
|
||
raise ValueError('存在不属于当前商户的销售品，无法关联到出货单')
|
||
|
||
return shipment
|
||
|
||
|
||
@transaction.atomic
|
||
def create_external_shipment(
|
||
customer_id: int,
|
||
shipment_date,
|
||
external_id: str,
|
||
external_finished_products: List[dict],
|
||
created_by,
|
||
remark: str = '',
|
||
) -> Shipment:
|
||
"""
|
||
创建出货单（external 版），并批量写入外部成品表并关联到出货单。
|
||
|
||
特点：
|
||
- 不绑定任何 SalesItem
|
||
- external_id 必填
|
||
- external_finished_products 必填（至少 1 条）
|
||
"""
|
||
from basic_info.models import Customer
|
||
|
||
# 验证客户存在
|
||
try:
|
||
customer = Customer.objects.get(id=customer_id)
|
||
except Customer.DoesNotExist:
|
||
raise ValueError(f'客户 {customer_id} 不存在')
|
||
|
||
# merchant 隔离
|
||
emp = getattr(created_by, 'employee', None)
|
||
merchant = getattr(emp, 'merchant', None) if emp else None
|
||
if not merchant:
|
||
raise ValueError('用户未关联商户，无法创建出货单')
|
||
if customer.merchant_id != merchant.id:
|
||
raise ValueError('无权限为该客户创建出货单')
|
||
|
||
external_id = (external_id or '').strip()
|
||
if not external_id:
|
||
raise ValueError('external_id 不能为空')
|
||
if not external_finished_products:
|
||
raise ValueError('external_finished_products 不能为空')
|
||
|
||
shipment = Shipment.objects.create(
|
||
merchant=merchant,
|
||
customer=customer,
|
||
shipment_date=shipment_date,
|
||
remark=remark,
|
||
created_by=created_by,
|
||
external_id=external_id,
|
||
)
|
||
|
||
objs = []
|
||
for item in external_finished_products:
|
||
objs.append(ExternalFinishedProduct(
|
||
shipment=shipment,
|
||
style_name=item.get('style_name', ''),
|
||
num_of_rolls=item.get('num_of_rolls', 0),
|
||
created_by=created_by,
|
||
))
|
||
|
||
ExternalFinishedProduct.objects.bulk_create(objs)
|
||
return shipment
|