377 lines
16 KiB
Python
377 lines
16 KiB
Python
"""Расчёт шахматки по готовой сводной ОСВ и выбранным наборам счетов."""
|
||
|
||
import uuid
|
||
from collections import defaultdict
|
||
from collections.abc import Iterable, Sequence
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal
|
||
from typing import Literal
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.common.exceptions import DomainValidationError
|
||
from src.domain.consolidated_osv import ConsolidatedRow, Entity
|
||
from src.domain.policy import VisibilityScope
|
||
from src.models.accounting import ChartOfAccount
|
||
from src.models.consolidation import Consolidation, ConsolidationVersion
|
||
from src.models.directories import CompanyGroup
|
||
from src.models.enums import AccountKind, PeriodQuarter
|
||
from src.repositories import directories as directory_repo
|
||
from src.schemas.consolidations import (
|
||
IntercompanyMatrixCellOut,
|
||
IntercompanyMatrixCompanyOut,
|
||
IntercompanyMatrixOut,
|
||
IntercompanyResultAmountOut,
|
||
IntercompanyResultArticleOut,
|
||
IntercompanyResultTableOut,
|
||
)
|
||
from src.services.consolidated_osv import ConsolidatedOsvService
|
||
|
||
Measure = Literal["closing", "turnover"]
|
||
Side = Literal["D", "K"]
|
||
ZERO = Decimal("0")
|
||
VARIANCE_THRESHOLD_PERCENT = Decimal("5")
|
||
DEBIT_KINDS = frozenset({AccountKind.A, AccountKind.AP})
|
||
CREDIT_KINDS = frozenset({AccountKind.P, AccountKind.AP})
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ResultArticleSpec:
|
||
code: str
|
||
name: str
|
||
side: Side
|
||
account_codes: tuple[str, ...]
|
||
excluded_codes: tuple[str, ...] = ()
|
||
|
||
|
||
RESULT_ARTICLES = (
|
||
ResultArticleSpec(
|
||
"receivables",
|
||
"Дебиторская задолженность",
|
||
"D",
|
||
("60.02", "60.22", "60.32", "62.01", "62.03", "62.21", "62.31", "76"),
|
||
("76.07.5", "76.07.9"),
|
||
),
|
||
ResultArticleSpec(
|
||
"payables",
|
||
"Кредиторская задолженность",
|
||
"K",
|
||
("60.01", "60.03", "60.21", "60.31", "62.02", "62.22", "62.32", "76"),
|
||
("76.07.1",),
|
||
),
|
||
ResultArticleSpec("financial_investments", "Финансовые вложения", "D", ("58.01", "58.03")),
|
||
ResultArticleSpec("long_term_loans", "Долгосрочные кредиты и займы", "K", ("67", "67.03")),
|
||
ResultArticleSpec("short_term_loans", "Краткосрочные кредиты и займы", "K", ("66", "66.03", "66.04")),
|
||
)
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MatrixSelection:
|
||
"""Счета дебетовой и кредитовой сторон и показатель выбранного режима."""
|
||
|
||
company_1_account_ids: frozenset[uuid.UUID]
|
||
company_2_account_ids: frozenset[uuid.UUID]
|
||
measure: Measure
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class MatrixExportInfo:
|
||
group_name: str
|
||
year: int
|
||
quarter: PeriodQuarter
|
||
version_no: int
|
||
mode: Literal["ALL", "MANUAL"]
|
||
debit_account_codes: tuple[str, ...]
|
||
credit_account_codes: tuple[str, ...]
|
||
|
||
@property
|
||
def filename(self) -> str:
|
||
return f"intercompany-matrix_{self.group_name}_{self.year}-{self.quarter.value}_v{self.version_no}.xlsx"
|
||
|
||
|
||
def _amount(row: ConsolidatedRow, measure: Measure, side: Side) -> Decimal:
|
||
if measure == "closing":
|
||
value = row.closing_debit if side == "D" else row.closing_credit
|
||
else:
|
||
value = row.turnover_debit if side == "D" else row.turnover_credit
|
||
return value if value is not None else ZERO
|
||
|
||
|
||
def calculate_matrix(
|
||
rows: Iterable[ConsolidatedRow],
|
||
member_companies: Sequence[IntercompanyMatrixCompanyOut],
|
||
selection: MatrixSelection,
|
||
) -> IntercompanyMatrixOut:
|
||
"""Суммы по парам компаний;
|
||
|
||
Сторона 1: дебет ОСВ компании A по контрагенту Б.
|
||
Сторона 2: кредит ОСВ компании Б по контрагенту A.
|
||
"""
|
||
if selection.measure not in ("closing", "turnover"):
|
||
raise ValueError(f"неизвестный показатель правила: {selection.measure}")
|
||
|
||
company_info = tuple({company.id: company for company in member_companies}.values())
|
||
company_ids = tuple(company.id for company in company_info)
|
||
members = set(company_ids)
|
||
amounts_1: dict[tuple[uuid.UUID, uuid.UUID], dict] = defaultdict(lambda: {"amount": ZERO, "entry_osv_ids": set()})
|
||
amounts_2: dict[tuple[uuid.UUID, uuid.UUID], dict] = defaultdict(lambda: {"amount": ZERO, "entry_osv_ids": set()})
|
||
skipped_entries: dict[tuple[uuid.UUID, uuid.UUID], int] = defaultdict(int)
|
||
|
||
for row in rows:
|
||
owner = row.legal_entity_id
|
||
counterparty = row.counterparty_id
|
||
if owner not in members or counterparty not in members or owner == counterparty:
|
||
continue
|
||
pair = (owner, counterparty)
|
||
if row.account_id is None:
|
||
skipped_entries[pair] += len(row.entry_ids)
|
||
continue
|
||
if row.account_id in selection.company_1_account_ids:
|
||
item = amounts_1[pair]
|
||
item["amount"] += _amount(row, selection.measure, "D")
|
||
item["entry_osv_ids"].update(row.entry_ids)
|
||
if row.account_id in selection.company_2_account_ids:
|
||
item = amounts_2[pair]
|
||
item["amount"] += _amount(row, selection.measure, "K")
|
||
item["entry_osv_ids"].update(row.entry_ids)
|
||
|
||
cells: list[IntercompanyMatrixCellOut] = []
|
||
for company_a in company_ids:
|
||
for company_b in company_ids:
|
||
if company_a == company_b:
|
||
continue
|
||
item_a = amounts_1.get((company_a, company_b))
|
||
item_b = amounts_2.get((company_b, company_a))
|
||
|
||
amount_a = item_a["amount"] if item_a is not None else ZERO
|
||
amount_b = item_b["amount"] if item_b is not None else ZERO
|
||
|
||
variance_amount = abs(amount_a - amount_b)
|
||
base = min(amount_a, amount_b)
|
||
variance_percent = None if base == ZERO else variance_amount / base * 100
|
||
exceeds_threshold = variance_percent is not None and variance_percent > VARIANCE_THRESHOLD_PERCENT
|
||
debit_ids = item_a["entry_osv_ids"] if item_a is not None else ()
|
||
credit_ids = item_b["entry_osv_ids"] if item_b is not None else ()
|
||
cells.append(
|
||
IntercompanyMatrixCellOut(
|
||
company_a_id=company_a,
|
||
company_b_id=company_b,
|
||
amount_a=amount_a,
|
||
amount_b=amount_b,
|
||
variance_amount=variance_amount,
|
||
variance_percent=variance_percent,
|
||
exceeds_threshold=exceeds_threshold,
|
||
one_sided=bool(debit_ids) != bool(credit_ids),
|
||
skipped_osv_entry_count=skipped_entries[(company_a, company_b)]
|
||
+ skipped_entries[(company_b, company_a)],
|
||
osv_entry_ids_debit=debit_ids,
|
||
osv_entry_ids_credit=credit_ids,
|
||
)
|
||
)
|
||
return IntercompanyMatrixOut(companies=list(company_info), cells=cells)
|
||
|
||
|
||
def _account_tree(
|
||
accounts: Sequence[ChartOfAccount],
|
||
) -> tuple[dict[uuid.UUID, ChartOfAccount], dict[uuid.UUID, list[uuid.UUID]]]:
|
||
by_id = {account.id: account for account in accounts}
|
||
children: dict[uuid.UUID, list[uuid.UUID]] = defaultdict(list)
|
||
for account in accounts:
|
||
if account.parent_id is not None:
|
||
children[account.parent_id].append(account.id)
|
||
return by_id, children
|
||
|
||
|
||
def _expand(roots: Iterable[uuid.UUID], children: dict[uuid.UUID, list[uuid.UUID]]) -> set[uuid.UUID]:
|
||
"""Выбранные счета и все их потомки."""
|
||
result: set[uuid.UUID] = set()
|
||
pending = list(roots)
|
||
while pending:
|
||
account_id = pending.pop()
|
||
if account_id not in result:
|
||
result.add(account_id)
|
||
pending.extend(children.get(account_id, ()))
|
||
return result
|
||
|
||
|
||
def _kinds(side: Side) -> frozenset[AccountKind]:
|
||
return DEBIT_KINDS if side == "D" else CREDIT_KINDS
|
||
|
||
|
||
def _selected_accounts(
|
||
included: Iterable[uuid.UUID],
|
||
excluded: Iterable[uuid.UUID],
|
||
side: Side,
|
||
by_id: dict[uuid.UUID, ChartOfAccount],
|
||
children: dict[uuid.UUID, list[uuid.UUID]],
|
||
) -> frozenset[uuid.UUID]:
|
||
roots = set(included)
|
||
missing = roots - by_id.keys()
|
||
if missing:
|
||
missing_ids = ", ".join(map(str, sorted(missing)))
|
||
raise DomainValidationError("ACCOUNT_NOT_FOUND", f"Выбранного счёта нет ({missing_ids})")
|
||
if any(by_id[account_id].kind not in _kinds(side) for account_id in roots):
|
||
raise DomainValidationError("ACCOUNT_KIND_MISMATCH", "Сторона счёта не соответствует направлению расчёта")
|
||
selected = _expand(roots, children) - _expand(excluded, children)
|
||
return frozenset(account_id for account_id in selected if by_id[account_id].kind in _kinds(side))
|
||
|
||
|
||
def calculate_result_table(
|
||
rows: Iterable[ConsolidatedRow],
|
||
member_companies: Sequence[IntercompanyMatrixCompanyOut],
|
||
accounts: Sequence[ChartOfAccount],
|
||
) -> IntercompanyResultTableOut:
|
||
"""Фиксированные статьи ВГО: сальдо по всем контрагентам периметра для каждого ЮЛ."""
|
||
by_id, children = _account_tree(accounts)
|
||
by_code = {account.code: account.id for account in accounts}
|
||
required_codes = {code for article in RESULT_ARTICLES for code in (*article.account_codes, *article.excluded_codes)}
|
||
missing = sorted(required_codes - by_code.keys())
|
||
if missing:
|
||
raise DomainValidationError("RESULT_ACCOUNTS_MISSING", f"В плане счетов нет счетов: {', '.join(missing)}")
|
||
|
||
selected = {
|
||
article.code: _selected_accounts(
|
||
(by_code[code] for code in article.account_codes),
|
||
(by_code[code] for code in article.excluded_codes),
|
||
article.side,
|
||
by_id,
|
||
children,
|
||
)
|
||
for article in RESULT_ARTICLES
|
||
}
|
||
companies = tuple({company.id: company for company in member_companies}.values())
|
||
member_ids = {company.id for company in companies}
|
||
amounts: dict[str, dict[uuid.UUID, Decimal]] = {
|
||
article.code: defaultdict(lambda: ZERO) for article in RESULT_ARTICLES
|
||
}
|
||
for row in rows:
|
||
if (
|
||
row.legal_entity_id not in member_ids
|
||
or row.counterparty_id not in member_ids
|
||
or row.legal_entity_id == row.counterparty_id
|
||
or row.account_id is None
|
||
):
|
||
continue
|
||
for article in RESULT_ARTICLES:
|
||
if row.account_id in selected[article.code]:
|
||
amounts[article.code][row.legal_entity_id] += _amount(row, "closing", article.side)
|
||
|
||
articles: list[IntercompanyResultArticleOut] = []
|
||
for article in RESULT_ARTICLES:
|
||
company_amounts = [
|
||
IntercompanyResultAmountOut(company_id=company.id, amount=amounts[article.code].get(company.id, ZERO))
|
||
for company in companies
|
||
]
|
||
articles.append(
|
||
IntercompanyResultArticleOut(
|
||
code=article.code,
|
||
name=article.name,
|
||
amounts=company_amounts,
|
||
group_total=sum((item.amount for item in company_amounts), ZERO),
|
||
)
|
||
)
|
||
return IntercompanyResultTableOut(companies=list(companies), articles=articles)
|
||
|
||
|
||
class IntercompanyMatrixService:
|
||
"""Получает свод и периметр версии, затем вызывает чистый расчёт матрицы."""
|
||
|
||
def __init__(self, session: AsyncSession, scope: VisibilityScope, service_osv: ConsolidatedOsvService) -> None:
|
||
self._session = session
|
||
self._scope = scope
|
||
self._osv_service = service_osv
|
||
|
||
async def build(
|
||
self,
|
||
version_id: uuid.UUID,
|
||
selection: MatrixSelection,
|
||
) -> IntercompanyMatrixOut:
|
||
"""Счета принимает готовыми; выбор режима ALL/MANUAL выполняется вызывающим кодом."""
|
||
consolidated, _, _, members = await self._osv_service.build_with_members(version_id)
|
||
return self._calculate(consolidated.rows, members, selection)
|
||
|
||
def _calculate(
|
||
self, rows: Iterable[ConsolidatedRow], members: dict[uuid.UUID, Entity], selection: MatrixSelection
|
||
) -> IntercompanyMatrixOut:
|
||
"""Общий вызов расчёта с участниками из уже построенного свода."""
|
||
companies = [IntercompanyMatrixCompanyOut(id=entity.id, name=entity.name) for entity in members.values()]
|
||
return calculate_matrix(rows, companies, selection)
|
||
|
||
async def build_result_table(self, version_id: uuid.UUID) -> IntercompanyResultTableOut:
|
||
"""Статичная таблица по пяти статьям"""
|
||
consolidated, _, _, members = await self._osv_service.build_with_members(version_id)
|
||
accounts = await directory_repo.list_accounts(self._session)
|
||
companies = [IntercompanyMatrixCompanyOut(id=entity.id, name=entity.name) for entity in members.values()]
|
||
return calculate_result_table(consolidated.rows, companies, accounts)
|
||
|
||
async def build_accounts(
|
||
self,
|
||
version_id: uuid.UUID,
|
||
*,
|
||
mode: Literal["ALL", "MANUAL"],
|
||
debit_account_ids: Sequence[uuid.UUID] = (),
|
||
credit_account_ids: Sequence[uuid.UUID] = (),
|
||
) -> IntercompanyMatrixOut:
|
||
"""ALL использует все счета нужного вида; MANUAL раскрывает выбранные счета и потомков."""
|
||
matrix, _, _, _ = await self._build_accounts(
|
||
version_id, mode=mode, debit_account_ids=debit_account_ids, credit_account_ids=credit_account_ids
|
||
)
|
||
return matrix
|
||
|
||
async def build_export(
|
||
self,
|
||
version_id: uuid.UUID,
|
||
*,
|
||
mode: Literal["ALL", "MANUAL"],
|
||
debit_account_ids: Sequence[uuid.UUID] = (),
|
||
credit_account_ids: Sequence[uuid.UUID] = (),
|
||
) -> tuple[IntercompanyMatrixOut, MatrixExportInfo]:
|
||
"""Матрица и реквизиты книги из одного расчёта свода."""
|
||
matrix, version, consolidation, accounts = await self._build_accounts(
|
||
version_id, mode=mode, debit_account_ids=debit_account_ids, credit_account_ids=credit_account_ids
|
||
)
|
||
group_name = (
|
||
await self._session.execute(
|
||
select(CompanyGroup.name).where(CompanyGroup.id == consolidation.company_group_id)
|
||
)
|
||
).scalar_one()
|
||
by_id = {account.id: account.code for account in accounts}
|
||
info = MatrixExportInfo(
|
||
group_name=group_name,
|
||
year=consolidation.year,
|
||
quarter=consolidation.quarter,
|
||
version_no=version.version_no,
|
||
mode=mode,
|
||
debit_account_codes=tuple(by_id[account_id] for account_id in debit_account_ids),
|
||
credit_account_codes=tuple(by_id[account_id] for account_id in credit_account_ids),
|
||
)
|
||
return matrix, info
|
||
|
||
async def _build_accounts(
|
||
self,
|
||
version_id: uuid.UUID,
|
||
*,
|
||
mode: Literal["ALL", "MANUAL"],
|
||
debit_account_ids: Sequence[uuid.UUID],
|
||
credit_account_ids: Sequence[uuid.UUID],
|
||
) -> tuple[IntercompanyMatrixOut, ConsolidationVersion, Consolidation, list[ChartOfAccount]]:
|
||
consolidated, version, consolidation, members = await self._osv_service.build_with_members(version_id)
|
||
accounts = await directory_repo.list_accounts(self._session)
|
||
by_id, children = _account_tree(accounts)
|
||
if mode == "ALL":
|
||
if debit_account_ids or credit_account_ids:
|
||
raise DomainValidationError("ACCOUNTS_NOT_ALLOWED", "Для режима ALL не нужно указывать счета")
|
||
debit = frozenset(account.id for account in accounts if account.kind in DEBIT_KINDS)
|
||
credit = frozenset(account.id for account in accounts if account.kind in CREDIT_KINDS)
|
||
elif mode == "MANUAL":
|
||
if not debit_account_ids or not credit_account_ids:
|
||
raise DomainValidationError("ACCOUNTS_REQUIRED", "Выберите дебетовые и кредитовые счета")
|
||
debit = _selected_accounts(debit_account_ids, (), "D", by_id, children)
|
||
credit = _selected_accounts(credit_account_ids, (), "K", by_id, children)
|
||
else:
|
||
raise DomainValidationError("INVALID_MODE", "Неизвестный режим выбора счетов")
|
||
selection = MatrixSelection(debit, credit, "closing")
|
||
matrix = self._calculate(consolidated.rows, members, selection)
|
||
return matrix, version, consolidation, accounts
|