"""Структура документа базы и сериализация в/из 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 = 6 ONE = Decimal(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 PackPrice: """Цены одной фасовки в момент снимка истории. Хранится копией, а не ссылкой на фасовку: фасовку могут переименовать или удалить, а история обязана остаться читаемой. """ name: str cost_price: Decimal = m.ZERO retail_price: Decimal = m.ZERO @classmethod def from_dict(cls, d: dict) -> "PackPrice": return cls( name=d.get("name", ""), cost_price=m.money(d.get("cost_price")), retail_price=m.money(d.get("retail_price")), ) def to_dict(self) -> dict: return { "name": self.name, "cost_price": m.dumps(self.cost_price), "retail_price": m.dumps(self.retail_price), } @dataclass class PricePoint: """Снимок цен товара на момент изменения. Пишется при каждой правке цены, поэтому история цен видна прямо в карточке товара, не полагаясь на историю git. Цены фасовок входят в тот же снимок: иначе у товара история была бы, а у коробки — нет, хотя дорожают они обычно вместе. """ date: date cost_price: Decimal retail_price: Decimal note: str = "" packs: list[PackPrice] = field(default_factory=list) def pack(self, name: str) -> "PackPrice | None": return next((p for p in self.packs if p.name == name), None) @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", ""), packs=[PackPrice.from_dict(p) for p in d.get("packs", [])], ) 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, "packs": [p.to_dict() for p in self.packs], } @dataclass class Pack: """Фасовка: сколько базовых единиц внутри, почём берётся и почём отдаётся. Пачка печенья по 10 штук и та же печенька поштучно — это один товар с двумя фасовками, а не два разных товара. Остатки и себестоимость всегда считаются в базовых единицах, поэтому поштучные продажи корректно вычитаются из купленных пачек. **Обе цены — за всю фасовку целиком**, и обе свои, а не производные. Коробку и берут, и отдают по другой цене, чем поштучный товар: ради этого её и берут. Ноль означает «своей цены нет, считай от цены базовой единицы» — так ведут себя все фасовки, заведённые до появления этого поля. """ name: str size: Decimal retail_price: Decimal = m.ZERO cost_price: Decimal = m.ZERO @classmethod def from_dict(cls, d: dict) -> "Pack": return cls( name=d.get("name", ""), size=m.qty(d.get("size", 1)), retail_price=m.money(d.get("retail_price")), cost_price=m.money(d.get("cost_price")), ) def to_dict(self) -> dict: return { "name": self.name, "size": m.dumps(self.size), "retail_price": m.dumps(self.retail_price), "cost_price": m.dumps(self.cost_price), } @dataclass class Product: """Товар. Цены хранятся за базовую единицу, фасовки — надстройка над ней.""" id: str name: str unit: str = "шт" cost_price: Decimal = m.ZERO retail_price: Decimal = m.ZERO archived: bool = False note: str = "" packs: list[Pack] = field(default_factory=list) price_history: list[PricePoint] = field(default_factory=list) # --- фасовки --- def uom_names(self) -> list[str]: """Базовая единица плюс все фасовки, в порядке для выпадающего списка.""" return [self.unit] + [p.name for p in self.packs] def pack(self, uom: str) -> Pack | None: return next((p for p in self.packs if p.name == uom), None) def size_of(self, uom: str) -> Decimal: """Сколько базовых единиц в одной штуке выбранной фасовки.""" pack = self.pack(uom) return pack.size if pack else ONE def cost_for(self, uom: str) -> Decimal: """Себестоимость одной штуки выбранной фасовки. У фасовки закупочная цена своя: коробку почти всегда отдают дешевле, чем те же штуки поодиночке, — ради этого её и берут. Выводить её умножением поштучной цены на размер нельзя: получилось бы, что опт стоит столько же, сколько розница. """ return self._pack_price(uom, "cost_price", self.cost_price) def retail_for(self, uom: str) -> Decimal: """Цена продажи одной штуки выбранной фасовки.""" return self._pack_price(uom, "retail_price", self.retail_price) def _pack_price(self, uom: str, attr: str, base: Decimal) -> Decimal: """Цена за фасовку целиком: своя, а если не задана — от базовой.""" pack = self.pack(uom) if pack is None: return base own = getattr(pack, attr) if own > 0: return own return m.money(base * pack.size) @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", ""), packs=[Pack.from_dict(p) for p in d.get("packs", [])], 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, "packs": [p.to_dict() for p in self.packs], "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 DocumentLine: """Общее для строк закупки и продажи. Количество и цена хранятся в той фасовке, которую выбрал пользователь — «2 пачки по 200 ₽» так и остаётся двумя пачками по 200. Размер фасовки сохраняется слепком: если пачку потом переопределят с 10 штук на 12, уже записанные документы не должны поехать. """ product_id: str qty: Decimal uom: str = "" uom_size: Decimal = ONE def unit_name(self, product: "Product | None") -> str: return self.uom or (product.unit if product else "") @property def base_qty(self) -> Decimal: """Количество в базовых единицах — в них считаются остатки и FIFO.""" return m.qty(self.qty * self.uom_size) def _base_price(self, price: Decimal) -> Decimal: """Цена за базовую единицу. Без округления: делим 200 на 3 без потерь.""" if self.uom_size <= 0: return m.ZERO return price / self.uom_size @staticmethod def _uom_from_dict(d: dict) -> tuple[str, Decimal]: size = m.qty(d.get("uom_size", 1)) return d.get("uom", ""), size if size > 0 else ONE @dataclass class BatchLine(DocumentLine): unit_cost: Decimal = m.ZERO @property def total(self) -> Decimal: return m.money(self.qty * self.unit_cost) @property def base_unit_cost(self) -> Decimal: return self._base_price(self.unit_cost) @classmethod def from_dict(cls, d: dict) -> "BatchLine": uom, size = cls._uom_from_dict(d) return cls( product_id=d["product_id"], qty=m.qty(d.get("qty")), uom=uom, uom_size=size, unit_cost=m.money(d.get("unit_cost")), ) def to_dict(self) -> dict: return { "product_id": self.product_id, "qty": m.dumps(self.qty), "uom": self.uom, "uom_size": m.dumps(self.uom_size), "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: """Всего базовых единиц в партии: 2 пачки по 10 — это 20 штук.""" return m.qty(sum((line.base_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(DocumentLine): unit_price: Decimal = m.ZERO @property def total(self) -> Decimal: return m.money(self.qty * self.unit_price) @property def base_unit_price(self) -> Decimal: return self._base_price(self.unit_price) @classmethod def from_dict(cls, d: dict) -> "SaleLine": uom, size = cls._uom_from_dict(d) return cls( product_id=d["product_id"], qty=m.qty(d.get("qty")), uom=uom, uom_size=size, unit_price=m.money(d.get("unit_price")), ) def to_dict(self) -> dict: return { "product_id": self.product_id, "qty": m.dumps(self.qty), "uom": self.uom, "uom_size": m.dumps(self.uom_size), "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) # Деньги сверх стоимости товара. Живут отдельно от payments намеренно: # иначе чаевые раздували бы «оплачено» и продажа выглядела бы закрытой, # когда за булки ещё должны. tip: Decimal = m.ZERO 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 received(self) -> Decimal: """Сколько денег реально пришло по этой продаже, вместе с чаевыми.""" return m.money(self.paid + self.tip) @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", [])], tip=m.money(d.get("tip")), 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], "tip": m.dumps(self.tip), "note": self.note, } BAKERY_CASH = "cash" BAKERY_DISCOUNT = "discount" BAKERY_KIND_LABELS = { BAKERY_CASH: "Платёж", BAKERY_DISCOUNT: "Скидка пекарни", } @dataclass class BakeryPayment: """Расчёт с пекарней. Два вида, и различать их обязательно: * `cash` — деньги реально ушли из кармана; * `discount` — пекарня простила остаток. Долг закрыт, но денег никто не отдавал, и для тебя это доход. Свалить их в одну кучу нельзя: тогда «отдано пекарне» в статистике показывало бы суммы, которых ты не платил. batch_id=None означает «просто отдал денег» — такой платёж гасит долг по старейшим непокрытым партиям (см. ledger). """ id: str date: date amount: Decimal batch_id: str | None = None kind: str = BAKERY_CASH note: str = "" @property def is_discount(self) -> bool: return self.kind == BAKERY_DISCOUNT @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, kind=d.get("kind", BAKERY_CASH), 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, "kind": self.kind, "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 = "https://gt.ser.gay/kizya/food-records.git" 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