"""Единственный путь записи в базу. Ни один экран не меняет документ напрямую — всё идёт через функции этого модуля. Каждая из них снимает состояние сущности до и после, считает разницу и дописывает запись в аудит-лог. Так сделано потому, что git хранит зашифрованные снимки: `git diff` по ним бесполезен, а после схлопывания коммитов по дням истории внутри суток там нет вовсе. Журнал — единственное место, где видно «когда я поднял цену на булку с маком». Если бы запись шла в обход, новый экран однажды забыл бы залогировать изменение, и дыру заметили бы только когда история понадобится. Побочный, но важный эффект: правки, ничего не меняющие по существу, журнал не засоряют — при пустой разнице запись не создаётся и база не помечается изменённой, то есть лишнего коммита в git тоже не будет. """ from __future__ import annotations import socket from datetime import date, datetime from decimal import Decimal from . import money as m from .models import ( ONE, SALE_KIND_LABELS, BakeryPayment, Batch, BatchLine, Change, Counterparty, JournalEntry, Pack, Payment, PricePoint, Product, Sale, SaleLine, dump_date, new_id, parse_date, ) # Значения этих настроек в журнал не попадают — только факт изменения. SECRET_SETTINGS = frozenset({"token"}) class ValidationError(Exception): """Действие невозможно: UI показывает текст пользователю как есть.""" def _now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") def _host() -> str: try: return socket.gethostname() except OSError: return "?" def _record( vault, action: str, entity: str, entity_id: str, label: str, changes: list[Change], ) -> JournalEntry: entry = JournalEntry( id=new_id("j"), ts=_now_iso(), host=_host(), action=action, entity=entity, entity_id=entity_id, label=label, changes=changes, ) vault.doc.journal.append(entry) vault.touch() return entry def _diff(before: dict[str, str], after: dict[str, str]) -> list[Change]: keys = list(before) + [k for k in after if k not in before] return [ Change(field=k, old=before.get(k, ""), new=after.get(k, "")) for k in keys if before.get(k, "") != after.get(k, "") ] def _as_created(flat: dict[str, str]) -> list[Change]: return [Change(field=k, old="", new=v) for k, v in flat.items() if v] def _as_deleted(flat: dict[str, str]) -> list[Change]: return [Change(field=k, old=v, new="") for k, v in flat.items() if v] # --- Плоские снимки сущностей для журнала --------------------------------- # # Значения сразу человекочитаемые: журнал читает человек, а не программа, # и «30,00 ₽» полезнее, чем «30». def _flat_product(doc, p: Product) -> dict[str, str]: return { "название": p.name, "единица": p.unit, "себестоимость": m.fmt_money(p.cost_price, doc.settings.currency), "цена продажи": m.fmt_money(p.retail_price, doc.settings.currency), "фасовки": _packs_text(doc, p), "в архиве": "да" if p.archived else "", "заметка": p.note, } def _flat_counterparty(doc, c: Counterparty) -> dict[str, str]: return {"имя": c.name, "тип": c.kind, "заметка": c.note} def _lines_text(doc, lines, price_attr: str) -> str: cur = doc.settings.currency parts = [] for x in lines: unit = x.unit_name(doc.product(x.product_id)) parts.append( f"{doc.product_name(x.product_id)} × {m.fmt_qty(x.qty)} {unit} " f"по {m.fmt_money(getattr(x, price_attr), cur)}" ) return "; ".join(parts) def _packs_text(doc, product: Product) -> str: cur = doc.settings.currency return "; ".join( f"{p.name} = {m.fmt_qty(p.size)} {product.unit} по {m.fmt_money(p.retail_price, cur)}" for p in product.packs ) def _flat_batch(doc, b: Batch) -> dict[str, str]: return { "дата": dump_date(b.date), "покрыть до": dump_date(b.due_date), "состав": _lines_text(doc, b.lines, "unit_cost"), "сумма": m.fmt_money(b.cost_total, doc.settings.currency), "закрыта": "да" if b.closed else "", "заметка": b.note, } def _flat_sale(doc, s: Sale) -> dict[str, str]: cur = doc.settings.currency return { "дата": dump_date(s.date), "тип": SALE_KIND_LABELS.get(s.kind, s.kind), "кому": doc.counterparty_name(s.counterparty_id), "состав": _lines_text(doc, s.lines, "unit_price"), "сумма": m.fmt_money(s.total, cur), "оплачено": m.fmt_money(s.paid, cur), "чаевые": m.fmt_money(s.tip, cur) if s.tip else "", "заметка": s.note, } def _batch_label(b: Batch) -> str: return f"Партия №{b.number} от {b.date.strftime('%d.%m.%Y')}" def _sale_label(doc, s: Sale) -> str: who = doc.counterparty_name(s.counterparty_id) kind = SALE_KIND_LABELS.get(s.kind, s.kind) base = f"{kind} от {s.date.strftime('%d.%m.%Y')}" return f"{base} — {who}" if who else base # --- Номенклатура --------------------------------------------------------- def create_product( vault, name: str, unit: str = "шт", cost_price=0, retail_price=0, note: str = "", ) -> Product: name = name.strip() if not name: raise ValidationError("У товара должно быть название.") product = Product( id=new_id("p"), name=name, unit=unit.strip() or "шт", cost_price=m.money(cost_price), retail_price=m.money(retail_price), note=note.strip(), ) # Стартовая цена — тоже точка истории, иначе первый период остался бы # без объяснения, откуда взялась цифра. product.price_history.append( PricePoint( date=date.today(), cost_price=product.cost_price, retail_price=product.retail_price, note="создание товара", ) ) vault.doc.products.append(product) _record( vault, "product.create", "product", product.id, product.name, _as_created(_flat_product(vault.doc, product)), ) return product def update_product( vault, product_id: str, name: str | None = None, unit: str | None = None, note: str | None = None, archived: bool | None = None, ) -> Product: """Правка карточки без цен — для цен есть change_prices с историей.""" doc = vault.doc product = doc.product(product_id) if product is None: raise ValidationError("Товар не найден.") before = _flat_product(doc, product) if name is not None: if not name.strip(): raise ValidationError("У товара должно быть название.") product.name = name.strip() if unit is not None: product.unit = unit.strip() or "шт" if note is not None: product.note = note.strip() if archived is not None: product.archived = bool(archived) changes = _diff(before, _flat_product(doc, product)) if changes: action = "product.archive" if archived is not None and archived else "product.update" _record(vault, action, "product", product.id, product.name, changes) return product def change_prices( vault, product_id: str, cost_price, retail_price, on_date: date | None = None, note: str = "", ) -> Product: """Поменять цены товара, оставив след в истории цен.""" doc = vault.doc product = doc.product(product_id) if product is None: raise ValidationError("Товар не найден.") cost_price = m.money(cost_price) retail_price = m.money(retail_price) if cost_price < 0 or retail_price < 0: raise ValidationError("Цена не может быть отрицательной.") if cost_price == product.cost_price and retail_price == product.retail_price: return product before = _flat_product(doc, product) product.cost_price = cost_price product.retail_price = retail_price product.price_history.append( PricePoint( date=on_date or date.today(), cost_price=cost_price, retail_price=retail_price, note=note.strip(), ) ) _record( vault, "product.price_change", "product", product.id, product.name, _diff(before, _flat_product(doc, product)), ) return product def set_packs(vault, product_id: str, packs: list[dict]) -> Product: """Задать фасовки товара. Уже записанные документы помнят размер фасовки слепком, поэтому правка здесь не пересчитывает историю — меняется только то, что подставится в новые строки. """ doc = vault.doc product = doc.product(product_id) if product is None: raise ValidationError("Товар не найден.") built: list[Pack] = [] seen = {product.unit} for raw in packs: name = (raw.get("name") or "").strip() if not name: raise ValidationError("У фасовки должно быть название.") if name in seen: raise ValidationError(f"Фасовка «{name}» повторяется или совпадает с единицей товара.") seen.add(name) size = m.qty(raw.get("size")) if size <= 0: raise ValidationError(f"Размер фасовки «{name}» должен быть больше нуля.") price = m.money(raw.get("retail_price")) if price < 0: raise ValidationError("Цена фасовки не может быть отрицательной.") built.append(Pack(name=name, size=size, retail_price=price)) before = _flat_product(doc, product) product.packs = built changes = _diff(before, _flat_product(doc, product)) if changes: _record(vault, "product.packs", "product", product.id, product.name, changes) return product def delete_product(vault, product_id: str) -> None: """Удалить товар, если он нигде не использован. Удалять использованный нельзя: закупки и продажи ссылаются на него, и отчёты по закрытым периодам молча превратились бы в «(удалённый товар)». Для вышедших из оборота булок есть архив. """ doc = vault.doc product = doc.product(product_id) if product is None: raise ValidationError("Товар не найден.") used_in_batches = sum( 1 for b in doc.batches for line in b.lines if line.product_id == product_id ) used_in_sales = sum( 1 for s in doc.sales for line in s.lines if line.product_id == product_id ) if used_in_batches or used_in_sales: raise ValidationError( f"«{product.name}» используется в закупках ({used_in_batches}) " f"и продажах ({used_in_sales}), удалить нельзя. " "Отправь товар в архив — он исчезнет из списков, а история останется." ) flat = _flat_product(doc, product) doc.products.remove(product) _record(vault, "product.delete", "product", product_id, product.name, _as_deleted(flat)) # --- Контрагенты ---------------------------------------------------------- def create_counterparty(vault, name: str, kind: str = "friend", note: str = "") -> Counterparty: name = name.strip() if not name: raise ValidationError("У контрагента должно быть имя.") cp = Counterparty(id=new_id("c"), name=name, kind=kind, note=note.strip()) vault.doc.counterparties.append(cp) _record( vault, "counterparty.create", "counterparty", cp.id, cp.name, _as_created(_flat_counterparty(vault.doc, cp)), ) return cp def update_counterparty( vault, cp_id: str, name: str | None = None, kind: str | None = None, note: str | None = None, ) -> Counterparty: doc = vault.doc cp = doc.counterparty(cp_id) if cp is None: raise ValidationError("Контрагент не найден.") before = _flat_counterparty(doc, cp) if name is not None: if not name.strip(): raise ValidationError("У контрагента должно быть имя.") cp.name = name.strip() if kind is not None: cp.kind = kind if note is not None: cp.note = note.strip() changes = _diff(before, _flat_counterparty(doc, cp)) if changes: _record(vault, "counterparty.update", "counterparty", cp.id, cp.name, changes) return cp def delete_counterparty(vault, cp_id: str) -> None: doc = vault.doc cp = doc.counterparty(cp_id) if cp is None: raise ValidationError("Контрагент не найден.") used = sum(1 for s in doc.sales if s.counterparty_id == cp_id) if used: raise ValidationError( f"На «{cp.name}» ссылаются продажи ({used}), удалить нельзя." ) flat = _flat_counterparty(doc, cp) doc.counterparties.remove(cp) _record(vault, "counterparty.delete", "counterparty", cp_id, cp.name, _as_deleted(flat)) # --- Закупки -------------------------------------------------------------- def _resolve_line(doc, raw: dict, what: str) -> tuple[Product, Decimal, str, Decimal]: """Проверить строку документа и разложить её на товар, количество и фасовку.""" product_id = raw.get("product_id") product = doc.product(product_id) if product_id else None if product is None: raise ValidationError(f"В {what} есть строка без существующего товара.") quantity = m.qty(raw.get("qty")) if quantity <= 0: raise ValidationError(f"Количество для «{product.name}» должно быть больше нуля.") uom = (raw.get("uom") or "").strip() if not uom or uom == product.unit: return product, quantity, product.unit, ONE pack = product.pack(uom) if pack is None: raise ValidationError(f"У товара «{product.name}» нет фасовки «{uom}».") # Размер запоминается слепком: переопределят пачку — старые документы # не должны поехать. return product, quantity, pack.name, pack.size def _build_batch_lines(doc, lines: list[dict]) -> list[BatchLine]: result: list[BatchLine] = [] for raw in lines: _, quantity, uom, size = _resolve_line(doc, raw, "закупке") cost = m.money(raw.get("unit_cost")) if cost < 0: raise ValidationError("Себестоимость не может быть отрицательной.") result.append( BatchLine( product_id=raw["product_id"], qty=quantity, uom=uom, uom_size=size, unit_cost=cost, ) ) if not result: raise ValidationError("В закупке должна быть хотя бы одна строка.") return result def create_batch( vault, on_date: date, due_date: date, lines: list[dict], note: str = "", ) -> Batch: doc = vault.doc on_date = parse_date(on_date) due_date = parse_date(due_date) if due_date < on_date: raise ValidationError("Дата покрытия не может быть раньше даты закупки.") batch = Batch( id=new_id("b"), number=doc.next_batch_number(), date=on_date, due_date=due_date, lines=_build_batch_lines(doc, lines), note=note.strip(), ) doc.batches.append(batch) _record( vault, "batch.create", "batch", batch.id, _batch_label(batch), _as_created(_flat_batch(doc, batch)), ) return batch def update_batch( vault, batch_id: str, on_date: date | None = None, due_date: date | None = None, lines: list[dict] | None = None, note: str | None = None, ) -> Batch: doc = vault.doc batch = doc.batch(batch_id) if batch is None: raise ValidationError("Партия не найдена.") before = _flat_batch(doc, batch) new_date = parse_date(on_date) if on_date is not None else batch.date new_due = parse_date(due_date) if due_date is not None else batch.due_date if new_due < new_date: raise ValidationError("Дата покрытия не может быть раньше даты закупки.") batch.date = new_date batch.due_date = new_due if lines is not None: batch.lines = _build_batch_lines(doc, lines) if note is not None: batch.note = note.strip() changes = _diff(before, _flat_batch(doc, batch)) if changes: _record(vault, "batch.update", "batch", batch.id, _batch_label(batch), changes) return batch def set_batch_closed(vault, batch_id: str, closed: bool) -> Batch: doc = vault.doc batch = doc.batch(batch_id) if batch is None: raise ValidationError("Партия не найдена.") if batch.closed == closed: return batch before = _flat_batch(doc, batch) batch.closed = closed _record( vault, "batch.close" if closed else "batch.reopen", "batch", batch.id, _batch_label(batch), _diff(before, _flat_batch(doc, batch)), ) return batch def delete_batch(vault, batch_id: str) -> None: doc = vault.doc batch = doc.batch(batch_id) if batch is None: raise ValidationError("Партия не найдена.") label = _batch_label(batch) flat = _flat_batch(doc, batch) doc.batches.remove(batch) # Платежи, привязанные к удалённой партии, отвязываем, а не удаляем: # деньги пекарне реально ушли, и терять их из учёта нельзя. for payment in doc.bakery_payments: if payment.batch_id == batch_id: payment.batch_id = None _record(vault, "batch.delete", "batch", batch_id, label, _as_deleted(flat)) # --- Продажи и выбытие ---------------------------------------------------- def _build_sale_lines(doc, lines: list[dict]) -> list[SaleLine]: result: list[SaleLine] = [] for raw in lines: _, quantity, uom, size = _resolve_line(doc, raw, "продаже") price = m.money(raw.get("unit_price")) if price < 0: raise ValidationError("Цена не может быть отрицательной.") result.append( SaleLine( product_id=raw["product_id"], qty=quantity, uom=uom, uom_size=size, unit_price=price, ) ) if not result: raise ValidationError("В продаже должна быть хотя бы одна строка.") return result def split_overpayment(total: Decimal, paid: Decimal, tip: Decimal) -> tuple[Decimal, Decimal]: """Разложить полученные деньги на оплату товара и чаевые. Дали больше, чем стоит булка, — значит сдачу оставили. Излишек уходит в чаевые, а не раздувает «оплачено»: иначе долг и процент оплаты продажи считались бы по числу, которого за товар никто не должен. """ if paid > total: return total, m.money(tip + (paid - total)) return paid, tip def create_sale( vault, on_date: date, kind: str, lines: list[dict], counterparty_id: str | None = None, note: str = "", paid_amount=None, tip=None, ) -> Sale: """Записать выбытие товара. paid_amount=None означает «оплачено полностью» — самый частый случай. Явный ноль или частичная сумма создают долг, а переплата — чаевые. """ doc = vault.doc sale = Sale( id=new_id("s"), date=parse_date(on_date), kind=kind, counterparty_id=counterparty_id or None, lines=_build_sale_lines(doc, lines), note=note.strip(), ) if not sale.is_consumption: amount = sale.total if paid_amount is None else m.money(paid_amount) tip_amount = m.money(tip or 0) if amount < 0 or tip_amount < 0: raise ValidationError("Оплата и чаевые не могут быть отрицательными.") amount, tip_amount = split_overpayment(sale.total, amount, tip_amount) sale.tip = tip_amount if amount > 0: sale.payments.append(Payment(date=sale.date, amount=amount)) doc.sales.append(sale) _record( vault, "sale.create", "sale", sale.id, _sale_label(doc, sale), _as_created(_flat_sale(doc, sale)), ) return sale def update_sale( vault, sale_id: str, on_date: date | None = None, kind: str | None = None, lines: list[dict] | None = None, counterparty_id: str | None = None, note: str | None = None, tip=None, ) -> Sale: """Правка продажи. Платежи не трогает — для них есть отдельные операции.""" doc = vault.doc sale = doc.sale(sale_id) if sale is None: raise ValidationError("Продажа не найдена.") before = _flat_sale(doc, sale) if on_date is not None: sale.date = parse_date(on_date) if kind is not None: sale.kind = kind if lines is not None: sale.lines = _build_sale_lines(doc, lines) if counterparty_id is not None: sale.counterparty_id = counterparty_id or None if note is not None: sale.note = note.strip() if tip is not None: amount = m.money(tip) if amount < 0: raise ValidationError("Чаевые не могут быть отрицательными.") sale.tip = amount if sale.is_consumption: # Съеденное и подаренное денег не приносит по определению. sale.tip = m.ZERO changes = _diff(before, _flat_sale(doc, sale)) if changes: _record(vault, "sale.update", "sale", sale.id, _sale_label(doc, sale), changes) return sale def delete_sale(vault, sale_id: str) -> None: doc = vault.doc sale = doc.sale(sale_id) if sale is None: raise ValidationError("Продажа не найдена.") label = _sale_label(doc, sale) flat = _flat_sale(doc, sale) doc.sales.remove(sale) _record(vault, "sale.delete", "sale", sale_id, label, _as_deleted(flat)) def add_sale_payment(vault, sale_id: str, amount, on_date: date | None = None, note: str = "") -> Sale: """Принять оплату по продаже. Заплатили больше, чем оставалось должны, — излишек становится чаевыми, а не исчезает. Раньше он просто терялся: долг упирался в ноль, и деньги нигде не отражались. """ doc = vault.doc sale = doc.sale(sale_id) if sale is None: raise ValidationError("Продажа не найдена.") amount = m.money(amount) if amount <= 0: raise ValidationError("Сумма оплаты должна быть больше нуля.") before = _flat_sale(doc, sale) payment, sale.tip = split_overpayment(sale.debt, amount, sale.tip) if payment > 0: sale.payments.append( Payment( date=parse_date(on_date) if on_date else date.today(), amount=payment, note=note.strip(), ) ) _record( vault, "sale.payment_add", "sale", sale.id, _sale_label(doc, sale), _diff(before, _flat_sale(doc, sale)), ) return sale def set_sale_tip(vault, sale_id: str, amount) -> Sale: """Задать чаевые по уже записанной продаже.""" doc = vault.doc sale = doc.sale(sale_id) if sale is None: raise ValidationError("Продажа не найдена.") if sale.is_consumption: raise ValidationError("По съеденному и подаренному чаевых не бывает.") amount = m.money(amount) if amount < 0: raise ValidationError("Чаевые не могут быть отрицательными.") if amount == sale.tip: return sale before = _flat_sale(doc, sale) sale.tip = amount _record( vault, "sale.tip", "sale", sale.id, _sale_label(doc, sale), _diff(before, _flat_sale(doc, sale)), ) return sale def delete_sale_payment(vault, sale_id: str, index: int) -> Sale: doc = vault.doc sale = doc.sale(sale_id) if sale is None: raise ValidationError("Продажа не найдена.") if not 0 <= index < len(sale.payments): raise ValidationError("Такой оплаты нет.") before = _flat_sale(doc, sale) sale.payments.pop(index) _record( vault, "sale.payment_delete", "sale", sale.id, _sale_label(doc, sale), _diff(before, _flat_sale(doc, sale)), ) return sale # --- Расчёты с пекарней --------------------------------------------------- def add_bakery_payment( vault, amount, on_date: date | None = None, batch_id: str | None = None, note: str = "", ) -> BakeryPayment: doc = vault.doc amount = m.money(amount) if amount <= 0: raise ValidationError("Сумма платежа должна быть больше нуля.") if batch_id and doc.batch(batch_id) is None: raise ValidationError("Партия не найдена.") payment = BakeryPayment( id=new_id("bp"), date=parse_date(on_date) if on_date else date.today(), amount=amount, batch_id=batch_id or None, note=note.strip(), ) doc.bakery_payments.append(payment) batch = doc.batch(batch_id) if batch_id else None _record( vault, "bakery_payment.add", "bakery_payment", payment.id, f"Платёж пекарне {m.fmt_money(amount, doc.settings.currency)}", [ Change("дата", "", dump_date(payment.date)), Change("сумма", "", m.fmt_money(amount, doc.settings.currency)), Change("партия", "", _batch_label(batch) if batch else "без привязки"), Change("заметка", "", payment.note), ], ) return payment def delete_bakery_payment(vault, payment_id: str) -> None: doc = vault.doc payment = next((p for p in doc.bakery_payments if p.id == payment_id), None) if payment is None: raise ValidationError("Платёж не найден.") doc.bakery_payments.remove(payment) _record( vault, "bakery_payment.delete", "bakery_payment", payment_id, f"Платёж пекарне {m.fmt_money(payment.amount, doc.settings.currency)}", [ Change("дата", dump_date(payment.date), ""), Change("сумма", m.fmt_money(payment.amount, doc.settings.currency), ""), ], ) # --- Настройки ------------------------------------------------------------ def update_settings(vault, currency=None, default_period_days=None, git: dict | None = None): """Правка настроек. Значение токена в журнал не пишется — иначе секрет, ради которого всё и шифруется, лежал бы открытым текстом внутри той же базы и уезжал бы в открытый экспорт. """ doc = vault.doc settings = doc.settings changes: list[Change] = [] if currency is not None and currency != settings.currency: changes.append(Change("валюта", settings.currency, currency)) settings.currency = currency if default_period_days is not None and int(default_period_days) != settings.default_period_days: changes.append( Change("период по умолчанию", str(settings.default_period_days), str(default_period_days)) ) settings.default_period_days = int(default_period_days) for key, value in (git or {}).items(): if not hasattr(settings.git, key): continue old = getattr(settings.git, key) if isinstance(old, bool): value = bool(value) elif isinstance(old, int) and not isinstance(old, bool): value = int(value) if old == value: continue setattr(settings.git, key, value) if key in SECRET_SETTINGS: changes.append(Change(f"git.{key}", "скрыто", "изменён" if value else "удалён")) else: changes.append(Change(f"git.{key}", str(old), str(value))) if changes: _record(vault, "settings.update", "settings", "settings", "Настройки", changes) return settings # --- Обслуживание журнала ------------------------------------------------- def prune_journal(vault, before: date) -> int: """Выбросить записи журнала старше указанной даты. Возвращает их число.""" doc = vault.doc cutoff = dump_date(parse_date(before)) kept = [e for e in doc.journal if e.ts[:10] >= cutoff] removed = len(doc.journal) - len(kept) if not removed: return 0 doc.journal = kept _record( vault, "journal.prune", "journal", "journal", "Журнал", [Change("удалено записей старше " + cutoff, "", str(removed))], ) return removed