"""Подготовка строк сводной ОСВ для распределения ВГО по правилам.""" import uuid from collections import defaultdict from collections.abc import Mapping, Sequence, Set from dataclasses import dataclass, field from decimal import Decimal from typing import Literal, cast from sqlalchemy.ext.asyncio import AsyncSession from src.api.dependencies.audit import RequestAudit from src.common.clock import utc_now from src.common.exceptions import DomainValidationError from src.domain.consolidated_osv import ConsolidatedRow from src.domain.elimination_packing import calculate_rule_totals from src.domain.policy import VisibilityScope from src.domain.vgo_distribution import ( Distribution, DistributionInput, account_owners, distribute_vgo_rows, select_exact_aggregates, ) from src.models.accounting import EliminationRule from src.models.enums import AuditAction, AuditSection, StatementType from src.models.kfo import VgoAssignment from src.repositories import directories as directories_repo from src.repositories import elimination_rules as elimination_rules_repo from src.repositories import vgo_assignments as vgo_assignments_repo from src.schemas.consolidations import ( EliminationMatchedPairOut, VgoAssignmentAttachRuleIn, VgoAssignmentCandidatesOut, VgoAssignmentDetachRuleIn, VgoAssignmentExcludeIn, VgoAssignmentGroupOut, VgoAssignmentOut, VgoAssignmentPositionOut, VgoAssignmentRowOut, VgoUnallocatedPositionOut, ) from src.services.consolidated_osv import ConsolidatedOsvBuildResult, ConsolidatedOsvService from src.services.consolidation_guard import lock_editable_version from src.services.intercompany_matrix import IntercompanyMatrixService, _make_account_tree ZERO = Decimal("0") GroupKey = tuple[uuid.UUID, uuid.UUID, uuid.UUID] ManualItem = tuple[VgoAssignmentRowOut, Decimal, Decimal] def assignment_candidates(rows: Sequence[ConsolidatedRow]) -> list[ConsolidatedRow]: """Оставить строки с контрагентом и ненулевым оборотом или конечным остатком. Все четыре суммы проверяются вместе; `None` равнозначен нулю. Начальный остаток не учитывается. """ candidates = [] for row in rows: if row.counterparty_id is None: continue amounts = (row.turnover_debit, row.turnover_credit, row.closing_debit, row.closing_credit) for amount in amounts: if amount not in (None, ZERO): candidates.append(row) break return candidates def _ids_from_aggregate(aggregate: EliminationMatchedPairOut) -> set[uuid.UUID]: return set(aggregate.osv_entry_ids_a) | set(aggregate.osv_entry_ids_b) def _assignment_row(row: ConsolidatedRow, assignment: VgoAssignment | None = None) -> VgoAssignmentRowOut: output = VgoAssignmentRowOut( entry_ids=list(row.entry_ids), legal_entity_id=row.legal_entity_id, legal_entity_name=row.legal_entity_name, counterparty_id=row.counterparty_id, counterparty_name=row.counterparty_entity_name or row.counterparty_name, account_code=row.account_code, period=row.period, opening_debit=row.opening_debit, opening_credit=row.opening_credit, turnover_debit=row.turnover_debit, turnover_credit=row.turnover_credit, closing_debit=row.closing_debit, closing_credit=row.closing_credit, ) if assignment is not None: output.action = "ASSIGN" if assignment.action == "ASSIGN" else "UNASSIGNED" output.is_exclude = assignment.is_exclude return output def _manual_key_amount(row: ConsolidatedRow, rule: EliminationRule) -> tuple[GroupKey, Decimal, Decimal]: """Определить сторону ручной строки по сумме и стороне A правила.""" if rule.measure == "turnover": debit, credit = row.turnover_debit or ZERO, row.turnover_credit or ZERO else: debit, credit = row.closing_debit or ZERO, row.closing_credit or ZERO first, second = (debit, credit) if rule.side_1 == "D" else (credit, debit) assert row.counterparty_id is not None if first != ZERO or second == ZERO: return (rule.id, row.legal_entity_id, row.counterparty_id), first, ZERO return (rule.id, row.counterparty_id, row.legal_entity_id), ZERO, second @dataclass class _ManualResult: """Ручные решения по строкам свода и группам правил.""" groups: dict[GroupKey, list[ManualItem]] by_row: dict[tuple[uuid.UUID, ...], VgoAssignment] assigned_ids: set[uuid.UUID] partial_rows: set[tuple[uuid.UUID, ...]] def _manual_assignments( rows: Sequence[ConsolidatedRow], assignments: Sequence[VgoAssignment], rules: Sequence[EliminationRule] ) -> _ManualResult: by_entry = {assignment.osv_entry_id: assignment for assignment in assignments} rules_by_id = {rule.id: rule for rule in rules} groups: defaultdict[GroupKey, list[ManualItem]] = defaultdict(list) by_row: dict[tuple[uuid.UUID, ...], VgoAssignment] = {} assigned_ids: set[uuid.UUID] = set() partial_rows: set[tuple[uuid.UUID, ...]] = set() for row in rows: found = [by_entry[entry_id] for entry_id in row.entry_ids if entry_id in by_entry] if not found: continue if len(found) != len(row.entry_ids): by_row[row.entry_ids] = found[0] partial_rows.add(row.entry_ids) continue assignment = found[0] if any( (item.action, item.rule_id, item.is_exclude) != (assignment.action, assignment.rule_id, assignment.is_exclude) for item in found[1:] ): raise DomainValidationError( "CONFLICTING_VGO_ASSIGNMENT", "У исходных строк одной сводной ОСВ разные ручные назначения", ) by_row[row.entry_ids] = assignment rule = rules_by_id.get(assignment.rule_id) if assignment.rule_id is not None else None if assignment.action == "ASSIGN" and rule is not None and row.counterparty_id is not None: key, amount_a, amount_b = _manual_key_amount(row, rule) groups[key].append((_assignment_row(row, assignment), amount_a, amount_b)) assigned_ids.update(row.entry_ids) return _ManualResult(groups, by_row, assigned_ids, partial_rows) def _assign_unpaired( rows: Sequence[ConsolidatedRow], manual: _ManualResult, auto_rules: Mapping[tuple[uuid.UUID, ...], uuid.UUID], rules: Mapping[uuid.UUID, EliminationRule], ) -> None: """Отнести оставшиеся без агрегата строки к единственному правилу их счёта.""" for row in rows: rule_id = auto_rules.get(row.entry_ids) if rule_id is None: continue rule = rules[rule_id] key, amount_a, amount_b = _manual_key_amount(row, rule) output = _assignment_row(row) output.action = "ASSIGN" output.is_exclude = False output.is_auto_assigned = True manual.groups.setdefault(key, []).append((output, amount_a, amount_b)) manual.assigned_ids.update(row.entry_ids) def _unassigned_rows( rows: Sequence[ConsolidatedRow], manual: _ManualResult, distribution: Distribution, ) -> list[VgoAssignmentRowOut]: result = [] for row in rows: if row.entry_ids not in distribution.unassigned or not manual.assigned_ids.isdisjoint(row.entry_ids): continue if row.entry_ids in manual.partial_rows: item = _assignment_row(row) item.unassigned_reason = "PARTIAL_MANUAL_ASSIGNMENT" else: item = _assignment_row(row, manual.by_row.get(row.entry_ids)) result.append(item) return result @dataclass class _AggregateSearchResult: """Принятые агрегаты, кандидаты с расхождением и занятые строки.""" matched_aggregates: defaultdict[GroupKey, list[EliminationMatchedPairOut]] = field( default_factory=lambda: defaultdict(list) ) diff_aggregates: list[EliminationMatchedPairOut] = field(default_factory=list) used_ids: set[uuid.UUID] = field(default_factory=set) def _select_diff_aggregates( search: _AggregateSearchResult, distribution: Distribution ) -> dict[GroupKey, list[EliminationMatchedPairOut]]: """Сгруппировать выбранные доменной функцией расхождения.""" diff_aggregates: defaultdict[GroupKey, list[EliminationMatchedPairOut]] = defaultdict(list) for index in distribution.selected_diff: aggregate = search.diff_aggregates[index] key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id) diff_aggregates[key].append(aggregate) return diff_aggregates def _position( row: ConsolidatedRow | VgoAssignmentRowOut, side: Literal["D", "K"], amount: Decimal, *, action: Literal["ASSIGN"] | None = None, candidate_rule_ids: Sequence[uuid.UUID] = (), ) -> VgoAssignmentPositionOut | None: if row.counterparty_id is None: return None return VgoAssignmentPositionOut( entry_ids=list(row.entry_ids), legal_entity_id=row.legal_entity_id, counterparty_id=row.counterparty_id, account_code=row.account_code or "", side=side, amount=amount, action=action, candidate_rule_ids=list(candidate_rule_ids), ) def _build_groups( # noqa: PLR0913, PLR0917 - результат поиска и справочники передаются раздельно search: _AggregateSearchResult, diff_aggregates: dict[GroupKey, list[EliminationMatchedPairOut]], rules: Sequence[EliminationRule], manual: dict[GroupKey, list[ManualItem]], names: Mapping[uuid.UUID, str], rows: Sequence[ConsolidatedRow], owners: Mapping[uuid.UUID, Set[uuid.UUID]], ) -> list[VgoAssignmentGroupOut]: """Сгруппировать принятые агрегаты и посчитать суммы по правилу и компаниям.""" rule_by_id = {rule.id: rule for rule in rules} rows_by_entry_id = {entry_id: row for row in rows for entry_id in row.entry_ids} keys = list(dict.fromkeys([*search.matched_aggregates, *diff_aggregates, *manual])) groups = [] for rule_id, company_a_id, company_b_id in keys: key = (rule_id, company_a_id, company_b_id) matched = search.matched_aggregates.get(key, []) with_discrepancy = diff_aggregates.get(key, []) aggregates = [*matched, *with_discrepancy] manual_rows = manual.get(key, []) rule = rule_by_id[rule_id] if rule.side_1 not in ("D", "K"): raise DomainValidationError("INVALID_VGO_RULE_SIDE", "У правила задана неизвестная сторона") side_a = cast(Literal["D", "K"], rule.side_1) side_b: Literal["D", "K"] = "K" if side_a == "D" else "D" positions: dict[tuple[tuple[uuid.UUID, ...], str], VgoAssignmentPositionOut] = {} for aggregate in aggregates: for entry_ids, side in ((aggregate.osv_entry_ids_a, side_a), (aggregate.osv_entry_ids_b, side_b)): selected = { tuple(rows_by_entry_id[entry_id].entry_ids): rows_by_entry_id[entry_id] for entry_id in entry_ids } for source_position in selected.values(): if rule.measure == "turnover": amount = ( source_position.turnover_debit if side == "D" else source_position.turnover_credit ) or ZERO else: amount = ( source_position.closing_debit if side == "D" else source_position.closing_credit ) or ZERO candidate_ids = ( owners.get(source_position.account_id, frozenset()) if source_position.account_id is not None else frozenset() ) position = _position( source_position, side, amount, candidate_rule_ids=sorted( candidate_ids - {rule_id}, key=lambda item: (rule_by_id[item].rule_no, str(item)), ), ) if position is not None: positions[(tuple(position.entry_ids), position.side)] = position for manual_row, amount_a, amount_b in manual_rows: side = side_a if amount_a != ZERO or amount_b == ZERO else side_b source_row = rows_by_entry_id.get(manual_row.entry_ids[0]) if manual_row.entry_ids else None candidate_ids = ( owners.get(source_row.account_id, frozenset()) if source_row is not None and source_row.account_id is not None else frozenset() ) position = _position( manual_row, side, amount_a if side == side_a else amount_b, action=None if manual_row.is_auto_assigned else "ASSIGN", candidate_rule_ids=sorted( candidate_ids - {rule_id}, key=lambda item: (rule_by_id[item].rule_no, str(item)), ), ) if position is not None: positions[(tuple(position.entry_ids), position.side)] = position amount_a = sum((aggregate.amount_a for aggregate in aggregates), ZERO) amount_b = sum((aggregate.amount_b for aggregate in aggregates), ZERO) amount_a += sum((item_amount_a for _, item_amount_a, _ in manual_rows), ZERO) amount_b += sum((item_amount_b for _, _, item_amount_b in manual_rows), ZERO) totals = calculate_rule_totals(amount_a, amount_b, manual=False) groups.append( VgoAssignmentGroupOut( rule_id=rule_id, rule_no=rule_by_id[rule_id].rule_no, article_1_line=rule_by_id[rule_id].article_1_line, article_2_line=rule_by_id[rule_id].article_2_line, measure=rule_by_id[rule_id].measure, company_a_id=company_a_id, company_a_name=names.get(company_a_id), company_b_id=company_b_id, company_b_name=names.get(company_b_id), amount_a=amount_a, amount_b=amount_b, deviation_amount=totals.deviation_amount, deviation_percent=totals.deviation_percent, matched=matched, with_discrepancy=with_discrepancy, manual=[row for row, _, _ in manual_rows], positions=list(positions.values()), ) ) return groups def _unallocated_positions( rows: Sequence[VgoAssignmentRowOut], source_rows: Sequence[ConsolidatedRow], owners: Mapping[uuid.UUID, Set[uuid.UUID]], rule_order: Mapping[uuid.UUID, int], ) -> list[VgoUnallocatedPositionOut]: """Дополнить результат разнесения представлением нераспределённых позиций.""" source_by_key = {row.entry_ids: row for row in source_rows} result = [] for row in rows: source = source_by_key.get(tuple(row.entry_ids)) if source is None or row.counterparty_id is None: continue debit, credit = source.closing_debit or ZERO, source.closing_credit or ZERO side: Literal["D", "K"] = "D" if debit != ZERO or credit == ZERO else "K" candidate_ids = owners.get(source.account_id, frozenset()) if source.account_id is not None else frozenset() reason: Literal["MULTIPLE_RULES", "UNASSIGNED"] = ( "UNASSIGNED" if row.action == "UNASSIGNED" or row.unassigned_reason is not None else "MULTIPLE_RULES" ) result.append( VgoUnallocatedPositionOut( entry_ids=row.entry_ids, legal_entity_id=row.legal_entity_id, legal_entity_name=row.legal_entity_name, counterparty_id=row.counterparty_id, counterparty_name=row.counterparty_name or str(row.counterparty_id), account_code=row.account_code or "", side=side, amount=debit if side == "D" else credit, candidate_rule_ids=sorted(candidate_ids, key=lambda item: (rule_order.get(item, 0), str(item))), reason=reason, excluded=row.is_exclude, ) ) return result class VgoService: """Операции ВГО над сводной ОСВ версии консолидации.""" def __init__( self, session: AsyncSession, osv_service: ConsolidatedOsvService, matrix_service: IntercompanyMatrixService ) -> None: self._osv_service = osv_service self._matrix_service = matrix_service self.session = session async def _rule_accounts( self, osv_result: ConsolidatedOsvBuildResult, rules: Sequence[EliminationRule] ) -> dict[uuid.UUID, frozenset[uuid.UUID]]: """Раскрыть дочерние счета и определить их принадлежность правилам.""" accounts = await directories_repo.list_accounts(self.session) by_id, children = _make_account_tree(accounts) selections = [] for rule in rules: selection, _ = await self._matrix_service._resolve_preset_selection( rule.id, osv_result.version, by_id, children ) selections.append((rule.id, selection.company_1_account_ids | selection.company_2_account_ids)) return account_owners(selections) async def _find_aggregates( self, version_id: uuid.UUID, osv_result: ConsolidatedOsvBuildResult, rules: Sequence[EliminationRule], rows: Sequence[ConsolidatedRow], ) -> _AggregateSearchResult: """Перебрать правила и компании, сразу закрепляя только точные агрегаты.""" search = _AggregateSearchResult() for rule in rules: for company_a in osv_result.members: for company_b in osv_result.members: if company_a == company_b: continue free_rows = [row for row in rows if search.used_ids.isdisjoint(row.entry_ids)] result = await self._matrix_service.build_pairs( version_id, rule.id, company_a, company_b, free_rows, osv_result=osv_result ) accepted, used_ids = select_exact_aggregates( search.used_ids, [_ids_from_aggregate(aggregate) for aggregate in result.matched] ) search.used_ids.update(used_ids) for index in accepted: aggregate = result.matched[index] key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id) search.matched_aggregates[key].append(aggregate) search.diff_aggregates.extend(result.with_discrepancy) return search async def assignment_candidates(self, version_id: uuid.UUID) -> VgoAssignmentCandidatesOut: """Найти агрегаты и общие для всех правил строки без назначения, не записывая их в БД.""" osv_result = await self._osv_service.build_with_members(version_id) all_rules = await elimination_rules_repo.list_rules(self.session, osv_result.version.elimination_rules_version) rules = sorted( (rule for rule in all_rules if rule.statement_type == StatementType.balance and rule.measure == "closing"), key=lambda rule: rule.rule_no, ) rows = assignment_candidates(osv_result.consolidated.rows) assignments = await vgo_assignments_repo.list_assignments(self.session, version_id, StatementType.balance) manual = _manual_assignments(osv_result.consolidated.rows, assignments, rules) free_rows = [row for row in rows if row.entry_ids not in manual.by_row] owners = await self._rule_accounts(osv_result, rules) if any(row.account_id is not None for row in rows) else {} search = await self._find_aggregates(version_id, osv_result, rules, free_rows) candidate_keys = {row.entry_ids for row in rows} decision_rows = [*rows] decision_rows.extend( row for row in osv_result.consolidated.rows if row.entry_ids not in candidate_keys and row.entry_ids in manual.by_row ) distribution = distribute_vgo_rows( DistributionInput( rows=decision_rows, exact_ids=search.used_ids, diff_ids=[_ids_from_aggregate(aggregate) for aggregate in search.diff_aggregates], manual_keys=set(manual.by_row), owners=owners, members=set(osv_result.members), ) ) diff_aggregates = _select_diff_aggregates(search, distribution) _assign_unpaired(free_rows, manual, distribution.auto_rules, {rule.id: rule for rule in rules}) names = {company_id: entity.name for company_id, entity in osv_result.members.items()} for row in osv_result.consolidated.rows: if row.counterparty_id is not None: name = row.counterparty_name or row.counterparty_entity_name if name is not None: names.setdefault(row.counterparty_id, name) groups = _build_groups( search, diff_aggregates, rules, manual.groups, names, osv_result.consolidated.rows, owners ) unassigned = _unassigned_rows(decision_rows, manual, distribution) unallocated = _unallocated_positions( unassigned, osv_result.consolidated.rows, owners, {rule.id: rule.rule_no for rule in rules}, ) return VgoAssignmentCandidatesOut(groups=groups, unassigned=unassigned, unallocated=unallocated) class VgoAssignmentCrudService: """Ручная правка назначений исходных строк ОСВ одной версии.""" def __init__( self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, osv_service: ConsolidatedOsvService, ) -> None: self.session = session self.audit = audit self.scope = scope self.osv_service = osv_service async def _check_rows(self, version_id: uuid.UUID, entry_ids: list[uuid.UUID]) -> None: """Проверить, что ID образуют одну или несколько полных строк свода.""" osv_result = await self.osv_service.build_with_members(version_id) requested = set(entry_ids) matched_rows = [row for row in osv_result.consolidated.rows if requested.intersection(row.entry_ids)] matched_ids = {entry_id for row in matched_rows for entry_id in row.entry_ids} if requested != matched_ids: raise DomainValidationError( "INVALID_OSV_ENTRY_IDS", "entry_ids должны содержать все исходные ID выбранных строк сводной ОСВ" ) async def attach_rule(self, version_id: uuid.UUID, data: VgoAssignmentAttachRuleIn) -> list[VgoAssignmentOut]: """Привязать выбранные полные строки сводной ОСВ версии к правилу.""" version, _ = await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id) rule = await self.session.get(EliminationRule, data.rule_id) if rule is None or rule.version_no != version.elimination_rules_version: raise DomainValidationError("INVALID_VGO_RULE", "правило не относится к версии консолидации") await self._check_rows(version_id, data.entry_ids) existing = await vgo_assignments_repo.for_entries(self.session, version_id, rule.statement_type, data.entry_ids) return await self._persist_rule(version_id, rule, data.entry_ids, existing) async def set_exclusion(self, version_id: uuid.UUID, data: VgoAssignmentExcludeIn) -> list[VgoAssignmentOut]: """Установить признак исключения, не меняя назначенное правило.""" await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id) await self._check_rows(version_id, data.entry_ids) existing = await vgo_assignments_repo.for_entries(self.session, version_id, data.statement_type, data.entry_ids) if data.is_exclude and any(item.rule_id is not None for item in existing.values()): raise DomainValidationError("VGO_RULE_ASSIGNED", "строку с правилом нельзя исключить из КФО") before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()} result = [] for entry_id in data.entry_ids: item = existing.get(entry_id) if item is None: item = VgoAssignment( consolidation_version_id=version_id, statement_type=data.statement_type, osv_entry_id=entry_id, action="UNASSIGNED", is_exclude=data.is_exclude, created_by=self.audit.user_id, ) self.session.add(item) else: item.is_exclude = data.is_exclude item.updated_by = self.audit.user_id item.updated_at = utc_now() result.append(item) await self.session.flush() await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result}) return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result] async def detach_rule(self, version_id: uuid.UUID, data: VgoAssignmentDetachRuleIn) -> list[VgoAssignmentOut]: """Снять правило и создать отсутствующие записи как исключённые.""" await lock_editable_version(self.session, self.scope, version_id, self.audit.user_id) await self._check_rows(version_id, data.entry_ids) existing = await vgo_assignments_repo.for_entries(self.session, version_id, data.statement_type, data.entry_ids) before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()} result = [] for entry_id in data.entry_ids: item = existing.get(entry_id) if item is None: item = VgoAssignment( consolidation_version_id=version_id, statement_type=data.statement_type, osv_entry_id=entry_id, created_by=self.audit.user_id, ) self.session.add(item) else: item.updated_by = self.audit.user_id item.updated_at = utc_now() item.action = "UNASSIGNED" item.rule_id = None item.is_exclude = True result.append(item) await self.session.flush() await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result}) return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result] async def _persist_rule( self, version_id: uuid.UUID, rule: EliminationRule, entry_ids: list[uuid.UUID], existing: dict[uuid.UUID, VgoAssignment], ) -> list[VgoAssignmentOut]: """Сохранить привязку к правилу и аудит для проверенных ID.""" before = {str(entry_id): _assignment_snapshot(item) for entry_id, item in existing.items()} result = [] for entry_id in entry_ids: item = existing.get(entry_id) if item is None: item = VgoAssignment( consolidation_version_id=version_id, statement_type=rule.statement_type, osv_entry_id=entry_id, created_by=self.audit.user_id, ) self.session.add(item) else: item.updated_by = self.audit.user_id item.updated_at = utc_now() item.action = "ASSIGN" item.rule_id = rule.id item.is_exclude = False result.append(item) await self.session.flush() await self._audit(version_id, before, {str(item.osv_entry_id): _assignment_snapshot(item) for item in result}) return [VgoAssignmentOut.model_validate(item, from_attributes=True) for item in result] async def _audit(self, version_id: uuid.UUID, before: Mapping[str, object], after: Mapping[str, object]) -> None: await self.audit.record( action=AuditAction.ASSIGNMENT_CHANGE, section=AuditSection.KFO, object_type="vgo_assignment", object_id=version_id, consolidation_version_id=version_id, before_value={"entries": before}, after_value={"entries": after}, ) def _assignment_snapshot(item: VgoAssignment) -> dict[str, object]: return {"action": item.action, "rule_id": item.rule_id, "is_exclude": item.is_exclude}