RSHB/arm-core
9
0
Fork 0
Code Issues 26 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

добавить расчёт шахматки и результирующую таблицу ВГО

Browse Source
This commit is contained in:
PotapovaA 2026-10-12 14:26:10 +03:00
parent bb6795c052
commit f0b5d19162
8 changed files with 688 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
api/src/api/dependencies/consolidations.py
View File

@ -15,6 +15,7 @@ from src.services.consolidation_status import ConsolidationStatusService
from src.services.consolidations import ConsolidationService
from src.services.counterparty_mapping import CounterpartyMappingService
from src.services.elimination_rules import EliminationRuleService
from src.services.intercompany_matrix import IntercompanyMatrixService
def get_consolidation_service(
@ -84,3 +85,15 @@ def get_elimination_rule_service(
EliminationRuleServiceDep = Annotated[EliminationRuleService, Depends(get_elimination_rule_service)]
def get_intercompany_matrix_service(
session: SessionDep,
scope: Annotated[VisibilityScope, Depends(get_scope)],
osv_service: ConsolidatedOsvServiceDep,
) -> IntercompanyMatrixService:
"""Сервис отчёта intercompany-matrix (шахматки)."""
return IntercompanyMatrixService(session, scope, osv_service)
IntercompanyMatrixServiceDep = Annotated[IntercompanyMatrixService, Depends(get_intercompany_matrix_service)]

2
api/src/api/endpoints/consolidations/__init__.py
View File

@ -6,6 +6,7 @@ from src.api.endpoints.consolidations import (
checklist,
elimination_rule,
export,
intercompany_matrix,
locks,
mapping,
members,
@ -26,3 +27,4 @@ router.include_router(osv.router)
router.include_router(export.router)
router.include_router(mapping.router)
router.include_router(elimination_rule.router)
router.include_router(intercompany_matrix.router)

86
api/src/api/endpoints/consolidations/intercompany_matrix.py Normal file
View File

@ -0,0 +1,86 @@
"""Расчёт шахматки по выбранным счетам или правилу элиминации."""
import uuid
from typing import Any
from fastapi import APIRouter, Depends, status
from src.api.dependencies.auth import require
from src.api.dependencies.consolidations import IntercompanyMatrixServiceDep
from src.schemas.common import ErrorDetail
from src.schemas.consolidations import IntercompanyMatrixManualIn, IntercompanyMatrixOut, IntercompanyResultTableOut
router = APIRouter(prefix="/consolidations/versions")
VIEW = [Depends(require("CONSOLIDATION_VIEW"))]
ERRORS: dict[int | str, dict[str, Any]] = {
status.HTTP_404_NOT_FOUND: {"model": ErrorDetail, "description": "Версия или правило не найдены либо недоступны"},
status.HTTP_422_UNPROCESSABLE_CONTENT: {"model": ErrorDetail, "description": "Неверный выбор счетов или правило"},
}
@router.get(
"/{version_id}/intercompany-matrix/all",
response_model=IntercompanyMatrixOut,
dependencies=VIEW,
responses=ERRORS,
summary="Рассчитать шахматку по всем счетам",
)
async def build_intercompany_matrix_all(
version_id: uuid.UUID,
service: IntercompanyMatrixServiceDep,
) -> IntercompanyMatrixOut:
"""Все счета с подходящим видом дебета или кредита."""
return await service.build_accounts(version_id, mode="ALL")
@router.post(
"/{version_id}/intercompany-matrix/manual",
response_model=IntercompanyMatrixOut,
dependencies=VIEW,
responses=ERRORS,
summary="Рассчитать шахматку по выбранным счетам",
)
async def build_intercompany_matrix_manual(
version_id: uuid.UUID,
selection: IntercompanyMatrixManualIn,
service: IntercompanyMatrixServiceDep,
) -> IntercompanyMatrixOut:
"""Выбранные счета раскрываются по иерархии."""
return await service.build_accounts(
version_id,
mode="MANUAL",
debit_account_ids=selection.debit_account_ids,
credit_account_ids=selection.credit_account_ids,
)
@router.get(
"/{version_id}/intercompany-matrix/result-table",
response_model=IntercompanyResultTableOut,
dependencies=VIEW,
responses=ERRORS,
summary="Результирующая таблица ВГО",
)
async def get_intercompany_result_table(
version_id: uuid.UUID,
service: IntercompanyMatrixServiceDep,
) -> IntercompanyResultTableOut:
"""Пять фиксированных статей: суммы по компаниям и итоги по группе."""
return await service.build_result_table(version_id)
@router.get(
"/{version_id}/intercompany-matrix/presets/{rule_no}",
response_model=IntercompanyMatrixOut,
dependencies=VIEW,
responses=ERRORS,
summary="Рассчитать шахматку по пресету",
)
async def build_intercompany_matrix_preset(
version_id: uuid.UUID,
rule_no: int,
service: IntercompanyMatrixServiceDep,
) -> IntercompanyMatrixOut:
"""Правило выбирается по номеру из версии правил консолидации."""
return await service.build_preset(version_id, rule_no)

3
api/src/domain/consolidated_osv.py
View File

@ -101,6 +101,7 @@ class SourceRow:
turnover_credit: Decimal | None
closing_debit: Decimal | None
closing_credit: Decimal | None
account_id: uuid.UUID | None = None
@dataclass(frozen=True)
@ -132,6 +133,7 @@ class ConsolidatedRow:
closing_debit: Decimal | None
closing_credit: Decimal | None
warnings: tuple[str, ...] = ()
account_id: uuid.UUID | None = None
@dataclass(frozen=True)
@ -342,6 +344,7 @@ def _compose( # noqa: PLR0913 - сборка строки свода из ча
legal_entity_inn=entity.inn,
account_code=representative.account_code,
account_root=representative.account_root,
account_id=representative.account_id,
period=context.period,
is_cumulative=is_cumulative,
row_number=representative.row_number,

55
api/src/schemas/consolidations.py
View File

@ -325,3 +325,58 @@ class EliminationRuleOut(BaseModel):
measure: Literal["closing", "turnover"]
company_1: IntercompanyMatrixRuleSideOut
company_2: IntercompanyMatrixRuleSideOut
class IntercompanyMatrixManualIn(BaseModel):
"""Выбранные дебетовые и кредитовые счета для ручного расчёта шахматки."""
debit_account_ids: list[uuid.UUID] = Field(min_length=1)
credit_account_ids: list[uuid.UUID] = Field(min_length=1)
class IntercompanyMatrixCellOut(BaseModel):
"""Суммы и расхождение для пары компаний A и Б."""
company_a_id: uuid.UUID
company_b_id: uuid.UUID
amount_a: Decimal
amount_b: Decimal
variance_amount: Decimal
variance_percent: Decimal | None
class IntercompanyMatrixCompanyOut(BaseModel):
"""Компания на обеих осях шахматки."""
id: uuid.UUID
name: str
class IntercompanyMatrixOut(BaseModel):
"""Компании задают порядок обеих осей; диагональные ячейки отсутствуют."""
companies: list[IntercompanyMatrixCompanyOut]
cells: list[IntercompanyMatrixCellOut]
class IntercompanyResultAmountOut(BaseModel):
"""Сумма статьи у компании по всем остальным компаниям периметра."""
company_id: uuid.UUID
amount: Decimal
class IntercompanyResultArticleOut(BaseModel):
"""Статья ВГО и её суммы по компаниям и группе."""
code: str
name: str
amounts: list[IntercompanyResultAmountOut]
group_total: Decimal
class IntercompanyResultTableOut(BaseModel):
"""Фиксированная результирующая таблица ВГО выбранной версии."""
companies: list[IntercompanyMatrixCompanyOut]
articles: list[IntercompanyResultArticleOut]

11
api/src/services/consolidated_osv.py
View File

@ -93,6 +93,7 @@ def _source_row(entry: OsvEntry) -> domain.SourceRow:
row_number=entry.row_number,
account_code=entry.account_code,
account_root=entry.account_root,
account_id=entry.account_id,
counterparty_name=entry.counterparty_name,
counterparty_inn=entry.counterparty_inn,
counterparty_id=entry.counterparty_id,
@ -134,7 +135,7 @@ class ConsolidatedOsvService:
self._session = session
self._scope = scope
async def _build(self, version_id: uuid.UUID) -> tuple[domain.Consolidated, ConsolidationVersion, Consolidation]:
async def build(self, version_id: uuid.UUID) -> tuple[domain.Consolidated, ConsolidationVersion, Consolidation]:
"""Свод версии (`domain.build`) вместе с версией и консолидацией; версия вне видимости — 404."""
found = await repo.get_version(self._session, self._scope, version_id)
if found is None:
@ -164,7 +165,7 @@ class ConsolidatedOsvService:
async def page(self, version_id: uuid.UUID, f: OsvFilters, page: OsvPageParams) -> Page[ConsolidatedRowOut]:
"""Страница свода после фильтров: `total` — число строк после фильтра, срез — в памяти (порядок D-16)."""
consolidated, _, _ = await self._build(version_id)
consolidated, _, _ = await self.build(version_id)
rows = domain.apply_filter(consolidated.rows, domain_filter(f))
window = rows[page.offset : page.offset + page.limit]
return Page[ConsolidatedRowOut](
@ -175,7 +176,7 @@ class ConsolidatedOsvService:
"""xlsx сводной ОСВ (spec «Экспорт xlsx», D-14): строки после тех же фильтров, что у `page`, без пагинации, в
порядке D-16; лист пробелов — по своду целиком. Имя файла — `svod-osv_<ГК>_<год>-<квартал>_v<n>.xlsx` (без
percent-кодирования: оно в маршруте). `audit_event(EXPORT)` пишется после сборки книги, в транзакции запроса."""
consolidated, version, consolidation = await self._build(version_id)
consolidated, version, consolidation = await self.build(version_id)
rows = domain.apply_filter(consolidated.rows, domain_filter(f))
content = await run_in_threadpool(build_workbook, consolidated, rows)
group_name = (
@ -202,7 +203,7 @@ class ConsolidatedOsvService:
async def summary(self, version_id: uuid.UUID) -> OsvSummaryOut:
"""Сводка: строки, несопоставленные, автосопоставленные по имени/алиасу, ОСВ-документы версии и пробелы."""
consolidated, _, _ = await self._build(version_id)
consolidated, _, _ = await self.build(version_id)
rows_total, unmapped_total, auto_matched_total = domain.summary(consolidated)
return OsvSummaryOut(
rows_total=rows_total,
@ -215,7 +216,7 @@ class ConsolidatedOsvService:
async def row_for_entry(self, version_id: uuid.UUID, entry_id: uuid.UUID) -> ConsolidatedRowOut | None:
"""Строка свода, в `entry_ids` которой есть `entry_id`; `None` — строка в своде не участвует (пара с пробелом
или документ не выбран источником при наличии нарастающего)."""
consolidated, _, _ = await self._build(version_id)
consolidated, _, _ = await self.build(version_id)
row = next((row for row in consolidated.rows if entry_id in row.entry_ids), None)
return None if row is None else row_out(row)

338
api/src/services/intercompany_matrix.py Normal file
View File

@ -0,0 +1,338 @@
"""Расчёт шахматки по готовой сводной ОСВ и выбранным наборам счетов."""
import uuid
from collections import defaultdict
from collections.abc import Iterable, Sequence
from dataclasses import dataclass
from decimal import Decimal
from typing import Literal, cast
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from src.common.exceptions import DomainValidationError, NotFoundError
from src.domain.consolidated_osv import ConsolidatedRow
from src.domain.policy import VisibilityScope
from src.models.accounting import ChartOfAccount, EliminationRuleAccount
from src.models.enums import AccountKind
from src.repositories import consolidations as repo
from src.repositories import elimination_rules as rules_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")
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
side_1: Side
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}")
if selection.side_1 not in ("D", "K"):
raise ValueError(f"неизвестная сторона правила: {selection.side_1}")
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)
side_2: Side = "K" if selection.side_1 == "D" else "D"
amounts_1: dict[tuple[uuid.UUID, uuid.UUID], Decimal] = defaultdict(lambda: ZERO)
amounts_2: dict[tuple[uuid.UUID, uuid.UUID], Decimal] = defaultdict(lambda: ZERO)
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 or row.account_id is None:
continue
pair = (owner, counterparty)
if row.account_id in selection.company_1_account_ids:
amounts_1[pair] += _amount(row, selection.measure, selection.side_1)
if row.account_id in selection.company_2_account_ids:
amounts_2[pair] += _amount(row, selection.measure, side_2)
cells: list[IntercompanyMatrixCellOut] = []
for company_a in company_ids:
for company_b in company_ids:
if company_a == company_b:
continue
amount_a = amounts_1.get((company_a, company_b), ZERO)
amount_b = amounts_2.get((company_b, company_a), 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
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,
)
)
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:
raise DomainValidationError("ACCOUNT_NOT_FOUND", "Выбранный счёт не найден в плане счетов")
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 _preset_accounts(
accounts: Sequence[EliminationRuleAccount],
company_no: int,
) -> tuple[set[uuid.UUID], set[uuid.UUID]]:
included = {
account.account_id for account in accounts if account.company_no == company_no and not account.is_excluded
}
excluded = {account.account_id for account in accounts if account.company_no == company_no and account.is_excluded}
return included, excluded
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/PRESET выполняется вызывающим кодом."""
consolidated, version, _ = await self._osv_service.build(version_id)
return await self._calculate(consolidated.rows, version.id, selection)
async def _calculate(
self, rows: Iterable[ConsolidatedRow], version_id: uuid.UUID, selection: MatrixSelection
) -> IntercompanyMatrixOut:
"""Общий вызов расчёта после проверки видимости версии в сервисе сводной ОСВ."""
members = await repo.members(self._session, version_id)
companies = [IntercompanyMatrixCompanyOut(id=entity.id, name=entity.name) for _, entity in members]
return calculate_matrix(rows, companies, selection)
async def build_result_table(self, version_id: uuid.UUID) -> IntercompanyResultTableOut:
"""Статичная таблица по пяти статьям, независимо от режима и пресета шахматки."""
consolidated, version, _ = await self._osv_service.build(version_id)
members = await repo.members(self._session, version.id)
accounts = list((await self._session.scalars(select(ChartOfAccount))).all())
companies = [IntercompanyMatrixCompanyOut(id=entity.id, name=entity.name) for _, entity in members]
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 раскрывает выбранные счета и потомков."""
consolidated, version, _ = await self._osv_service.build(version_id)
accounts = list((await self._session.scalars(select(ChartOfAccount))).all())
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", "D")
return await self._calculate(consolidated.rows, version.id, selection)
async def build_preset(self, version_id: uuid.UUID, rule_no: int) -> IntercompanyMatrixOut:
"""Пресет выбирается по номеру из версии правил; учитываются показатель, направление и исключения."""
consolidated, version, _ = await self._osv_service.build(version_id)
if version.elimination_rules_version is None:
raise DomainValidationError("RULES_VERSION_NOT_SET", "Для версии консолидации не задана версия правил")
rule_rows = await rules_repo.list_rules_with_accounts(self._session, version_id)
selected_rows = [(rule, account) for rule, account in rule_rows if rule.rule_no == rule_no]
if not selected_rows:
raise NotFoundError("правило элиминации не найдено")
if len({rule.id for rule, _ in selected_rows}) > 1:
raise DomainValidationError("AMBIGUOUS_RULE_NO", "Номер правила не уникален в версии правил")
rule = selected_rows[0][0]
if rule.measure not in ("closing", "turnover") or rule.side_1 not in ("D", "K"):
raise DomainValidationError("INVALID_RULE", "У правила неверный показатель или направление")
rule_accounts = [account for _, account in selected_rows if account is not None]
accounts = list((await self._session.scalars(select(ChartOfAccount))).all())
by_id, children = _account_tree(accounts)
side_1 = cast(Side, rule.side_1)
side_2: Side = "K" if side_1 == "D" else "D"
included_1, excluded_1 = _preset_accounts(rule_accounts, 1)
included_2, excluded_2 = _preset_accounts(rule_accounts, 2)
accounts_1 = _selected_accounts(included_1, excluded_1, side_1, by_id, children)
accounts_2 = _selected_accounts(included_2, excluded_2, side_2, by_id, children)
if not accounts_1 or not accounts_2:
raise DomainValidationError("RULE_ACCOUNTS_EMPTY", "У правила нет счетов одной из сторон")
selection = MatrixSelection(accounts_1, accounts_2, cast(Measure, rule.measure), side_1)
return await self._calculate(consolidated.rows, version.id, selection)

185
api/tests/test_intercompany_matrix.py Normal file
View File

@ -0,0 +1,185 @@
"""Выбор счетов и показателя в режимах шахматки без обращения к БД."""
import uuid
from decimal import Decimal
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from src.api.endpoints.consolidations.intercompany_matrix import router
from src.common.exceptions import DomainValidationError
from src.models.enums import AccountKind, StatementType
from src.schemas.consolidations import IntercompanyMatrixCompanyOut, IntercompanyMatrixOut, IntercompanyResultTableOut
from src.services import intercompany_matrix as matrix
def _account(kind: AccountKind, parent_id: uuid.UUID | None = None) -> SimpleNamespace:
return SimpleNamespace(id=uuid.uuid4(), kind=kind, parent_id=parent_id)
def _row(
owner: uuid.UUID,
counterparty: uuid.UUID,
account_id: uuid.UUID,
*,
debit: str = "0",
credit: str = "0",
) -> SimpleNamespace:
return SimpleNamespace(
legal_entity_id=owner,
counterparty_id=counterparty,
account_id=account_id,
closing_debit=Decimal(debit),
closing_credit=Decimal(credit),
turnover_debit=Decimal("0"),
turnover_credit=Decimal("0"),
)
def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None:
paths = {(route.path, tuple(sorted(route.methods))) for route in router.routes}
assert ("/consolidations/versions/{version_id}/intercompany-matrix/all", ("GET",)) in paths
assert ("/consolidations/versions/{version_id}/intercompany-matrix/manual", ("POST",)) in paths
assert ("/consolidations/versions/{version_id}/intercompany-matrix/result-table", ("GET",)) in paths
assert (
"/consolidations/versions/{version_id}/intercompany-matrix/presets/{rule_no}",
("GET",),
) in paths
debit_parent = _account(AccountKind.A)
debit_child = _account(AccountKind.A, debit_parent.id)
credit_parent = _account(AccountKind.P)
credit_child = _account(AccountKind.P, credit_parent.id)
by_id, children = matrix._account_tree([debit_parent, debit_child, credit_parent, credit_child])
debit = matrix._selected_accounts([debit_parent.id], [], "D", by_id, children)
credit = matrix._selected_accounts([credit_parent.id], [], "K", by_id, children)
assert debit == {debit_parent.id, debit_child.id}
assert credit == {credit_parent.id, credit_child.id}
company_a, company_b = uuid.uuid4(), uuid.uuid4()
result = matrix.calculate_matrix(
[
_row(company_a, company_b, debit_child.id, debit="100"),
_row(company_b, company_a, credit_child.id, credit="90"),
],
[
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
],
matrix.MatrixSelection(debit, credit, "closing", "D"),
)
assert [(company.id, company.name) for company in result.companies] == [
(company_a, "Компания А"),
(company_b, "Компания Б"),
]
assert result.cells[0].amount_a == Decimal("100")
assert result.cells[0].amount_b == Decimal("90")
assert result.cells[0].variance_percent == Decimal("100") / Decimal("9")
assert isinstance(result, IntercompanyMatrixOut)
response = result
assert "exceeds_threshold" not in response.model_dump(mode="json")["cells"][0]
assert response.model_dump(mode="json")["companies"] == [
{"id": str(company_a), "name": "Компания А"},
{"id": str(company_b), "name": "Компания Б"},
]
def test_result_table_uses_fixed_articles_and_group_totals() -> None:
company_a, company_b, external = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
required_codes = {
code for article in matrix.RESULT_ARTICLES for code in (*article.account_codes, *article.excluded_codes)
}
kinds = {
code: AccountKind.A if article.side == "D" else AccountKind.P
for article in matrix.RESULT_ARTICLES
for code in article.account_codes
}
kinds.update({"76": AccountKind.AP, "76.07.1": AccountKind.P, "76.07.5": AccountKind.A, "76.07.9": AccountKind.A})
ids = {code: uuid.uuid4() for code in required_codes}
parents = {"67.03": "67", "66.03": "66", "66.04": "66", "76.07.1": "76", "76.07.5": "76", "76.07.9": "76"}
accounts = [
SimpleNamespace(id=ids[code], code=code, kind=kinds[code], parent_id=ids.get(parents.get(code)))
for code in required_codes
]
rows = [
_row(company_a, company_b, ids["60.02"], debit="100"),
_row(company_a, company_b, ids["76.07.5"], debit="50"),
_row(company_a, company_b, ids["76.07.1"], credit="20"),
_row(company_a, company_b, ids["58.01"], debit="30"),
_row(company_b, company_a, ids["60.01"], credit="70"),
_row(company_b, company_a, ids["67.03"], credit="40"),
_row(company_a, company_b, ids["66.04"], credit="10"),
_row(company_a, external, ids["60.02"], debit="999"),
]
result = matrix.calculate_result_table(
rows,
[
IntercompanyMatrixCompanyOut(id=company_a, name="Компания А"),
IntercompanyMatrixCompanyOut(id=company_b, name="Компания Б"),
],
accounts,
)
articles = {article.code: article for article in result.articles}
assert list(articles) == [article.code for article in matrix.RESULT_ARTICLES]
assert [item.amount for item in articles["receivables"].amounts] == [Decimal("100"), Decimal("0")]
assert articles["receivables"].group_total == Decimal("100")
assert articles["payables"].group_total == Decimal("70")
assert articles["financial_investments"].group_total == Decimal("30")
assert articles["long_term_loans"].group_total == Decimal("40")
assert articles["short_term_loans"].group_total == Decimal("10")
assert isinstance(result, IntercompanyResultTableOut)
response = result
assert response.articles[0].amounts[0].company_id == company_a
@pytest.mark.asyncio
async def test_preset_uses_rule_version_measure_side_and_exclusions(monkeypatch: pytest.MonkeyPatch) -> None:
version_id, company_a, company_b = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
debit_root = _account(AccountKind.A)
debit_excluded = _account(AccountKind.A, debit_root.id)
credit_root = _account(AccountKind.P)
rows = [
_row(company_a, company_b, credit_root.id, credit="999"),
_row(company_b, company_a, debit_root.id, debit="999"),
_row(company_b, company_a, debit_excluded.id, debit="999"),
]
rows[0].turnover_credit = Decimal("80")
rows[1].turnover_debit = Decimal("70")
rows[2].turnover_debit = Decimal("20")
version = SimpleNamespace(id=version_id, elimination_rules_version=2)
osv = SimpleNamespace(build=AsyncMock(return_value=(SimpleNamespace(rows=rows), version, None)))
session = SimpleNamespace(
scalars=AsyncMock(return_value=SimpleNamespace(all=lambda: [debit_root, debit_excluded, credit_root]))
)
rule = SimpleNamespace(
id=uuid.uuid4(), statement_type=StatementType.income, rule_no=3, measure="turnover", side_1="K"
)
rule_accounts = [
SimpleNamespace(company_no=1, account_id=credit_root.id, is_excluded=False),
SimpleNamespace(company_no=2, account_id=debit_root.id, is_excluded=False),
SimpleNamespace(company_no=2, account_id=debit_excluded.id, is_excluded=True),
]
list_rules = AsyncMock(return_value=[(rule, account) for account in rule_accounts])
members = AsyncMock(
return_value=[
(None, SimpleNamespace(id=company_a, name="Компания А")),
(None, SimpleNamespace(id=company_b, name="Компания Б")),
]
)
monkeypatch.setattr(matrix.rules_repo, "list_rules_with_accounts", list_rules)
monkeypatch.setattr(matrix.repo, "members", members)
service = matrix.IntercompanyMatrixService(session, SimpleNamespace(), osv)
result = await service.build_preset(version_id, 3)
list_rules.assert_awaited_once_with(session, version_id)
assert result.cells[0].amount_a == Decimal("80")
assert result.cells[0].amount_b == Decimal("70")
assert result.cells[0].variance_amount == Decimal("10")
assert [company.name for company in result.companies] == ["Компания А", "Компания Б"]
duplicate = SimpleNamespace(id=uuid.uuid4(), statement_type=StatementType.balance, rule_no=3)
list_rules.return_value = [(rule, rule_accounts[0]), (duplicate, None)]
with pytest.raises(DomainValidationError, match="Номер правила не уникален"):
await service.build_preset(version_id, 3)