223 lines
10 KiB
Python
223 lines
10 KiB
Python
"""Граница с разнесением ОСВ: адаптация готового результа ``VgoService``."""
|
||
|
||
import uuid
|
||
from collections.abc import Sequence
|
||
from dataclasses import dataclass, field
|
||
from decimal import Decimal
|
||
from typing import Literal, Protocol, cast
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from src.common.exceptions import ConflictError
|
||
from src.domain.consolidated_osv import ConsolidatedRow
|
||
from src.domain.corrections import calculate_totals
|
||
from src.models.accounting import ChartOfAccount, EliminationRule, EliminationRuleAccount
|
||
from src.models.consolidation import ConsolidationVersion
|
||
from src.models.enums import StatementType
|
||
from src.schemas.consolidations import RefOut, VgoAssignmentGroupOut, VgoAssignmentRowOut
|
||
from src.schemas.corrections import (
|
||
AggregateSideOut,
|
||
CorrectionAggregateOut,
|
||
CorrectionEntryOut,
|
||
CorrectionPairOut,
|
||
CorrectionRuleOut,
|
||
)
|
||
from src.services.consolidated_osv import ConsolidatedOsvService
|
||
from src.services.vgo import VgoService, _manual_key_amount
|
||
|
||
ZERO = Decimal("0")
|
||
COMPANY_NUMBERS = (1, 2)
|
||
Side = Literal["D", "K"]
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CorrectionSourceTable:
|
||
rules: list[CorrectionRuleOut] = field(default_factory=list)
|
||
aggregates: list[CorrectionAggregateOut] = field(default_factory=list)
|
||
|
||
|
||
class CorrectionAggregateSource(Protocol):
|
||
async def table(self, version_id: uuid.UUID, statement_type: StatementType) -> CorrectionSourceTable: ...
|
||
|
||
|
||
@dataclass
|
||
class _RuleSpec:
|
||
rule: EliminationRule
|
||
included_codes: dict[int, list[str]] = field(default_factory=lambda: {1: [], 2: []})
|
||
excluded_codes: dict[int, list[str]] = field(default_factory=lambda: {1: [], 2: []})
|
||
|
||
|
||
def _direction(rule: EliminationRule, company_no: int) -> Side:
|
||
if rule.side_1 not in ("D", "K"):
|
||
raise ConflictError("CORRECTION_SOURCE_INVALID", "В правиле задана неизвестная сторона")
|
||
first = cast(Side, rule.side_1)
|
||
if company_no == 1:
|
||
return first
|
||
return "K" if first == "D" else "D"
|
||
|
||
|
||
class VgoCorrectionSource:
|
||
"""Преобразует готовое автоматическое и ручное разнесение в таблицу корректировок."""
|
||
|
||
def __init__(self, session: AsyncSession, vgo_service: VgoService, osv_service: ConsolidatedOsvService) -> None:
|
||
self._session = session
|
||
self._vgo_service = vgo_service
|
||
self._osv_service = osv_service
|
||
|
||
async def _rules(self, statement_type: StatementType, version_no: int) -> dict[uuid.UUID, _RuleSpec]:
|
||
accounts = list((await self._session.scalars(select(ChartOfAccount))).all())
|
||
by_id = {account.id: account for account in accounts}
|
||
rows = (
|
||
await self._session.execute(
|
||
select(EliminationRule, EliminationRuleAccount)
|
||
.outerjoin(EliminationRuleAccount, EliminationRuleAccount.rule_id == EliminationRule.id)
|
||
.where(
|
||
EliminationRule.statement_type == statement_type,
|
||
EliminationRule.version_no == version_no,
|
||
)
|
||
.order_by(EliminationRule.rule_no, EliminationRuleAccount.company_no)
|
||
)
|
||
).all()
|
||
specs: dict[uuid.UUID, _RuleSpec] = {}
|
||
for rule, account in rows:
|
||
spec = specs.setdefault(rule.id, _RuleSpec(rule))
|
||
if account is None or account.company_no not in COMPANY_NUMBERS:
|
||
continue
|
||
root = by_id.get(account.account_id)
|
||
if root is not None:
|
||
codes = spec.excluded_codes if account.is_excluded else spec.included_codes
|
||
codes[account.company_no].append(root.code)
|
||
return specs
|
||
|
||
async def table(self, version_id: uuid.UUID, statement_type: StatementType) -> CorrectionSourceTable:
|
||
version = await self._session.get(ConsolidationVersion, version_id)
|
||
if version is None or statement_type != StatementType.balance:
|
||
return CorrectionSourceTable()
|
||
specs = await self._rules(statement_type, version.elimination_rules_version)
|
||
allocation = await self._vgo_service.assignment_candidates(version_id)
|
||
osv_result = await self._osv_service.build_with_members(version_id)
|
||
aggregates = [
|
||
self._aggregate(group, specs[group.rule_id], osv_result.consolidated.rows) for group in allocation.groups
|
||
]
|
||
used_rules = {group.rule_id for group in allocation.groups}
|
||
rules = [self._rule_out(spec) for rule_id, spec in specs.items() if rule_id in used_rules]
|
||
return CorrectionSourceTable(rules=rules, aggregates=aggregates)
|
||
|
||
@staticmethod
|
||
def _rule_out(spec: _RuleSpec) -> CorrectionRuleOut:
|
||
rule = spec.rule
|
||
return CorrectionRuleOut(
|
||
id=rule.id,
|
||
rule_no=rule.rule_no,
|
||
article_1_line=rule.article_1_line,
|
||
article_2_line=rule.article_2_line,
|
||
side_1=_direction(rule, 1),
|
||
measure=cast(Literal["closing", "turnover"], rule.measure),
|
||
company_1_accounts=sorted(set(spec.included_codes[1])),
|
||
company_2_accounts=sorted(set(spec.included_codes[2])),
|
||
company_1_excluded=sorted(set(spec.excluded_codes[1])),
|
||
company_2_excluded=sorted(set(spec.excluded_codes[2])),
|
||
)
|
||
|
||
@staticmethod
|
||
def _aggregate(
|
||
group: VgoAssignmentGroupOut, spec: _RuleSpec, rows: Sequence[ConsolidatedRow]
|
||
) -> CorrectionAggregateOut:
|
||
rule = spec.rule
|
||
side_1_code = _direction(rule, 1)
|
||
rows_by_entry_id = {entry_id: row for row in rows for entry_id in row.entry_ids}
|
||
|
||
def amount(row: ConsolidatedRow | VgoAssignmentRowOut, side: Side) -> Decimal:
|
||
if rule.measure == "turnover":
|
||
return (row.turnover_debit if side == "D" else row.turnover_credit) or ZERO
|
||
return (row.closing_debit if side == "D" else row.closing_credit) or ZERO
|
||
|
||
def entry(row: ConsolidatedRow | VgoAssignmentRowOut, side: Side, value: Decimal) -> CorrectionEntryOut | None:
|
||
if row.counterparty_id is None:
|
||
return None
|
||
return CorrectionEntryOut(
|
||
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=value,
|
||
)
|
||
|
||
entries_by_key: dict[tuple[tuple[uuid.UUID, ...], Side], CorrectionEntryOut] = {}
|
||
calculated = (*group.matched, *group.with_discrepancy)
|
||
for pair in calculated:
|
||
for entry_ids, side_code in (
|
||
(pair.osv_entry_ids_a, side_1_code),
|
||
(pair.osv_entry_ids_b, _direction(rule, 2)),
|
||
):
|
||
selected: dict[tuple[uuid.UUID, ...], ConsolidatedRow] = {}
|
||
for entry_id in entry_ids:
|
||
auto_row = rows_by_entry_id.get(entry_id)
|
||
if auto_row is not None:
|
||
selected.setdefault(tuple(auto_row.entry_ids), auto_row)
|
||
for auto_row in selected.values():
|
||
item = entry(auto_row, side_code, amount(auto_row, side_code))
|
||
if item is not None:
|
||
entries_by_key[(tuple(auto_row.entry_ids), side_code)] = item
|
||
|
||
for manual_row in group.manual:
|
||
if manual_row.is_exclude:
|
||
continue
|
||
_, amount_1, amount_2 = _manual_key_amount(manual_row, rule) # type: ignore[arg-type]
|
||
side_code = side_1_code if amount_1 != ZERO or amount_2 == ZERO else _direction(rule, 2)
|
||
item = entry(manual_row, side_code, amount_1 if side_code == side_1_code else amount_2)
|
||
if item is not None:
|
||
entries_by_key[(tuple(manual_row.entry_ids), side_code)] = item
|
||
|
||
entries = list(entries_by_key.values())
|
||
|
||
def side(first: bool) -> AggregateSideOut:
|
||
side_entries = [entry for entry in entries if (entry.side == side_1_code) == first]
|
||
return AggregateSideOut(
|
||
article_line=rule.article_1_line if first else rule.article_2_line,
|
||
accounts=sorted({entry.account_code for entry in side_entries}),
|
||
amount=sum((entry.amount for entry in side_entries), ZERO),
|
||
entries=side_entries,
|
||
)
|
||
|
||
def pair_side(entry_ids: list[uuid.UUID], *, first: bool, amount: Decimal) -> AggregateSideOut:
|
||
ids = set(entry_ids)
|
||
pair_entries = [entry for entry in entries if ids.intersection(entry.entry_ids)]
|
||
return AggregateSideOut(
|
||
article_line=rule.article_1_line if first else rule.article_2_line,
|
||
accounts=sorted({entry.account_code for entry in pair_entries}),
|
||
amount=amount,
|
||
entries=pair_entries,
|
||
)
|
||
|
||
first, second = side(True), side(False)
|
||
first.amount = group.amount_a
|
||
second.amount = group.amount_b
|
||
pairs = [
|
||
CorrectionPairOut(
|
||
side_1=pair_side(pair.osv_entry_ids_a, first=True, amount=pair.amount_a),
|
||
side_2=pair_side(pair.osv_entry_ids_b, first=False, amount=pair.amount_b),
|
||
deviation_amount=pair.deviation_amount,
|
||
deviation_percent=pair.deviation_percent,
|
||
)
|
||
for pair in (*group.matched, *group.with_discrepancy)
|
||
]
|
||
totals = calculate_totals(first.amount, second.amount, manual=rule.elimination_mode == "manual")
|
||
return CorrectionAggregateOut(
|
||
rule_id=group.rule_id,
|
||
rule_no=group.rule_no,
|
||
company_1=RefOut(id=group.company_a_id, name=group.company_a_name or str(group.company_a_id)),
|
||
company_2=RefOut(id=group.company_b_id, name=group.company_b_name or str(group.company_b_id)),
|
||
side_1=first,
|
||
side_2=second,
|
||
pairs=pairs,
|
||
deviation_amount=totals.deviation_amount,
|
||
deviation_percent=totals.deviation_percent,
|
||
exceeds_threshold=totals.exceeds_threshold,
|
||
status=totals.status,
|
||
eliminated_amount=totals.eliminated_amount,
|
||
correction=None,
|
||
)
|