"""Источник: synthetic-group/gen/ledger.py (#35). Книга проводок компании: двойная запись, сальдо и обороты как свёртка списка.""" import csv from dataclasses import dataclass from datetime import date from decimal import Decimal from pathlib import Path CENT = Decimal("0.01") @dataclass(frozen=True) class An: cp: str | None contract: str | None = None item: str | None = None name_override: str | None = None inn_override: str | None = None @dataclass(frozen=True) class Posting: day: date dt: str dt_an: An | None ct: str ct_an: An | None amount: Decimal memo: str Key = tuple[str, An | None] class Ledger: def __init__(self, company: str): self.company = company self.postings: list[Posting] = [] self._cache: dict = {} def post(self, day, dt, dt_an, ct, ct_an, amount, memo) -> None: if amount <= 0 or amount != amount.quantize(CENT): raise ValueError(f"сумма должна быть > 0 с двумя знаками: {amount}") self.postings.append(Posting(day, dt, dt_an, ct, ct_an, amount, memo)) self._cache.clear() @staticmethod def _match(code: str, prefix: str | None) -> bool: return prefix is None or code == prefix or code.startswith(prefix + ".") def balance(self, at: date, account_prefix: str | None = None) -> dict[Key, Decimal]: ck = ("b", at, account_prefix) if ck not in self._cache: res: dict[Key, Decimal] = {} for p in self.postings: if p.day > at: continue if self._match(p.dt, account_prefix): res[(p.dt, p.dt_an)] = res.get((p.dt, p.dt_an), Decimal(0)) + p.amount if self._match(p.ct, account_prefix): res[(p.ct, p.ct_an)] = res.get((p.ct, p.ct_an), Decimal(0)) - p.amount self._cache[ck] = {k: v for k, v in res.items() if v != 0} return dict(self._cache[ck]) def turnover(self, start: date, end: date, account_prefix: str | None = None) -> dict[Key, tuple[Decimal, Decimal]]: ck = ("t", start, end, account_prefix) if ck not in self._cache: res: dict[Key, list[Decimal]] = {} for p in self.postings: if not (start <= p.day <= end): continue if self._match(p.dt, account_prefix): res.setdefault((p.dt, p.dt_an), [Decimal(0), Decimal(0)])[0] += p.amount if self._match(p.ct, account_prefix): res.setdefault((p.ct, p.ct_an), [Decimal(0), Decimal(0)])[1] += p.amount self._cache[ck] = {k: (v[0], v[1]) for k, v in res.items()} return dict(self._cache[ck]) def to_csv(self, path: Path) -> None: with Path(path).open("w", encoding="utf-8", newline="") as f: w = csv.writer(f, delimiter=";", lineterminator="\n") w.writerow("day;dt;dt_cp;dt_contract;dt_item;ct;ct_cp;ct_contract;ct_item;amount;memo".split(";")) for p in self.postings: da, ca = p.dt_an, p.ct_an w.writerow([ p.day.isoformat(), p.dt, (da.cp or "") if da else "", (da.contract or "") if da else "", (da.item or "") if da else "", p.ct, (ca.cp or "") if ca else "", (ca.contract or "") if ca else "", (ca.item or "") if ca else "", f"{p.amount:.2f}", p.memo, ]) def opening(ledger: Ledger, day: date, account: str, an: An | None, amount: Decimal) -> None: """Входящий остаток проводкой против вспомогательного счёта 000; amount со знаком (Дт > 0).""" if amount > 0: ledger.post(day, account, an, "000", None, amount, "входящий остаток") elif amount < 0: ledger.post(day, "000", None, account, an, -amount, "входящий остаток")