120 lines
5.7 KiB
Python
120 lines
5.7 KiB
Python
"""Источник: 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")
|
||
|
||
# Тип проводки (решение 35): закрывающие проводки помечены явно, свёртка ОФР не разбирает memo.
|
||
KIND_ORDINARY = "" # обычная операция
|
||
KIND_CLOSE_90 = "close90" # перенос сальдо 90.01..90.08 на 90.09 — в ОФР не входит
|
||
KIND_CLOSE_99 = "close99" # закрытие 90.09 и 91 на 99 — в ОФР не входит
|
||
KIND_CLOSE_COST = "close_cost" # закрытие 26/44 на 90.08/90.07 — это расходы периода, входит в ОФР
|
||
|
||
|
||
@dataclass(frozen=True, slots=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, slots=True)
|
||
class Posting:
|
||
day: date
|
||
dt: str
|
||
dt_an: An | None
|
||
ct: str
|
||
ct_an: An | None
|
||
amount: Decimal
|
||
memo: str
|
||
kind: str = KIND_ORDINARY
|
||
|
||
|
||
Key = tuple[str, An | None]
|
||
|
||
|
||
class Ledger:
|
||
def __init__(self, company: str):
|
||
self.company = company
|
||
self.postings: list[Posting] = []
|
||
self._cache: dict = {}
|
||
# Индекс для быстрого сальдо «на конец»: накопленные итоги по (счёт, аналитика) и последний день проводок.
|
||
# Закрытие месяца спрашивает сальдо на дату не раньше всех проводок — тогда достаточно пройти по ключам, а не по проводкам.
|
||
self._tot: dict[Key, Decimal] = {}
|
||
self._maxday: date | None = None
|
||
self.meta: dict = {} # произвольные данные модели компании (используются economy)
|
||
|
||
def post(self, day, dt, dt_an, ct, ct_an, amount, memo, kind: str = KIND_ORDINARY) -> 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, kind))
|
||
self._cache.clear()
|
||
kd, kc = (dt, dt_an), (ct, ct_an)
|
||
self._tot[kd] = self._tot.get(kd, Decimal(0)) + amount
|
||
self._tot[kc] = self._tot.get(kc, Decimal(0)) - amount
|
||
if self._maxday is None or day > self._maxday:
|
||
self._maxday = day
|
||
|
||
@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] = {}
|
||
if self._maxday is None or at >= self._maxday:
|
||
for key, v in self._tot.items():
|
||
if self._match(key[0], account_prefix):
|
||
res[key] = v
|
||
else:
|
||
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, "входящий остаток")
|