from decimal import Decimal from django.db import models from django.db.models import Count, Q from django.utils import timezone from container.models import ContainerListModel class CountReason(models.Model): """Variance reason master data.""" code = models.CharField(max_length=32, unique=True, verbose_name="原因编码") description = models.CharField(max_length=255, verbose_name="原因说明") is_active = models.BooleanField(default=True, verbose_name="是否启用") sort = models.IntegerField(default=0, verbose_name="排序") note = models.CharField(max_length=255, blank=True, verbose_name="备注") create_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间") update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间") class Meta: db_table = "cycle_count_reason" verbose_name = "盘点差异原因" verbose_name_plural = "盘点差异原因" ordering = ["sort", "code"] def __str__(self) -> str: return f"{self.code}-{self.description}" class CountTask(models.Model): """Cycle count task header.""" TASK_FULL = "full" TASK_SAMPLE = "sample" TASK_ADHOC = "adhoc" TASK_TYPE_CHOICES = ( (TASK_FULL, "全盘"), (TASK_SAMPLE, "抽盘"), (TASK_ADHOC, "临盘"), ) STATUS_DRAFT = "draft" STATUS_RELEASED = "released" STATUS_IN_PROGRESS = "in_progress" STATUS_WAIT_RECOUNT = "wait_recount" STATUS_COMPLETED = "completed" STATUS_CLOSED = "closed" STATUS_CHOICES = ( (STATUS_DRAFT, "草稿"), (STATUS_RELEASED, "已下发"), (STATUS_IN_PROGRESS, "进行中"), (STATUS_WAIT_RECOUNT, "待复盘"), (STATUS_COMPLETED, "已完成"), (STATUS_CLOSED, "已关闭"), ) doc_no = models.CharField( max_length=40, unique=True, blank=True, verbose_name="任务单号", help_text="自动生成，也可手工录入", ) task_type = models.CharField( max_length=16, choices=TASK_TYPE_CHOICES, default=TASK_FULL, verbose_name="任务类型" ) status = models.CharField( max_length=20, choices=STATUS_CHOICES, default=STATUS_DRAFT, verbose_name="状态" ) source_batch = models.CharField(max_length=64, blank=True, verbose_name="来源批次") remark = models.CharField(max_length=255, blank=True, verbose_name="备注") scheduled_at = models.DateTimeField(null=True, blank=True, verbose_name="计划盘点时间") created_by = models.CharField(max_length=64, blank=True, verbose_name="创建人") total_details = models.IntegerField(default=0, verbose_name="明细数量") counted_details = models.IntegerField(default=0, verbose_name="已盘数量") variance_details = models.IntegerField(default=0, verbose_name="差异数量") create_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间") update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间") class Meta: db_table = "cycle_count_task" verbose_name = "盘点任务" verbose_name_plural = "盘点任务" ordering = ["-id"] def __str__(self) -> str: return self.doc_no @staticmethod def generate_doc_no() -> str: """Generate a readable unique code.""" return timezone.now().strftime("CC%Y%m%d%H%M%S%f") def save(self, *args, **kwargs): if not self.doc_no: self.doc_no = self.generate_doc_no() super().save(*args, **kwargs) def refresh_statistics(self): """Refresh counters & status based on detail lines.""" aggregates = self.details.aggregate( total=Count("id"), counted=Count( "id", filter=Q(initial_count_qty__isnull=False) | Q(recount_qty__isnull=False), ), variance=Count( "id", filter=Q(variance_qty__isnull=False) & ~Q(variance_qty=Decimal("0")), ), ) total = aggregates.get("total") or 0 counted = aggregates.get("counted") or 0 variance = aggregates.get("variance") or 0 current_status = self.status if total == 0: new_status = self.STATUS_DRAFT elif counted == 0: # 仍处于草稿态的任务不因生成/删除明细而自动下发 if current_status == self.STATUS_DRAFT: new_status = self.STATUS_DRAFT else: new_status = self.STATUS_RELEASED elif counted < total: new_status = self.STATUS_IN_PROGRESS elif variance > 0: new_status = self.STATUS_WAIT_RECOUNT else: new_status = self.STATUS_COMPLETED CountTask.objects.filter(pk=self.pk).update( total_details=total, counted_details=counted, variance_details=variance, status=new_status, update_time=timezone.now(), ) self.refresh_from_db(fields=["total_details", "counted_details", "variance_details", "status"]) class CountTaskDetail(models.Model): """Cycle count detail lines.""" STATUS_PENDING = "pending" STATUS_COUNTED = "counted" STATUS_VARIANCE = "variance" STATUS_COMPLETED = "completed" STATUS_CHOICES = ( (STATUS_PENDING, "待盘点"), (STATUS_COUNTED, "已初盘"), (STATUS_VARIANCE, "待复盘"), (STATUS_COMPLETED, "已完成"), ) task = models.ForeignKey( CountTask, related_name="details", on_delete=models.CASCADE, verbose_name="任务" ) container = models.ForeignKey( ContainerListModel, related_name="count_details", null=True, blank=True, on_delete=models.SET_NULL, verbose_name="托盘", ) container_code = models.CharField(max_length=64, blank=True, verbose_name="托盘编码") location_code = models.CharField(max_length=64, blank=True, verbose_name="库位") goods_code = models.CharField(max_length=64, blank=True, verbose_name="物料编码") goods_desc = models.CharField(max_length=255, blank=True, verbose_name="物料名称") goods_std = models.CharField(max_length=255, blank=True, verbose_name="规格型号") goods_unit = models.CharField(max_length=32, blank=True, verbose_name="单位") batch_number = models.CharField(max_length=64, blank=True, verbose_name="管理批次") book_qty = models.DecimalField( max_digits=14, decimal_places=3, default=Decimal("0"), verbose_name="账面数量" ) initial_count_qty = models.DecimalField( max_digits=14, decimal_places=3, null=True, blank=True, verbose_name="初盘数量" ) recount_qty = models.DecimalField( max_digits=14, decimal_places=3, null=True, blank=True, verbose_name="复盘数量" ) final_qty = models.DecimalField( max_digits=14, decimal_places=3, default=Decimal("0"), verbose_name="最终数量" ) variance_qty = models.DecimalField( max_digits=14, decimal_places=3, default=Decimal("0"), verbose_name="盘差数量" ) reason = models.ForeignKey( CountReason, related_name="details", on_delete=models.SET_NULL, null=True, blank=True, verbose_name="差异原因", ) counter_name = models.CharField(max_length=64, blank=True, verbose_name="初盘人") recount_counter = models.CharField(max_length=64, blank=True, verbose_name="复盘人") note = models.CharField(max_length=255, blank=True, verbose_name="备注") status = models.CharField( max_length=20, choices=STATUS_CHOICES, default=STATUS_PENDING, verbose_name="状态" ) create_time = models.DateTimeField(auto_now_add=True, verbose_name="创建时间") update_time = models.DateTimeField(auto_now=True, verbose_name="更新时间") class Meta: db_table = "cycle_count_task_detail" verbose_name = "盘点任务明细" verbose_name_plural = "盘点任务明细" ordering = ["id"] def __str__(self) -> str: return f"{self.task.doc_no}-{self.container_code or self.goods_code}" def sync_qty_fields(self): """Recalculate variance & final qty and derive status.""" book_qty = self.book_qty or Decimal("0") final_qty = None if self.recount_qty is not None: final_qty = self.recount_qty elif self.initial_count_qty is not None: final_qty = self.initial_count_qty else: final_qty = book_qty self.final_qty = final_qty self.variance_qty = (final_qty or Decimal("0")) - book_qty if self.recount_qty is not None: self.status = self.STATUS_COMPLETED elif self.initial_count_qty is not None: if self.variance_qty == Decimal("0"): self.status = self.STATUS_COUNTED else: self.status = self.STATUS_VARIANCE else: self.status = self.STATUS_PENDING def save(self, *args, **kwargs): self.sync_qty_fields() super().save(*args, **kwargs) if self.task_id: self.task.refresh_statistics() def delete(self, *args, **kwargs): task = self.task super().delete(*args, **kwargs) if task: task.refresh_statistics()