"""Расчёт шахматки по готовой сводной ОСВ и выбранным наборам счетов.""" 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, NotFoundError from src.domain.consolidated_osv import ConsolidatedRow, Entity from src.domain.elimination_packing import ( PackedGroup, PackRow, RowKey, RuleTotals, calculate_rule_totals, pack_rule_rows, ) from src.domain.policy import VisibilityScope from src.models.accounting import ChartOfAccount, EliminationRule from src.models.consolidation import Consolidation, ConsolidationVersion from src.models.directories import CompanyGroup from src.models.enums import AccountKind, PeriodQuarter, StatementType from src.repositories import directories as directory_repo from src.repositories import elimination_rules as elimination_rules_repo from src.schemas.consolidations import ( EliminationMatchedPairOut, EliminationPairRowOut, EliminationPairsOut, EliminationTotalItemOut, EliminationTotalsOut, EliminationTotalStatus, EliminationUnpairedRowOut, IntercompanyMatrixCellOut, IntercompanyMatrixCompanyOut, IntercompanyMatrixOut, IntercompanyResultAmountOut, IntercompanyResultArticleOut, IntercompanyResultTableOut, UnpairedReason, ) 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 MatrixBuildResult: """Матрица и данные свода, использованные при её построении.""" matrix: IntercompanyMatrixOut version: ConsolidationVersion consolidation: Consolidation accounts: list[ChartOfAccount] selection: MatrixSelection rule: EliminationRule | None @dataclass(frozen=True) class VersionExportInfo: """Реквизиты версии в шапке и имени книги.""" group_name: str year: int quarter: PeriodQuarter version_no: int def _filename(self, prefix: str) -> str: return f"{prefix}_{self.group_name}_{self.year}-{self.quarter.value}_v{self.version_no}.xlsx" @dataclass(frozen=True) class MatrixExportInfo(VersionExportInfo): mode: Literal["ALL", "MANUAL", "PRESET"] debit_account_codes: tuple[str, ...] credit_account_codes: tuple[str, ...] measure: Measure = "closing" rule_id: uuid.UUID | None = None rule_no: int | None = None rule_version_no: int | None = None rule_statement_type: StatementType | None = None article_1_line: str | None = None article_2_line: str | None = None @property def preset_details(self) -> tuple[StatementType, int, int, str, str]: """Реквизиты правила §4 для имени и шапки XLSX шахматки §6.""" details = ( self.rule_statement_type, self.rule_no, self.rule_version_no, self.article_1_line, self.article_2_line, ) if any(value is None for value in details): raise ValueError("Для выгрузки PRESET нужны тип, номер, версия и статьи правила") statement_type, rule_no, rule_version_no, article_1_line, article_2_line = details assert statement_type is not None and rule_no is not None and rule_version_no is not None assert article_1_line is not None and article_2_line is not None return statement_type, rule_no, rule_version_no, article_1_line, article_2_line @property def filename(self) -> str: preset = "" if self.mode == "PRESET": statement_type, rule_no, _, _, _ = self.preset_details preset = f"_rule-{statement_type.value}-{rule_no}" return f"intercompany-matrix_{self.group_name}_{self.year}-{self.quarter.value}_v{self.version_no}{preset}.xlsx" @dataclass(frozen=True) class ResultTableExportInfo(VersionExportInfo): @property def filename(self) -> str: return self._filename("intercompany-result-table") @dataclass(frozen=True) class ResultTableBuild: table: IntercompanyResultTableOut version: ConsolidationVersion consolidation: Consolidation @dataclass(frozen=True) class PairContext: rule: EliminationRule company_a_id: uuid.UUID company_b_id: uuid.UUID @dataclass(frozen=True) class RulePacking: matched: list[EliminationMatchedPairOut] with_discrepancy: list[EliminationMatchedPairOut] unpaired: list[EliminationUnpairedRowOut] totals: RuleTotals def _get_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: """Суммы по парам компаний для шахматки (§6). Сторона 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"] += _get_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"] += _get_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=(amount_a == ZERO) != (amount_b == ZERO), 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 _make_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 _get_kinds(side: Side) -> frozenset[AccountKind]: return DEBIT_KINDS if side == "D" else CREDIT_KINDS def _select_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 _get_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 _get_kinds(side)) def calculate_result_table( rows: Iterable[ConsolidatedRow], member_companies: Sequence[IntercompanyMatrixCompanyOut], accounts: Sequence[ChartOfAccount], ) -> IntercompanyResultTableOut: """Фиксированные статьи ВГО: сальдо по всем контрагентам периметра для каждого ЮЛ (§6).""" by_id, children = _make_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: _select_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] += _get_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) def _pair_row_out(row: ConsolidatedRow) -> EliminationPairRowOut: """Строка свода, уже отфильтрованная по счёту и встречному ЮЛ периметра.""" assert row.counterparty_id is not None and row.account_id is not None and row.account_code is not None return EliminationPairRowOut( entry_ids=list(row.entry_ids), legal_entity_id=row.legal_entity_id, counterparty_id=row.counterparty_id, account_id=row.account_id, account_code=row.account_code, closing_debit=row.closing_debit, closing_credit=row.closing_credit, ) def _matched_pair_out( group: PackedGroup, context: PairContext, ) -> EliminationMatchedPairOut: rule = context.rule rows_a = group.debit_rows if rule.side_1 == "D" else group.credit_rows rows_b = group.credit_rows if rule.side_1 == "D" else group.debit_rows amount_a = group.debit_amount if rule.side_1 == "D" else group.credit_amount amount_b = group.credit_amount if rule.side_1 == "D" else group.debit_amount return EliminationMatchedPairOut( rule_id=rule.id, company_a_id=context.company_a_id, company_b_id=context.company_b_id, amount_a=amount_a, amount_b=amount_b, deviation_amount=group.deviation_amount, deviation_percent=group.deviation_percent, osv_entry_ids_a=[entry_id for row in rows_a for entry_id in row.key], osv_entry_ids_b=[entry_id for row in rows_b for entry_id in row.key], ) @dataclass(frozen=True) class RuleSides: """Встречные строки тройки: исходные строки свода и строки сторон с ненулевой суммой со знаком.""" sources: dict[RowKey, ConsolidatedRow] debit_rows: list[PackRow] credit_rows: list[PackRow] def a_b(self, rule: EliminationRule) -> tuple[list[PackRow], list[PackRow]]: """Строки стороны A (side_1 правила) и стороны B.""" if rule.side_1 == "D": return self.debit_rows, self.credit_rows return self.credit_rows, self.debit_rows def _collect_rule_sides(rows: Sequence[ConsolidatedRow], selection: MatrixSelection, context: PairContext) -> RuleSides: """Встречные строки направленной пары компаний по одному правилу; суммы со знаком, как в шахматке (§5).""" rule = context.rule company_a_id = context.company_a_id company_b_id = context.company_b_id debit_company_id = company_a_id if rule.side_1 == "D" else company_b_id credit_company_id = company_b_id if debit_company_id == company_a_id else company_a_id sources: dict[RowKey, ConsolidatedRow] = {} debit_rows: list[PackRow] = [] credit_rows: list[PackRow] = [] for row in rows: if row.account_id is None: continue in_debit = ( row.legal_entity_id == debit_company_id and row.counterparty_id == credit_company_id and row.account_id in selection.company_1_account_ids ) in_credit = ( row.legal_entity_id == credit_company_id and row.counterparty_id == debit_company_id and row.account_id in selection.company_2_account_ids ) if not in_debit and not in_credit: continue sources[row.entry_ids] = row assert row.counterparty_id is not None amount = _get_amount(row, "closing", "D" if in_debit else "K") if amount != ZERO: pack_row = PackRow(row.entry_ids, row.legal_entity_id, row.counterparty_id, amount) (debit_rows if in_debit else credit_rows).append(pack_row) return RuleSides(sources, debit_rows, credit_rows) def _rule_totals(sides: RuleSides, rule: EliminationRule) -> RuleTotals: """Итог тройки по строкам сторон: ΣA, ΣB, расхождение и сумма элиминации (§5 п. 1).""" rows_a, rows_b = sides.a_b(rule) amount_a = sum((row.amount for row in rows_a), ZERO) amount_b = sum((row.amount for row in rows_b), ZERO) return calculate_rule_totals(amount_a, amount_b, manual=rule.elimination_mode == "manual") def _total_status(totals: RuleTotals, rule: EliminationRule) -> EliminationTotalStatus: """Статус тройки в списке версии (§5 п. 3).""" if rule.elimination_mode == "manual": return "MANUAL" if ZERO in (totals.amount_a, totals.amount_b): return "ONE_SIDED" # без автоматической элиминации — расхождение: превышен порог либо меньшая сторона отрицательна return "AUTO" if totals.eliminated_amount > ZERO else "DISCREPANCY" def calculate_rule_total_items( rows: Sequence[ConsolidatedRow], rules: Sequence[tuple[EliminationRule, MatrixSelection]], company_ids: Sequence[uuid.UUID], ) -> list[EliminationTotalItemOut]: """Итоги всех троек версии: правило × упорядоченная пара компаний; тройки без строк пропускаются (§5 п. 3).""" items: list[EliminationTotalItemOut] = [] for rule, selection in rules: for company_a_id in company_ids: for company_b_id in company_ids: if company_a_id == company_b_id: continue sides = _collect_rule_sides(rows, selection, PairContext(rule, company_a_id, company_b_id)) if not sides.debit_rows and not sides.credit_rows: continue totals = _rule_totals(sides, rule) rows_a, rows_b = sides.a_b(rule) items.append( EliminationTotalItemOut( rule_id=rule.id, rule_no=rule.rule_no, elimination_mode=rule.elimination_mode, company_a_id=company_a_id, company_b_id=company_b_id, amount_a=totals.amount_a, amount_b=totals.amount_b, deviation_amount=totals.deviation_amount, deviation_percent=totals.deviation_percent, eliminated_amount=totals.eliminated_amount, exceeds_threshold=totals.exceeds_threshold, status=_total_status(totals, rule), osv_entry_ids_a=[entry_id for row in rows_a for entry_id in row.key], osv_entry_ids_b=[entry_id for row in rows_b for entry_id in row.key], ) ) return items def _pack_preset_rows( rows: Sequence[ConsolidatedRow], selection: MatrixSelection, context: PairContext, ) -> RulePacking: """Итог тройки и упаковка её встречных строк (§5 п. 1–2).""" sides = _collect_rule_sides(rows, selection, context) sources, debit_rows, credit_rows = sides.sources, sides.debit_rows, sides.credit_rows totals = _rule_totals(sides, context.rule) packed = pack_rule_rows(debit_rows, credit_rows) matched: list[EliminationMatchedPairOut] = [] with_discrepancy: list[EliminationMatchedPairOut] = [] for group in packed.groups: pair = _matched_pair_out(group, context) (matched if group.deviation_amount == ZERO else with_discrepancy).append(pair) unpaired: list[EliminationUnpairedRowOut] = [] amounts = {item.key: item.amount for item in [*debit_rows, *credit_rows]} for key, row in sources.items(): if key in packed.used_keys: continue amount = amounts.get(key, ZERO) reason: UnpairedReason if key in packed.ambiguous_keys: reason = "AMBIGUOUS" elif amount > ZERO: reason = "NO_MATCH" elif amount < ZERO: reason = "NEGATIVE_AMOUNT" else: reason = "ZERO_AMOUNT" unpaired.append(EliminationUnpairedRowOut(row=_pair_row_out(row), reason=reason)) return RulePacking(matched, with_discrepancy, unpaired, totals) 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_pairs( self, version_id: uuid.UUID, rule_id: uuid.UUID, company_a_id: uuid.UUID, company_b_id: uuid.UUID ) -> EliminationPairsOut: """Счета правила ББ и строки одной пары компаний без сохранения в БД (§4–§5).""" osv_result = await self._osv_service.build_with_members(version_id) version = osv_result.version member_ids = set(osv_result.members) if company_a_id == company_b_id or company_a_id not in member_ids or company_b_id not in member_ids: raise DomainValidationError("INVALID_COMPANY_PAIR", "Выберите две разные компании периметра версии") accounts = await directory_repo.list_accounts(self._session) by_id, children = _make_account_tree(accounts) selection, rule = await self._resolve_preset_selection(rule_id, version, by_id, children) if rule.measure != "closing": # шахматка по такому правилу считает обороты; пары §5 — только по сальдо raise DomainValidationError("INVALID_RULE", "Пары ВГО строятся только по правилам сальдо") # TODO(упаковка ОСВ): занятость строки отслеживать отдельно для каждого rule_id — одна строка может попасть # в результаты разных правил. Глобальное исключение повторного использования потребует изменения модели # данных: нужно хранить/передавать принадлежность исходной строки конкретной паре между правилами. packing = _pack_preset_rows( osv_result.consolidated.rows, selection, PairContext(rule, company_a_id, company_b_id), ) return EliminationPairsOut( version_no=rule.version_no, rule_id=rule.id, rule_no=rule.rule_no, company_a_id=company_a_id, company_b_id=company_b_id, amount_a=packing.totals.amount_a, amount_b=packing.totals.amount_b, deviation_amount=packing.totals.deviation_amount, deviation_percent=packing.totals.deviation_percent, eliminated_amount=packing.totals.eliminated_amount, exceeds_threshold=packing.totals.exceeds_threshold, matched=packing.matched, with_discrepancy=packing.with_discrepancy, unpaired=packing.unpaired, ) async def build_totals(self, version_id: uuid.UUID) -> EliminationTotalsOut: """Итоги всех троек версии по правилам ББ её версии правил; сводная ОСВ строится один раз (§5 п. 3).""" osv_result = await self._osv_service.build_with_members(version_id) version = osv_result.version accounts = await directory_repo.list_accounts(self._session) by_id, children = _make_account_tree(accounts) rules: list[tuple[EliminationRule, MatrixSelection]] = [] for rule in await elimination_rules_repo.list_rules(self._session, version.elimination_rules_version): # пары §5 — только правила ББ по сальдо; правила ОФР ещё не формализованы (§4) if rule.statement_type != StatementType.balance or rule.measure != "closing": continue selection, _ = await self._resolve_preset_selection(rule.id, version, by_id, children) rules.append((rule, selection)) items = calculate_rule_total_items(osv_result.consolidated.rows, rules, list(osv_result.members)) return EliminationTotalsOut(version_no=version.elimination_rules_version, items=items) 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: """Статичная таблица по пяти статьям""" return (await self._build_result_table(version_id)).table async def build_result_export( self, version_id: uuid.UUID ) -> tuple[IntercompanyResultTableOut, ResultTableExportInfo]: """Таблица и реквизиты книги из одного расчёта свода.""" built = await self._build_result_table(version_id) info = ResultTableExportInfo( group_name=await self._group_name(built.consolidation), year=built.consolidation.year, quarter=built.consolidation.quarter, version_no=built.version.version_no, ) return built.table, info async def _group_name(self, consolidation: Consolidation) -> str: return ( await self._session.execute( select(CompanyGroup.name).where(CompanyGroup.id == consolidation.company_group_id) ) ).scalar_one() async def _build_result_table(self, version_id: uuid.UUID) -> ResultTableBuild: result = 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 result.members.values()] table = calculate_result_table(result.consolidated.rows, companies, accounts) return ResultTableBuild(table, result.version, result.consolidation) async def build_accounts( self, version_id: uuid.UUID, *, mode: Literal["ALL", "MANUAL", "PRESET"], debit_account_ids: Sequence[uuid.UUID] = (), credit_account_ids: Sequence[uuid.UUID] = (), rule_id: uuid.UUID | None = None, ) -> IntercompanyMatrixOut: """Построить шахматку по выбранному режиму. ALL включает балансовые счета соответствующей стороны (без забалансовых); MANUAL — выбранные счета и их потомков; PRESET берёт счета и показатель из правила с идентификатором `rule_id`. """ result = await self._build_accounts( version_id, mode=mode, debit_account_ids=debit_account_ids, credit_account_ids=credit_account_ids, rule_id=rule_id, ) return result.matrix async def build_export( self, version_id: uuid.UUID, *, mode: Literal["ALL", "MANUAL", "PRESET"], debit_account_ids: Sequence[uuid.UUID] = (), credit_account_ids: Sequence[uuid.UUID] = (), rule_id: uuid.UUID | None = None, ) -> tuple[IntercompanyMatrixOut, MatrixExportInfo]: """Матрица и реквизиты книги из одного расчёта свода.""" result = await self._build_accounts( version_id, mode=mode, debit_account_ids=debit_account_ids, credit_account_ids=credit_account_ids, rule_id=rule_id, ) group_name = await self._group_name(result.consolidation) by_id = {account.id: account.code for account in result.accounts} if mode == "PRESET": debit_codes = tuple(sorted(by_id[account_id] for account_id in result.selection.company_1_account_ids)) credit_codes = tuple(sorted(by_id[account_id] for account_id in result.selection.company_2_account_ids)) else: debit_codes = tuple(by_id[account_id] for account_id in debit_account_ids) credit_codes = tuple(by_id[account_id] for account_id in credit_account_ids) info = MatrixExportInfo( group_name=group_name, year=result.consolidation.year, quarter=result.consolidation.quarter, version_no=result.version.version_no, mode=mode, debit_account_codes=debit_codes, credit_account_codes=credit_codes, measure=result.selection.measure, rule_id=result.rule.id if result.rule is not None else None, rule_no=result.rule.rule_no if result.rule is not None else None, rule_version_no=result.rule.version_no if result.rule is not None else None, rule_statement_type=result.rule.statement_type if result.rule is not None else None, article_1_line=result.rule.article_1_line if result.rule is not None else None, article_2_line=result.rule.article_2_line if result.rule is not None else None, ) return result.matrix, info async def _resolve_preset_selection( self, rule_id: uuid.UUID | None, version: ConsolidationVersion, by_id: dict[uuid.UUID, ChartOfAccount], children: dict[uuid.UUID, list[uuid.UUID]], ) -> tuple[MatrixSelection, EliminationRule]: """Выбрать стороны счетов и показатель правила §4 для шахматки §6.""" if rule_id is None: raise DomainValidationError("RULE_REQUIRED", "Выберите правило-пресет") rule, rule_accounts = await elimination_rules_repo.rule_with_account(self._session, rule_id) if rule is None or not elimination_rules_repo.is_valid_for_version(rule, version.elimination_rules_version): raise NotFoundError("правило-пресет не найдено для версии консолидации") if rule.statement_type != StatementType.balance: raise DomainValidationError("INVALID_RULE", "Шахматка PRESET поддерживает только правила ББ") if rule.side_1 not in ("D", "K") or rule.measure not in ("closing", "turnover"): raise DomainValidationError("INVALID_RULE", "Неверное направление или показатель правила-пресета") sides: dict[int, tuple[list[uuid.UUID], list[uuid.UUID]]] = { 1: ([], []), 2: ([], []), } for rule_account in rule_accounts: if rule_account.company_no not in sides: raise DomainValidationError("INVALID_RULE", "Неверная сторона счёта в правиле-пресете") included, excluded = sides[rule_account.company_no] (excluded if rule_account.is_excluded else included).append(rule_account.account_id) debit_side = 1 if rule.side_1 == "D" else 2 credit_side = 2 if debit_side == 1 else 1 debit_ids, debit_excluded = sides[debit_side] credit_ids, credit_excluded = sides[credit_side] if not debit_ids or not credit_ids: raise DomainValidationError("ACCOUNTS_REQUIRED", "В правиле-пресете нужны счета обеих сторон") debit = _select_accounts(debit_ids, debit_excluded, "D", by_id, children) credit = _select_accounts(credit_ids, credit_excluded, "K", by_id, children) measure: Measure = "closing" if rule.measure == "closing" else "turnover" return MatrixSelection(debit, credit, measure), rule async def _build_accounts( self, version_id: uuid.UUID, *, mode: Literal["ALL", "MANUAL", "PRESET"], debit_account_ids: Sequence[uuid.UUID], credit_account_ids: Sequence[uuid.UUID], rule_id: uuid.UUID | None = None, ) -> MatrixBuildResult: osv_result = await self._osv_service.build_with_members(version_id) rows = osv_result.consolidated.rows accounts = await directory_repo.list_accounts(self._session) by_id, children = _make_account_tree(accounts) rule: EliminationRule | None = None 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 not account.is_off_balance and account.kind in DEBIT_KINDS ) credit = frozenset( account.id for account in accounts if not account.is_off_balance and account.kind in CREDIT_KINDS ) selection = MatrixSelection(debit, credit, "closing") elif mode == "MANUAL": if not debit_account_ids or not credit_account_ids: raise DomainValidationError("ACCOUNTS_REQUIRED", "Выберите дебетовые и кредитовые счета") debit = _select_accounts(debit_account_ids, (), "D", by_id, children) credit = _select_accounts(credit_account_ids, (), "K", by_id, children) selection = MatrixSelection(debit, credit, "closing") elif mode == "PRESET": selection, rule = await self._resolve_preset_selection(rule_id, osv_result.version, by_id, children) else: raise DomainValidationError("INVALID_MODE", "Неизвестный режим выбора счетов") matrix = self._calculate(rows, osv_result.members, selection) return MatrixBuildResult( matrix=matrix, version=osv_result.version, consolidation=osv_result.consolidation, accounts=accounts, selection=selection, rule=rule, )