Files

169 lines
6.2 KiB
Python

"""库存模型:Warehouse / Stock / StockMovement。
设计:
- Warehouse 仓库(在 tenant 下唯一编码)
- Stock 按 (tenant, warehouse, product) 唯一定义当前库存数 / 锁定数
- StockMovement 是每次库存变化的流水(不可变),用于审计与回溯
锁库存语义:
- on_hand: 在手数量(可卖/可用)
- locked: 已锁定(销售订单已确认但未出库)
- available = on_hand - locked
"""
from django.db import models
from apps.core.base_models import TenantScopedModel
from decimal import Decimal
class Warehouse(TenantScopedModel):
"""仓库。"""
code = models.CharField(max_length=64)
name = models.CharField(max_length=128)
location = models.CharField(max_length=255, blank=True, default="")
is_active = models.BooleanField(default=True)
is_default = models.BooleanField(default=False, help_text="默认仓库")
class Meta:
db_table = "inventory_warehouse"
unique_together = [("tenant", "code")]
verbose_name_plural = "仓库"
ordering = ["code"]
def __str__(self):
return f"{self.code} {self.name}"
class Stock(TenantScopedModel):
"""库存账面(每个 (warehouse, product) 一行)。"""
warehouse = models.ForeignKey(
Warehouse,
on_delete=models.PROTECT,
related_name="stocks",
)
product = models.ForeignKey(
"catalog.Product",
on_delete=models.PROTECT,
related_name="stocks",
)
on_hand = models.DecimalField(max_digits=18, decimal_places=4, default=0)
locked = models.DecimalField(max_digits=18, decimal_places=4, default=0)
avg_cost = models.DecimalField(max_digits=18, decimal_places=4, default=0)
class Meta:
db_table = "inventory_stock"
unique_together = [("tenant", "warehouse", "product")]
verbose_name_plural = "库存账面"
indexes = [
models.Index(fields=["tenant", "warehouse", "product"]),
models.Index(fields=["tenant", "product"]),
]
ordering = ["warehouse", "product"]
@property
def available(self) -> Decimal:
return self.on_hand - self.locked
def __str__(self):
return f"{self.warehouse.code}/{self.product.code} on_hand={self.on_hand} locked={self.locked}"
class StockMovement(TenantScopedModel):
"""库存流水(不可变)。"""
MOVEMENT_INBOUND = "inbound" # 入库(采购入库、调拨入库、其他入库)
MOVEMENT_OUTBOUND = "outbound" # 出库(销售出库、调拨出库、其他出库)
MOVEMENT_LOCK = "lock" # 锁定
MOVEMENT_UNLOCK = "unlock" # 释放锁定
MOVEMENT_ADJUST = "adjust" # 盘点调整
MOVEMENT_CHOICES = [
(MOVEMENT_INBOUND, "入库"),
(MOVEMENT_OUTBOUND, "出库"),
(MOVEMENT_LOCK, "锁定"),
(MOVEMENT_UNLOCK, "释放"),
(MOVEMENT_ADJUST, "盘点"),
]
SOURCE_CHOICES = [
("purchase", "采购入库"),
("sale", "销售出库"),
("transfer", "调拨"),
("adjust", "盘点调整"),
("manual", "手工"),
]
warehouse = models.ForeignKey(
Warehouse, on_delete=models.PROTECT, related_name="movements"
)
product = models.ForeignKey(
"catalog.Product", on_delete=models.PROTECT, related_name="movements"
)
movement_type = models.CharField(max_length=16, choices=MOVEMENT_CHOICES)
source_type = models.CharField(max_length=16, choices=SOURCE_CHOICES, default="manual")
source_ref = models.CharField(max_length=64, blank=True, default="", help_text="来源单据号")
quantity = models.DecimalField(max_digits=18, decimal_places=4, help_text="本次变动数量(正数)")
unit_cost = models.DecimalField(max_digits=18, decimal_places=4, default=0, help_text="单位成本")
batch_detail = models.JSONField(
default=list,
blank=True,
help_text="批次商品出库时的 FEFO 分摊明细 [{batch_no, quantity, expiry_date}]",
)
# 流水是不可变的:不允许 delete / 修改后影响金额
class Meta:
db_table = "inventory_movement"
verbose_name_plural = "库存流水"
indexes = [
models.Index(fields=["tenant", "warehouse", "-created_at"]),
models.Index(fields=["tenant", "product", "-created_at"]),
models.Index(fields=["source_type", "source_ref"]),
]
ordering = ["-created_at"]
def __str__(self):
sign = "+" if self.movement_type in (self.MOVEMENT_INBOUND, self.MOVEMENT_UNLOCK) else "-"
return f"[{self.created_at}] {self.warehouse.code}/{self.product.code} {sign}{self.quantity}"
class StockBatch(TenantScopedModel):
"""批次库存账(批次管理商品的 (warehouse, product, batch_no) 维度账面)。
与总账 Stock 并行维护:sum(StockBatch.on_hand) == Stock.on_hand。
出库按 FEFO(近效期优先)自动分摊。
"""
warehouse = models.ForeignKey(
Warehouse, on_delete=models.PROTECT, related_name="batches"
)
product = models.ForeignKey(
"catalog.Product", on_delete=models.PROTECT, related_name="batches"
)
batch_no = models.CharField(max_length=64, help_text="批次号")
production_date = models.DateField(null=True, blank=True, help_text="生产日期")
expiry_date = models.DateField(null=True, blank=True, help_text="到期日期")
on_hand = models.DecimalField(max_digits=18, decimal_places=4, default=0)
locked = models.DecimalField(max_digits=18, decimal_places=4, default=0)
unit_cost = models.DecimalField(max_digits=18, decimal_places=4, default=0)
class Meta:
db_table = "inventory_stock_batch"
unique_together = [("tenant", "warehouse", "product", "batch_no")]
verbose_name_plural = "批次库存"
indexes = [
models.Index(fields=["tenant", "product", "expiry_date"]),
models.Index(fields=["tenant", "warehouse", "product"]),
]
ordering = ["expiry_date", "id"]
@property
def available(self) -> Decimal:
return self.on_hand - self.locked
def __str__(self):
return f"{self.warehouse.code}/{self.product.code}/{self.batch_no} on_hand={self.on_hand}"