forked from erp-dev/erp
feat: completed
This commit is contained in:
151
cost/models.py
Normal file
151
cost/models.py
Normal file
@@ -0,0 +1,151 @@
|
||||
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]:
|
||||
...
|
||||
Reference in New Issue
Block a user