from __future__ import annotations from dataclasses import dataclass from datetime import date from decimal import Decimal from typing import Protocol, runtime_checkable from django.db import models from flower.common import ModelBase class CostCategory(ModelBase): """支出类目""" merchant = models.ForeignKey( 'basic_info.Merchant', on_delete=models.PROTECT, related_name='cost_categories', verbose_name='所属商户', ) unique_key = models.CharField(max_length=100, verbose_name='唯一标识键') name = models.CharField(max_length=100, verbose_name='类目名称') parent = models.ForeignKey( 'self', on_delete=models.PROTECT, null=True, blank=True, related_name='children', verbose_name='父类目', ) description = models.TextField(null=True, blank=True, verbose_name='描述') class Meta: verbose_name = '支出类目' verbose_name_plural = '支出类目' unique_together = ('merchant', 'unique_key') ordering = ('merchant', 'name') def __str__(self): return f'{self.name} ({self.unique_key})' class CostEntry(ModelBase): """支出明细""" merchant = models.ForeignKey( 'basic_info.Merchant', on_delete=models.PROTECT, related_name='cost_entries', verbose_name='所属商户', ) category = models.ForeignKey( CostCategory, on_delete=models.PROTECT, related_name='entries', verbose_name='支出类目', ) amount = models.DecimalField(max_digits=15, decimal_places=2, verbose_name='金额') occurred_at = models.DateField(verbose_name='发生日期') operator = models.ForeignKey( 'basic_info.Employee', on_delete=models.PROTECT, null=True, blank=True, related_name='cost_entries', verbose_name='经办人', ) image1 = models.ImageField(null=True, blank=True, upload_to='cost/entries/', verbose_name='凭证图片') image2 = models.ImageField(null=True, blank=True, upload_to='cost/entries/', verbose_name='备用凭证图片') source_module = models.CharField( max_length=50, null=True, blank=True, verbose_name='来源模块', ) source_id = models.CharField( max_length=100, null=True, blank=True, verbose_name='来源记录ID', ) remarks = models.TextField(null=True, blank=True, verbose_name='备注') class Meta: verbose_name = '支出明细' verbose_name_plural = '支出明细' ordering = ('-occurred_at', '-created_at') def __str__(self): return f'{self.category.name} {self.amount} ({self.occurred_at})' # ==================== 六边形端口协议 ==================== @dataclass(slots=True) class CostEntryInput: """Provider 返回的结构化成本条目""" category_key: str category_name: str amount: Decimal occurred_at: date source_module: str source_id: str remarks: str = '' @runtime_checkable class CostProviderPort(Protocol): """ 成本数据提供者接口。 其它模块实现此接口即可被 cost.services.collect_from_provider() 统一采集。 Provider 可以通过 category_key 声明自己使用的类目: - 如果 cost 模块已有匹配的 category_key,直接使用 - 如果没有,cost 模块会调用 ensure_category() 自动创建 Provider 也可以反向依赖 cost 模块的基础能力: - from cost.services import ensure_category - 在实现 get_cost_entries 前主动调用 ensure_category 预建类目 使用示例:: from cost.models import CostProviderPort, CostEntryInput from cost.services import ensure_category class PrintingCostProvider: category_key = 'printing_consumables' def get_cost_entries(self, *, merchant, start_date, end_date): ensure_category( merchant=merchant, category_key=self.category_key, category_name='印刷耗材', ) return [ CostEntryInput( category_key=self.category_key, category_name='印刷耗材', amount=Decimal('100.00'), occurred_at=date.today(), source_module='printing', source_id='PJ-001', ), ] """ category_key: str def get_cost_entries( self, *, merchant, start_date: date, end_date: date ) -> list[CostEntryInput]: ...