forked from erp-dev/erp
268 lines
7.9 KiB
Python
268 lines
7.9 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 = "",
|
||
area: 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,
|
||
area=(area or "").strip(),
|
||
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 = "",
|
||
area: 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,
|
||
area=(area or "").strip(),
|
||
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),
|
||
remark=item.get("remark") or "",
|
||
created_by=created_by,
|
||
)
|
||
)
|
||
|
||
ExternalFinishedProduct.objects.bulk_create(objs)
|
||
return shipment
|
||
|
||
|
||
@transaction.atomic
|
||
def create_sales_item(
|
||
printing_job_id: int,
|
||
name: str,
|
||
quantity: str,
|
||
unit: int,
|
||
created_by,
|
||
customer_id: int | None = None,
|
||
remark: str = "",
|
||
position: str = "",
|
||
) -> SalesItem:
|
||
"""
|
||
手动创建销售品
|
||
|
||
Args:
|
||
printing_job_id: 生产任务ID（必填）
|
||
name: 销售品名称
|
||
quantity: 数量（字符串，会被转换为Decimal）
|
||
unit: 单位（1=米, 2=件, 3=码, 4=个）
|
||
created_by: 创建人
|
||
customer_id: 客户ID（可选）
|
||
remark: 备注（可选）
|
||
position: 货位（可选）
|
||
|
||
Returns:
|
||
创建的 SalesItem 实例
|
||
|
||
Raises:
|
||
ValueError: 如果生产任务不存在或不属于当前商户
|
||
"""
|
||
from decimal import Decimal, InvalidOperation
|
||
from printing.models import PrintingJob
|
||
|
||
# 获取当前用户的商户
|
||
emp = getattr(created_by, "employee", None)
|
||
merchant = getattr(emp, "merchant", None) if emp else None
|
||
if not merchant:
|
||
raise ValueError("用户未关联商户，无法创建销售品")
|
||
|
||
# 验证生产任务存在且属于当前商户
|
||
try:
|
||
printing_job = PrintingJob.objects.select_related("printing_order").get(
|
||
id=printing_job_id,
|
||
merchant=merchant,
|
||
)
|
||
except PrintingJob.DoesNotExist:
|
||
raise ValueError(f"生产任务 {printing_job_id} 不存在或不属于当前商户")
|
||
|
||
# 转换数量为Decimal
|
||
try:
|
||
quantity_decimal = Decimal(str(quantity))
|
||
except (InvalidOperation, ValueError, TypeError) as e:
|
||
raise ValueError(f"数量 {quantity} 格式无效: {e}")
|
||
|
||
# 如果未提供客户ID，尝试从主订单获取
|
||
if (
|
||
customer_id is None
|
||
and printing_job.printing_order
|
||
and printing_job.printing_order.customer
|
||
):
|
||
customer_id = printing_job.printing_order.customer_id
|
||
|
||
# 创建销售品
|
||
sales_item = SalesItem.objects.create(
|
||
shipment=None, # 初始状态：待分配
|
||
merchant=merchant,
|
||
name=name,
|
||
quantity=quantity_decimal,
|
||
unit=unit,
|
||
created_by=created_by,
|
||
printing_job_id=printing_job.id,
|
||
customer_id=customer_id,
|
||
remark=remark,
|
||
position=position,
|
||
)
|
||
|
||
return sales_item
|