"""Структура документа базы и сериализация в/из JSON. Вся база — один словарь, который целиком помещается в память. Датаклассы ниже описывают его форму; на диск и обратно всё ходит через to_dict/from_dict, чтобы Decimal и date не разъезжались с их JSON-представлением. """ from __future__ import annotations import uuid from dataclasses import dataclass, field from datetime import date, datetime from decimal import Decimal from typing import Any from . import money as m SCHEMA_VERSION = 1 # --- Типы движения товара ------------------------------------------------- # # Тип определяет и подстановку цены в форме, и то, как продажа попадает # в отчёты: приносит ли она деньги или это чистое выбытие себестоимости. KIND_RETAIL = "retail" KIND_FRIEND = "friend" KIND_SELF = "self" KIND_GIFT = "gift" KIND_WRITEOFF = "writeoff" SALE_KINDS = (KIND_RETAIL, KIND_FRIEND, KIND_SELF, KIND_GIFT, KIND_WRITEOFF) SALE_KIND_LABELS = { KIND_RETAIL: "Продажа", KIND_FRIEND: "Другу по себестоимости", KIND_SELF: "Съел сам", KIND_GIFT: "Подарил", KIND_WRITEOFF: "Списание", } # Приносят выручку. Остальные типы — выбытие без денег: себестоимость всё равно # придётся вернуть пекарне, поэтому она гасится из маржи. REVENUE_KINDS = frozenset({KIND_RETAIL, KIND_FRIEND}) CONSUMPTION_KINDS = frozenset({KIND_SELF, KIND_GIFT, KIND_WRITEOFF}) def new_id(prefix: str) -> str: return f"{prefix}_{uuid.uuid4().hex[:12]}" def parse_date(value: Any) -> date: if isinstance(value, datetime): return value.date() if isinstance(value, date): return value return date.fromisoformat(str(value)) def dump_date(value: date) -> str: return value.isoformat() # --- Номенклатура --------------------------------------------------------- @dataclass class PricePoint: """Снимок цен товара на момент изменения. Пишется при каждой правке цены, поэтому история цен видна прямо в карточке товара, не полагаясь на историю git. """ date: date cost_price: Decimal retail_price: Decimal note: str = "" @classmethod def from_dict(cls, d: dict) -> "PricePoint": return cls( date=parse_date(d["date"]), cost_price=m.money(d.get("cost_price")), retail_price=m.money(d.get("retail_price")), note=d.get("note", ""), ) def to_dict(self) -> dict: return { "date": dump_date(self.date), "cost_price": m.dumps(self.cost_price), "retail_price": m.dumps(self.retail_price), "note": self.note, } @dataclass class Product: id: str name: str unit: str = "шт" cost_price: Decimal = m.ZERO retail_price: Decimal = m.ZERO archived: bool = False note: str = "" price_history: list[PricePoint] = field(default_factory=list) @classmethod def from_dict(cls, d: dict) -> "Product": return cls( id=d["id"], name=d.get("name", ""), unit=d.get("unit", "шт"), cost_price=m.money(d.get("cost_price")), retail_price=m.money(d.get("retail_price")), archived=bool(d.get("archived", False)), note=d.get("note", ""), price_history=[PricePoint.from_dict(p) for p in d.get("price_history", [])], ) def to_dict(self) -> dict: return { "id": self.id, "name": self.name, "unit": self.unit, "cost_price": m.dumps(self.cost_price), "retail_price": m.dumps(self.retail_price), "archived": self.archived, "note": self.note, "price_history": [p.to_dict() for p in self.price_history], } @dataclass class Counterparty: id: str name: str kind: str = "friend" note: str = "" @classmethod def from_dict(cls, d: dict) -> "Counterparty": return cls( id=d["id"], name=d.get("name", ""), kind=d.get("kind", "friend"), note=d.get("note", ""), ) def to_dict(self) -> dict: return {"id": self.id, "name": self.name, "kind": self.kind, "note": self.note} # --- Закупки -------------------------------------------------------------- @dataclass class BatchLine: product_id: str qty: Decimal unit_cost: Decimal @property def total(self) -> Decimal: return m.money(self.qty * self.unit_cost) @classmethod def from_dict(cls, d: dict) -> "BatchLine": return cls( product_id=d["product_id"], qty=m.qty(d.get("qty")), unit_cost=m.money(d.get("unit_cost")), ) def to_dict(self) -> dict: return { "product_id": self.product_id, "qty": m.dumps(self.qty), "unit_cost": m.dumps(self.unit_cost), } @dataclass class Batch: """Партия булок, взятая в пекарне в долг. unit_cost в строках — слепок цены на момент закупки: если цена у поставщика потом изменится, долг по уже взятой партии не должен поехать. """ id: str number: int date: date due_date: date lines: list[BatchLine] = field(default_factory=list) closed: bool = False note: str = "" @property def cost_total(self) -> Decimal: """Сколько надо вернуть пекарне за эту партию.""" return m.money(sum((line.total for line in self.lines), m.ZERO)) @property def qty_total(self) -> Decimal: return m.qty(sum((line.qty for line in self.lines), m.ZERO)) @classmethod def from_dict(cls, d: dict) -> "Batch": return cls( id=d["id"], number=int(d.get("number", 0)), date=parse_date(d["date"]), due_date=parse_date(d["due_date"]), lines=[BatchLine.from_dict(x) for x in d.get("lines", [])], closed=bool(d.get("closed", False)), note=d.get("note", ""), ) def to_dict(self) -> dict: return { "id": self.id, "number": self.number, "date": dump_date(self.date), "due_date": dump_date(self.due_date), "lines": [x.to_dict() for x in self.lines], "closed": self.closed, "note": self.note, } # --- Продажи и выбытие ---------------------------------------------------- @dataclass class SaleLine: product_id: str qty: Decimal unit_price: Decimal @property def total(self) -> Decimal: return m.money(self.qty * self.unit_price) @classmethod def from_dict(cls, d: dict) -> "SaleLine": return cls( product_id=d["product_id"], qty=m.qty(d.get("qty")), unit_price=m.money(d.get("unit_price")), ) def to_dict(self) -> dict: return { "product_id": self.product_id, "qty": m.dumps(self.qty), "unit_price": m.dumps(self.unit_price), } @dataclass class Payment: """Приход денег по конкретной продаже (может быть частичным).""" date: date amount: Decimal note: str = "" @classmethod def from_dict(cls, d: dict) -> "Payment": return cls( date=parse_date(d["date"]), amount=m.money(d.get("amount")), note=d.get("note", ""), ) def to_dict(self) -> dict: return { "date": dump_date(self.date), "amount": m.dumps(self.amount), "note": self.note, } @dataclass class Sale: """Любое выбытие товара: продажа, отдача другу, съел сам, списание.""" id: str date: date kind: str = KIND_RETAIL counterparty_id: str | None = None lines: list[SaleLine] = field(default_factory=list) payments: list[Payment] = field(default_factory=list) note: str = "" @property def total(self) -> Decimal: return m.money(sum((line.total for line in self.lines), m.ZERO)) @property def paid(self) -> Decimal: return m.money(sum((p.amount for p in self.payments), m.ZERO)) @property def debt(self) -> Decimal: """Сколько ещё должны по этой продаже (не бывает отрицательным).""" return max(m.ZERO, m.money(self.total - self.paid)) @property def is_consumption(self) -> bool: return self.kind in CONSUMPTION_KINDS @classmethod def from_dict(cls, d: dict) -> "Sale": return cls( id=d["id"], date=parse_date(d["date"]), kind=d.get("kind", KIND_RETAIL), counterparty_id=d.get("counterparty_id") or None, lines=[SaleLine.from_dict(x) for x in d.get("lines", [])], payments=[Payment.from_dict(x) for x in d.get("payments", [])], note=d.get("note", ""), ) def to_dict(self) -> dict: return { "id": self.id, "date": dump_date(self.date), "kind": self.kind, "counterparty_id": self.counterparty_id, "lines": [x.to_dict() for x in self.lines], "payments": [x.to_dict() for x in self.payments], "note": self.note, } @dataclass class BakeryPayment: """Возврат денег пекарне. batch_id=None означает «просто отдал денег» — такой платёж гасит долг по старейшим непокрытым партиям (см. ledger). """ id: str date: date amount: Decimal batch_id: str | None = None note: str = "" @classmethod def from_dict(cls, d: dict) -> "BakeryPayment": return cls( id=d["id"], date=parse_date(d["date"]), amount=m.money(d.get("amount")), batch_id=d.get("batch_id") or None, note=d.get("note", ""), ) def to_dict(self) -> dict: return { "id": self.id, "date": dump_date(self.date), "amount": m.dumps(self.amount), "batch_id": self.batch_id, "note": self.note, } # --- Журнал --------------------------------------------------------------- @dataclass class Change: field: str old: str new: str @classmethod def from_dict(cls, d: dict) -> "Change": return cls(field=d.get("field", ""), old=d.get("old", ""), new=d.get("new", "")) def to_dict(self) -> dict: return {"field": self.field, "old": self.old, "new": self.new} @dataclass class JournalEntry: """Запись аудит-лога. Пишется только через journal.apply, только append.""" id: str ts: str host: str action: str entity: str entity_id: str label: str changes: list[Change] = field(default_factory=list) @classmethod def from_dict(cls, d: dict) -> "JournalEntry": return cls( id=d["id"], ts=d.get("ts", ""), host=d.get("host", ""), action=d.get("action", ""), entity=d.get("entity", ""), entity_id=d.get("entity_id", ""), label=d.get("label", ""), changes=[Change.from_dict(c) for c in d.get("changes", [])], ) def to_dict(self) -> dict: return { "id": self.id, "ts": self.ts, "host": self.host, "action": self.action, "entity": self.entity, "entity_id": self.entity_id, "label": self.label, "changes": [c.to_dict() for c in self.changes], } # --- Настройки ------------------------------------------------------------ @dataclass class GitSettings: """Токен лежит здесь, то есть внутри зашифрованной базы. В .git/config он не попадает — иначе лежал бы открытым рядом с шифром. """ remote_url: str = "" token: str = "" branch: str = "main" author_name: str = "food-market" author_email: str = "food-market@localhost" auto_push_minutes: int = 60 enabled: bool = True daily_squash: bool = True @classmethod def from_dict(cls, d: dict) -> "GitSettings": base = cls() return cls( remote_url=d.get("remote_url", base.remote_url), token=d.get("token", base.token), branch=d.get("branch", base.branch) or base.branch, author_name=d.get("author_name") or base.author_name, author_email=d.get("author_email") or base.author_email, auto_push_minutes=int(d.get("auto_push_minutes", base.auto_push_minutes)), enabled=bool(d.get("enabled", base.enabled)), daily_squash=bool(d.get("daily_squash", base.daily_squash)), ) def to_dict(self) -> dict: return { "remote_url": self.remote_url, "token": self.token, "branch": self.branch, "author_name": self.author_name, "author_email": self.author_email, "auto_push_minutes": self.auto_push_minutes, "enabled": self.enabled, "daily_squash": self.daily_squash, } @dataclass class Settings: currency: str = "₽" default_period_days: int = 14 git: GitSettings = field(default_factory=GitSettings) @classmethod def from_dict(cls, d: dict) -> "Settings": base = cls() return cls( currency=d.get("currency", base.currency), default_period_days=int( d.get("default_period_days", base.default_period_days) ), git=GitSettings.from_dict(d.get("git", {})), ) def to_dict(self) -> dict: return { "currency": self.currency, "default_period_days": self.default_period_days, "git": self.git.to_dict(), } # --- Документ целиком ----------------------------------------------------- @dataclass class Document: schema_version: int = SCHEMA_VERSION settings: Settings = field(default_factory=Settings) products: list[Product] = field(default_factory=list) counterparties: list[Counterparty] = field(default_factory=list) batches: list[Batch] = field(default_factory=list) sales: list[Sale] = field(default_factory=list) bakery_payments: list[BakeryPayment] = field(default_factory=list) journal: list[JournalEntry] = field(default_factory=list) @classmethod def from_dict(cls, d: dict) -> "Document": return cls( schema_version=int(d.get("schema_version", SCHEMA_VERSION)), settings=Settings.from_dict(d.get("settings", {})), products=[Product.from_dict(x) for x in d.get("products", [])], counterparties=[ Counterparty.from_dict(x) for x in d.get("counterparties", []) ], batches=[Batch.from_dict(x) for x in d.get("batches", [])], sales=[Sale.from_dict(x) for x in d.get("sales", [])], bakery_payments=[ BakeryPayment.from_dict(x) for x in d.get("bakery_payments", []) ], journal=[JournalEntry.from_dict(x) for x in d.get("journal", [])], ) def to_dict(self) -> dict: return { "schema_version": self.schema_version, "settings": self.settings.to_dict(), "products": [x.to_dict() for x in self.products], "counterparties": [x.to_dict() for x in self.counterparties], "batches": [x.to_dict() for x in self.batches], "sales": [x.to_dict() for x in self.sales], "bakery_payments": [x.to_dict() for x in self.bakery_payments], "journal": [x.to_dict() for x in self.journal], } # --- поиск по id --- def product(self, product_id: str) -> Product | None: return _find(self.products, product_id) def counterparty(self, cp_id: str) -> Counterparty | None: return _find(self.counterparties, cp_id) def batch(self, batch_id: str) -> Batch | None: return _find(self.batches, batch_id) def sale(self, sale_id: str) -> Sale | None: return _find(self.sales, sale_id) def product_name(self, product_id: str) -> str: p = self.product(product_id) return p.name if p else "(удалённый товар)" def counterparty_name(self, cp_id: str | None) -> str: if not cp_id: return "" c = self.counterparty(cp_id) return c.name if c else "(удалённый контрагент)" def next_batch_number(self) -> int: return max((b.number for b in self.batches), default=0) + 1 def _find(items: list, item_id: str): for item in items: if item.id == item_id: return item return None