342 lines
15 KiB
Python
342 lines
15 KiB
Python
"""ОСВ по счёту в оформлении 1С 8.3 (спека §6.1): свёртка книги в строки и запись xlsx.
|
||
|
||
`rows` свёртывает книгу компании в строки «счёт → субсчёт → контрагент» за период (нарастающим итогом с 1 января):
|
||
сальдо на начало/конец — по каждой строке одной стороной (Дт или Кт), обороты — обе; строка родителя равна сумме детей
|
||
по каждой из шести колонок. Строки без сальдо и оборотов не пишутся, как в 1С. `write` рисует лист по шаблону
|
||
`templates/osv_style.json` (собран из образца скриптом `build_osv_style.py`); у ОСВ с ИНН между блоком текста и блоком
|
||
сумм вставляется колонка ИНН, остальная сетка сдвигается на одну колонку (решение 3).
|
||
|
||
Производительность (решение 33): для свёртки по периодам книга один раз индексируется по (счёт, контрагент) с
|
||
накопленными итогами по дням; `Ledger.balance/turnover` на каждый период проходили бы по всем проводкам.
|
||
"""
|
||
|
||
import json
|
||
from bisect import bisect_right
|
||
from dataclasses import dataclass
|
||
from datetime import timedelta
|
||
from decimal import Decimal
|
||
from functools import cache
|
||
from pathlib import Path
|
||
|
||
import openpyxl
|
||
from openpyxl.styles import Alignment, Border, Font, PatternFill, Side
|
||
from openpyxl.utils import get_column_letter
|
||
from openpyxl.worksheet.properties import Outline
|
||
|
||
from synthgen import xlsx
|
||
from synthgen.counterparties import OSV_ACCOUNTS
|
||
from synthgen.group import Group, Member, variant_spelling
|
||
from synthgen.ledger import Ledger
|
||
from synthgen.periods import Period
|
||
|
||
ZERO = Decimal(0)
|
||
STYLE_PATH = Path(__file__).with_name("templates") / "osv_style.json"
|
||
SHEET = "Лист_1"
|
||
Six = tuple[Decimal, Decimal, Decimal, Decimal, Decimal, Decimal]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class OsvRow:
|
||
level: int # 0 счёт, 1 субсчёт, 2 контрагент (глубже — вложенные субсчета)
|
||
label: str # код счёта/субсчёта или имя контрагента
|
||
inn: str | None
|
||
od: Decimal # сальдо на начало, Дт
|
||
oc: Decimal # сальдо на начало, Кт
|
||
td: Decimal # обороты, Дт
|
||
tc: Decimal # обороты, Кт
|
||
cd: Decimal # сальдо на конец, Дт
|
||
cc: Decimal # сальдо на конец, Кт
|
||
is_cp: bool = False # строка контрагента
|
||
|
||
@property
|
||
def six(self) -> Six:
|
||
return (self.od, self.oc, self.td, self.tc, self.cd, self.cc)
|
||
|
||
|
||
def file_name(account: str, period: Period) -> str:
|
||
# метка периода заканчивается точкой после «г», она же — точка перед расширением
|
||
return f"сч {account}_{period.label_1c[:-1]}.xlsx"
|
||
|
||
|
||
# --- индекс книги -----------------------------------------------------------------------------------------------
|
||
|
||
|
||
class _Series:
|
||
"""Накопленные по дням итоги одного ключа (счёт, контрагент): сальдо (Дт − Кт), обороты Дт и Кт."""
|
||
|
||
__slots__ = ("days", "net", "dt", "ct")
|
||
|
||
def __init__(self, items: list[tuple[int, Decimal, Decimal]]):
|
||
items.sort(key=lambda x: x[0])
|
||
self.days = [d for d, _, _ in items]
|
||
self.net, self.dt, self.ct = [], [], []
|
||
n = a = b = ZERO
|
||
for _, debit, credit in items:
|
||
a += debit
|
||
b += credit
|
||
n = a - b
|
||
self.net.append(n)
|
||
self.dt.append(a)
|
||
self.ct.append(b)
|
||
|
||
def _at(self, arr: list[Decimal], ordinal: int) -> Decimal:
|
||
i = bisect_right(self.days, ordinal)
|
||
return arr[i - 1] if i else ZERO
|
||
|
||
def amounts(self, start_ord: int, end_ord: int) -> tuple[Decimal, Decimal, Decimal, Decimal]:
|
||
"""(сальдо на начало, обороты Дт, обороты Кт, сальдо на конец) за [start, end]."""
|
||
before = start_ord - 1
|
||
return (
|
||
self._at(self.net, before),
|
||
self._at(self.dt, end_ord) - self._at(self.dt, before),
|
||
self._at(self.ct, end_ord) - self._at(self.ct, before),
|
||
self._at(self.net, end_ord),
|
||
)
|
||
|
||
|
||
def _index(ledger: Ledger) -> dict[str, dict[tuple[str, str | None], _Series]]:
|
||
"""Корень счёта → {(код счёта, контрагент или None): серия}; пересобирается при изменении числа проводок."""
|
||
cached = ledger.meta.get("_osv_index")
|
||
if cached is not None and cached[0] == len(ledger.postings):
|
||
return cached[1]
|
||
raw: dict[str, dict[tuple[str, str | None], list[tuple[int, Decimal, Decimal]]]] = {}
|
||
for p in ledger.postings:
|
||
ordinal = p.day.toordinal()
|
||
for code, an, debit, credit in ((p.dt, p.dt_an, p.amount, ZERO), (p.ct, p.ct_an, ZERO, p.amount)):
|
||
root = code.split(".")[0]
|
||
if root in OSV_ACCOUNTS:
|
||
raw.setdefault(root, {}).setdefault((code, an.cp if an is not None else None), []).append(
|
||
(ordinal, debit, credit)
|
||
)
|
||
index = {root: {key: _Series(items) for key, items in by_key.items()} for root, by_key in raw.items()}
|
||
ledger.meta["_osv_index"] = (len(ledger.postings), index)
|
||
return index
|
||
|
||
|
||
# --- имена контрагентов -----------------------------------------------------------------------------------------
|
||
|
||
_DIRECTORY: dict[int, tuple[Group, dict[str, tuple[str, str | None]]]] = {}
|
||
|
||
|
||
def _directory(group: Group) -> dict[str, tuple[str, str | None]]:
|
||
"""Код аналитики контрагента → (имя в 1С, ИНН): внешние и банки по ИНН, члены группы по коду, миноритарии."""
|
||
cached = _DIRECTORY.get(id(group))
|
||
if cached is not None and cached[0] is group:
|
||
return cached[1]
|
||
out: dict[str, tuple[str, str | None]] = {}
|
||
for org in [*group.externals, *group.banks, *(m.minority for m in group.members if m.minority)]:
|
||
out[org.inn] = (org.name, org.inn)
|
||
for m in group.members:
|
||
out[m.code] = (m.name_1c, m.inn)
|
||
_DIRECTORY[id(group)] = (group, out)
|
||
return out
|
||
|
||
|
||
def _identity(cp: str, member: Member, group: Group) -> tuple[str, str | None]:
|
||
entry = _directory(group).get(cp)
|
||
if entry is None:
|
||
raise ValueError(f"{member.code}: контрагент {cp!r} неизвестен группе (в ОСВ не выгружается)")
|
||
name, inn = entry
|
||
if member.code in group.no_inn:
|
||
variant_code = group.variant_name.get(member.code)
|
||
if variant_code == cp:
|
||
name = variant_spelling(next(m for m in group.members if m.code == cp))
|
||
return name, None
|
||
return name, inn
|
||
|
||
|
||
# --- свёртка ----------------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _natural_key(code: str) -> list[tuple[int, int, str]]:
|
||
return [(0, int(part), "") if part.isdigit() else (1, 0, part) for part in code.split(".")]
|
||
|
||
|
||
def _sided(net: Decimal) -> tuple[Decimal, Decimal]:
|
||
return (net, ZERO) if net > 0 else (ZERO, -net)
|
||
|
||
|
||
def _add(a: Six, b: Six) -> Six:
|
||
return tuple(x + y for x, y in zip(a, b, strict=True)) # type: ignore[return-value]
|
||
|
||
|
||
def rows(ledger: Ledger, account: str, period: Period, member: Member, group: Group) -> list[OsvRow]:
|
||
"""Строки ОСВ по счёту в порядке вывода (счёт, субсчета по возрастанию кода, контрагенты по имени); без «Итого»."""
|
||
start, end = period.start.toordinal(), period.end.toordinal()
|
||
leaves: dict[str, dict[str | None, Six]] = {}
|
||
for (code, cp), series in _index(ledger).get(account, {}).items():
|
||
opening, td, tc, closing = series.amounts(start, end)
|
||
od, oc = _sided(opening)
|
||
cd, cc = _sided(closing)
|
||
six: Six = (od, oc, td, tc, cd, cc)
|
||
if any(six):
|
||
leaves.setdefault(code, {})[cp] = six
|
||
|
||
children: dict[str, set[str]] = {}
|
||
for code in leaves:
|
||
parts = code.split(".")
|
||
for i in range(len(parts)):
|
||
prefix = ".".join(parts[: i + 1])
|
||
children.setdefault(prefix, set())
|
||
if i:
|
||
children[".".join(parts[:i])].add(prefix)
|
||
for code, subs in children.items():
|
||
if subs and code in leaves:
|
||
raise ValueError(f"{member.code} {account}: счёт {code} и с проводками, и с субсчетами")
|
||
|
||
def leaf_rows(code: str, level: int) -> tuple[list[OsvRow], Six]:
|
||
by_cp = leaves[code]
|
||
if None in by_cp:
|
||
if len(by_cp) > 1:
|
||
raise ValueError(f"{member.code} {code}: строки без контрагента и с контрагентом вместе")
|
||
return [], by_cp[None]
|
||
named = sorted(((*_identity(cp, member, group), cp) for cp in by_cp if cp is not None), key=lambda x: (x[0], x[2]))
|
||
out: list[OsvRow] = []
|
||
total: Six = (ZERO,) * 6
|
||
for name, inn, cp in named:
|
||
six = by_cp[cp]
|
||
total = _add(total, six)
|
||
out.append(OsvRow(level, name, inn, *six, is_cp=True))
|
||
return out, total
|
||
|
||
def build(code: str, level: int) -> tuple[list[OsvRow], Six]:
|
||
if code in leaves:
|
||
body, total = leaf_rows(code, level + 1)
|
||
else:
|
||
body, total = [], (ZERO,) * 6
|
||
for sub in sorted(children[code], key=_natural_key):
|
||
sub_rows, sub_total = build(sub, level + 1)
|
||
body += sub_rows
|
||
total = _add(total, sub_total)
|
||
return [OsvRow(level, code, None, *total), *body], total
|
||
|
||
if account not in children:
|
||
return []
|
||
return build(account, 0)[0]
|
||
|
||
|
||
def total_of(data: list[OsvRow]) -> Six:
|
||
"""«Итого» листа: сумма строк счетов уровня 0 (в листе она одна; у пустой ОСВ — нули)."""
|
||
result: Six = (ZERO,) * 6
|
||
for r in data:
|
||
if r.level == 0:
|
||
result = _add(result, r.six)
|
||
return result
|
||
|
||
|
||
# --- лист -------------------------------------------------------------------------------------------------------
|
||
|
||
|
||
@cache
|
||
def _style() -> dict:
|
||
return json.loads(STYLE_PATH.read_text(encoding="utf-8"))
|
||
|
||
|
||
def _side(spec: dict | None) -> Side:
|
||
if not spec:
|
||
return Side()
|
||
return Side(style=spec["style"], color=spec["color"]) if spec["color"] else Side(style=spec["style"])
|
||
|
||
|
||
class _Styles:
|
||
"""Готовые объекты openpyxl по шаблону: (тип строки, колонка шаблона) → (шрифт, заливка, рамка, выравнивание, формат)."""
|
||
|
||
def __init__(self, tpl: dict):
|
||
self._tpl = tpl
|
||
self._cache: dict[tuple[str, int, int | None], tuple] = {}
|
||
|
||
def get(self, kind: str, col: int, indent: int | None = None) -> tuple:
|
||
key = (kind, col, indent)
|
||
hit = self._cache.get(key)
|
||
if hit is None:
|
||
spec = self._spec(kind)["cells"][col - 1]
|
||
f, a = spec["font"], spec["align"]
|
||
font = Font(name=f["name"], sz=f["sz"], b=f["b"], i=f["i"], color=f["color"])
|
||
fill = PatternFill("solid", fgColor=spec["fill"]) if spec["fill"] else PatternFill()
|
||
bd = spec["border"]
|
||
border = Border(left=_side(bd["left"]), right=_side(bd["right"]), top=_side(bd["top"]), bottom=_side(bd["bottom"]))
|
||
align = Alignment(horizontal=a["h"], vertical=a["v"], wrap_text=a["wrap"], indent=a["indent"] if indent is None else indent)
|
||
hit = self._cache[key] = (font, fill, border, align, spec["nf"])
|
||
return hit
|
||
|
||
def _spec(self, kind: str) -> dict:
|
||
if kind.startswith("header"):
|
||
return self._tpl["header"][int(kind[6:]) - 1]
|
||
return self._tpl["total"] if kind == "total" else self._tpl["rows"][kind]
|
||
|
||
|
||
def _apply(cell, style: tuple) -> None:
|
||
cell.font, cell.fill, cell.border, cell.alignment, cell.number_format = style
|
||
|
||
|
||
def write(path: Path, member: Member, account: str, period: Period, data: list[OsvRow], with_inn: bool) -> None:
|
||
"""Пишет лист ОСВ по шаблону 1С; `with_inn` вставляет колонку ИНН после блока текста (E)."""
|
||
tpl = _style()
|
||
st = _Styles(tpl)
|
||
text_end = tpl["text_span"][1] # колонка D
|
||
inn_col = text_end + 1
|
||
shift = 1 if with_inn else 0
|
||
|
||
def col(c: int) -> int: # колонка шаблона → колонка листа
|
||
return c + shift if c > text_end else c
|
||
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = SHEET
|
||
ws.sheet_properties.outlinePr = Outline(summaryBelow=bool(tpl["summary_below"]), summaryRight=True)
|
||
ncols = tpl["ncols"] + shift
|
||
for c, width in tpl["col_widths"].items():
|
||
ws.column_dimensions[get_column_letter(col(int(c)))].width = width
|
||
if with_inn:
|
||
ws.column_dimensions[get_column_letter(inn_col)].width = tpl["inn_width"]
|
||
|
||
def style_row(r: int, kind: str, indent: int | None = None) -> None:
|
||
for c in range(1, ncols + 1):
|
||
tc = c if c <= text_end else (text_end if c == inn_col and with_inn else c - shift)
|
||
_apply(ws.cell(r, c), st.get(kind, tc, indent if tc == 1 else None))
|
||
|
||
placeholders = {"{org}": member.name, "{title}": f"Оборотно-сальдовая ведомость по счету {account} за {period.label_1c}"}
|
||
for i, row in enumerate(tpl["header"], start=1):
|
||
style_row(i, f"header{i}")
|
||
ws.row_dimensions[i].height = row["height"]
|
||
ws.row_dimensions[i].outlineLevel = row["outline"]
|
||
for c, value in row["values"].items():
|
||
ws.cell(i, col(int(c)), placeholders.get(value, value))
|
||
if with_inn:
|
||
ws.cell(10, inn_col, "ИНН")
|
||
for r1, c1, r2, c2 in tpl["header_merges"]:
|
||
ws.merge_cells(start_row=r1, start_column=col(c1), end_row=r2, end_column=col(c2))
|
||
if with_inn:
|
||
ws.merge_cells(start_row=10, start_column=inn_col, end_row=11, end_column=inn_col)
|
||
|
||
spans = [(col(a), col(b)) for a, b in tpl["sum_spans"]]
|
||
|
||
def put_sums(r: int, six: Six) -> None:
|
||
for (a, b), value in zip(spans, six, strict=True):
|
||
if value != 0:
|
||
ws.cell(r, a, float(value))
|
||
if b > a:
|
||
ws.merge_cells(start_row=r, start_column=a, end_row=r, end_column=b)
|
||
|
||
def put_label(r: int, text: str) -> None:
|
||
ws.cell(r, 1, text)
|
||
ws.merge_cells(start_row=r, start_column=1, end_row=r, end_column=text_end)
|
||
|
||
r = len(tpl["header"]) + 1
|
||
for item in data:
|
||
kind = "cp" if item.is_cp else ("account" if item.level == 0 else "sub")
|
||
style_row(r, kind, item.level)
|
||
spec = tpl["rows"][kind]
|
||
ws.row_dimensions[r].height = spec["height"]
|
||
ws.row_dimensions[r].outlineLevel = min(item.level, 7)
|
||
put_label(r, item.label)
|
||
if with_inn and item.inn:
|
||
ws.cell(r, inn_col, item.inn)
|
||
put_sums(r, item.six)
|
||
r += 1
|
||
style_row(r, "total")
|
||
ws.row_dimensions[r].height = tpl["total"]["height"]
|
||
put_label(r, "Итого")
|
||
put_sums(r, total_of(data))
|
||
xlsx.save_deterministic(wb, path)
|