67 lines
3.3 KiB
Python
67 lines
3.3 KiB
Python
"""План полной замены применённой корректировки."""
|
||
|
||
import datetime
|
||
import uuid
|
||
from collections.abc import Sequence
|
||
from decimal import ROUND_HALF_UP, Decimal
|
||
|
||
from src.models.enums import CorrectionStatus, StatementType
|
||
from src.models.kfo import Correction
|
||
from src.schemas.corrections import CorrectionAggregateOut, CorrectionIn
|
||
|
||
|
||
def _optional_numeric(value: Decimal | None, precision: int, scale: int) -> Decimal | None:
|
||
"""Необязательные справочные поля старой модели не ограничивают сохранение.
|
||
|
||
Полные значения остаются в вычисляемом DTO и аудите. Не подменяем выход за
|
||
диапазон обрезкой.
|
||
"""
|
||
limit = Decimal(10) ** (precision - scale)
|
||
if value is None or abs(value) >= limit:
|
||
return None
|
||
rounded = value.quantize(Decimal(10) ** -scale, rounding=ROUND_HALF_UP)
|
||
return rounded if abs(rounded) < limit else None
|
||
|
||
|
||
def correction_rows( # noqa: PLR0913 - данные решения и автор сохранения
|
||
aggregate: CorrectionAggregateOut,
|
||
data: CorrectionIn,
|
||
old: Sequence[Correction],
|
||
*,
|
||
version_id: uuid.UUID,
|
||
statement_type: StatementType,
|
||
user_id: uuid.UUID,
|
||
) -> tuple[list[Correction], list[Correction]]:
|
||
now = datetime.datetime.now(datetime.UTC)
|
||
revision = (aggregate.correction.version_no if aggregate.correction else 0) + 1
|
||
roots: list[Correction] = []
|
||
children: list[Correction] = []
|
||
for debit, lines in ((True, data.debit), (False, data.credit)):
|
||
if not lines:
|
||
continue
|
||
owner = aggregate.company_1.id if debit else aggregate.company_2.id
|
||
other = aggregate.company_2.id if debit else aggregate.company_1.id
|
||
previous = next((r for r in old if r.parent_correction_id is None and r.legal_entity_id == owner), None)
|
||
root_id = previous.id if previous else uuid.uuid4()
|
||
for index, line in enumerate(lines):
|
||
row = previous if index == 0 and previous else Correction(id=root_id if index == 0 else uuid.uuid4())
|
||
row.consolidation_version_id = version_id
|
||
row.statement_type = statement_type
|
||
row.rule_id = aggregate.rule_id
|
||
row.parent_correction_id = None if index == 0 else root_id
|
||
row.legal_entity_id, row.counterparty_id = owner, other
|
||
row.debit_line_code = line.line_code if debit else None
|
||
row.credit_line_code = None if debit else line.line_code
|
||
row.debit_adjustment = Decimal(line.amount) if debit else None
|
||
row.credit_adjustment = None if debit else Decimal(line.amount)
|
||
row.debit_account_id = row.credit_account_id = None
|
||
row.comment, row.reason_code = data.comment.strip(), data.reason_code
|
||
row.deviation_amount = _optional_numeric(aggregate.deviation_amount, 18, 2)
|
||
row.deviation_percent = _optional_numeric(aggregate.deviation_percent, 9, 4)
|
||
row.status, row.version_no = CorrectionStatus.APPLIED, revision
|
||
if row.created_at is None:
|
||
row.created_by, row.created_at = user_id, now
|
||
row.updated_by, row.updated_at = user_id, now
|
||
(roots if index == 0 else children).append(row)
|
||
return roots, children
|