diff --git a/api/src/api/dependencies/consolidations.py b/api/src/api/dependencies/consolidations.py index 8fdcfcf..600689a 100644 --- a/api/src/api/dependencies/consolidations.py +++ b/api/src/api/dependencies/consolidations.py @@ -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)] diff --git a/api/src/api/endpoints/consolidations/__init__.py b/api/src/api/endpoints/consolidations/__init__.py index d34008d..934c11e 100644 --- a/api/src/api/endpoints/consolidations/__init__.py +++ b/api/src/api/endpoints/consolidations/__init__.py @@ -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) diff --git a/api/src/api/endpoints/consolidations/intercompany_matrix.py b/api/src/api/endpoints/consolidations/intercompany_matrix.py new file mode 100644 index 0000000..04d12f5 --- /dev/null +++ b/api/src/api/endpoints/consolidations/intercompany_matrix.py @@ -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) diff --git a/api/src/domain/consolidated_osv.py b/api/src/domain/consolidated_osv.py index 8e6eb62..abb5ae7 100644 --- a/api/src/domain/consolidated_osv.py +++ b/api/src/domain/consolidated_osv.py @@ -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, diff --git a/api/src/schemas/consolidations.py b/api/src/schemas/consolidations.py index 98f9057..55e4ccb 100644 --- a/api/src/schemas/consolidations.py +++ b/api/src/schemas/consolidations.py @@ -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] diff --git a/api/src/services/consolidated_osv.py b/api/src/services/consolidated_osv.py index c20686b..b7cd4cc 100644 --- a/api/src/services/consolidated_osv.py +++ b/api/src/services/consolidated_osv.py @@ -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.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) diff --git a/api/src/services/intercompany_matrix.py b/api/src/services/intercompany_matrix.py new file mode 100644 index 0000000..f6a95cb --- /dev/null +++ b/api/src/services/intercompany_matrix.py @@ -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) diff --git a/api/tests/test_intercompany_matrix.py b/api/tests/test_intercompany_matrix.py new file mode 100644 index 0000000..a11c274 --- /dev/null +++ b/api/tests/test_intercompany_matrix.py @@ -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)