arm-core/api/src/services/goodwill.py

"""Таблица гудвилла, пакетное сохранение решений и переключатель — в транзакции запроса (§8, DOC-259025)."""

import re
from decimal import Decimal
from http import HTTPStatus
from uuid import UUID, uuid4

from fastapi.concurrency import run_in_threadpool
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.api.dependencies.audit import RequestAudit
from src.api.dependencies.auth import Principal
from src.common.clock import utc_now
from src.common.exceptions import ConflictError, DomainError, DomainValidationError
from src.domain import goodwill as domain
from src.domain.policy import VisibilityScope, ensure_version_editable
from src.models.access import AppUser
from src.models.directories import LegalEntity
from src.models.enums import AuditAction, AuditSection
from src.models.kfo import Goodwill
from src.repositories import consolidations as versions
from src.repositories import goodwill as repo
from src.schemas.consolidations import RefOut
from src.schemas.goodwill import (
    GoodwillGroupOut,
    GoodwillRowIn,
    GoodwillRowKeyIn,
    GoodwillRowOut,
    GoodwillSign,
    GoodwillTableOut,
    GoodwillTotalsOut,
    WarningOut,
)
from src.services.consolidation_guard import lock_editable_version
from src.services.goodwill_export import build_workbook, matches_sign
from src.services.goodwill_source import GoodwillSource

MAX_SHARE = Decimal(100)

MESSAGES = {
    "CAPITAL_SOURCE_MISSING": "Нет однозначного значения 1310 на конец отчётного периода",
    "CAPITAL_SOURCE_AMBIGUOUS": "Несколько подходящих балансов дочерней организации",
    "OWNERSHIP_TOTAL_EXCEEDED": "Сумма долей в дочерней организации превышает 100%",
}


def validate_decision(data: GoodwillRowIn) -> domain.Decision:
    value = data.ownership_share.replace(",", ".") if data.ownership_share is not None else None
    if value is None:
        share = None
    elif not re.fullmatch(r"\d{1,3}(?:\.\d{1,2})?", value) or Decimal(value) > MAX_SHARE:
        raise DomainValidationError("INVALID_SHARE", "Доля — число от 0 до 100, не больше двух знаков после запятой")
    else:
        share = Decimal(value)
    if data.include_in_kfo and share is None:
        raise DomainValidationError("INVALID_SHARE", "Заполните долю владения")
    comment = data.comment.strip() if data.comment is not None else None
    if data.include_in_kfo and not comment:
        raise DomainValidationError("COMMENT_REQUIRED", "Укажите комментарий")
    return domain.Decision(share, data.include_in_kfo, comment)


class GoodwillService:
    def __init__(self, session: AsyncSession, scope: VisibilityScope, principal: Principal, source: GoodwillSource):
        self.session, self.scope, self.principal, self.source = session, scope, principal, source

    async def table(self, version_id: UUID) -> GoodwillTableOut:
        sources = await self.source.load(version_id)
        ctx = sources.context
        saved = await repo.saved(self.session, version_id)
        decisions = {
            (r.head_legal_entity_id, r.subsidiary_legal_entity_id): domain.Decision(
                r.ownership_share, r.include_in_kfo, r.comment
            )
            for r in saved
        }
        rows = domain.build(sources.investments, decisions, sources.capital)
        old = {(r.head_legal_entity_id, r.subsidiary_legal_entity_id): r for r in saved}
        ids = {i for r in rows for i in (r.head_id, r.subsidiary_id)}
        entities = {
            e.id: e for e in (await self.session.scalars(select(LegalEntity).where(LegalEntity.id.in_(ids)))).all()
        }
        users = {
            u.id: u
            for u in (
                await self.session.scalars(
                    select(AppUser).where(AppUser.id.in_({r.updated_by for r in saved if r.updated_by}))
                )
            ).all()
        }
        reason = None
        if "CONSOLIDATION_EDIT" not in self.principal.permissions:
            reason = "VERSION_NOT_EDITABLE"
        else:
            try:
                ensure_version_editable(
                    ctx.version, await versions.active_lock(self.session, version_id), self.principal.user.id
                )
            except ConflictError as error:
                if error.code not in ("VERSION_NOT_EDITABLE", "LOCK_REQUIRED", "LOCKED_BY_OTHER"):
                    raise
                reason = error.code
        output = []
        for r in rows:
            previous = old.get((r.head_id, r.subsidiary_id))
            user = users.get(previous.updated_by) if previous and previous.updated_by else None
            codes = list(r.warnings)
            if r.has_investments and r.subsidiary_id in sources.warnings:
                codes = [c for c in codes if c != "CAPITAL_SOURCE_MISSING"] + [sources.warnings[r.subsidiary_id]]
            output.append(
                GoodwillRowOut(
                    head_legal_entity=RefOut(id=r.head_id, name=entities[r.head_id].name),
                    subsidiary_legal_entity=RefOut(id=r.subsidiary_id, name=entities[r.subsidiary_id].name),
                    ownership_share=r.decision.ownership_share,
                    financial_investment_amount=r.financial_investment_amount,
                    charter_capital_amount=r.charter_capital_amount,
                    additional_capital_amount=sources.additional.get(r.subsidiary_id) if r.has_investments else None,
                    participation_amount=r.participation_amount,
                    goodwill_amount=r.goodwill_amount,
                    include_in_kfo=r.decision.include_in_kfo,
                    comment=r.decision.comment,
                    has_investments=r.has_investments,
                    warnings=[WarningOut(code=c, message=MESSAGES[c]) for c in codes],
                    updated_by=RefOut(id=user.id, name=user.full_name) if user else None,
                    updated_at=previous.updated_at if previous else None,
                )
            )
        output.sort(
            key=lambda r: (
                r.head_legal_entity.name,
                r.subsidiary_legal_entity.name,
                r.head_legal_entity.id,
                r.subsidiary_legal_entity.id,
            )
        )
        groups = domain.group_totals(rows, ctx.version.goodwill_enabled)
        heads = sorted(groups, key=lambda head: (entities[head].name, head))
        return GoodwillTableOut(
            version_id=version_id,
            goodwill_enabled=ctx.version.goodwill_enabled,
            editable=reason is None,
            read_only_reason=reason,
            rows=output,
            groups=[
                GoodwillGroupOut(
                    head_legal_entity=RefOut(id=head, name=entities[head].name),
                    totals=GoodwillTotalsOut(**groups[head]),
                )
                for head in heads
            ],
            totals=GoodwillTotalsOut(**domain.totals(rows, ctx.version.goodwill_enabled)),
        )

    async def save_rows(self, version_id: UUID, rows: list[GoodwillRowKeyIn], audit: RequestAudit) -> GoodwillTableOut:
        """Пакет строк одной транзакцией (DOC-259025 §5.1 п. 4): любая ошибка — откат всего пакета.

        Сначала проверяются все строки, потом пишутся только изменённые; аудит `GOODWILL_EDIT` — по каждой изменённой.
        """
        await lock_editable_version(self.session, self.scope, version_id, self.principal.user.id)
        table = await self.table(version_id)
        current = {(r.head_legal_entity.id, r.subsidiary_legal_entity.id): r for r in table.rows}
        saved = {
            (r.head_legal_entity_id, r.subsidiary_legal_entity_id): r
            for r in await repo.saved(self.session, version_id)
        }
        _check_duplicates(rows)
        planned: list[tuple[GoodwillRowOut, domain.Decision]] = []
        for item in rows:
            key = item.head_legal_entity_id, item.subsidiary_legal_entity_id
            row = current.get(key)
            if row is None:
                raise DomainError(HTTPStatus.NOT_FOUND, "NOT_FOUND", "Пара не найдена", extra=_key_extra(key))
            decision = _decision(row, item, key)
            entity = saved.get(key)
            previous = (
                domain.Decision(entity.ownership_share, entity.include_in_kfo, entity.comment)
                if entity
                else domain.Decision()
            )
            if decision != previous:
                planned.append((row, decision))
        now = utc_now()
        written: list[tuple[Goodwill, dict | None]] = []
        for row, decision in planned:
            key = row.head_legal_entity.id, row.subsidiary_legal_entity.id
            entity = saved.get(key)
            before = row.model_dump(mode="json") if entity else None
            if entity is None:
                entity = Goodwill(
                    id=uuid4(),
                    consolidation_version_id=version_id,
                    head_legal_entity_id=key[0],
                    subsidiary_legal_entity_id=key[1],
                    created_by=self.principal.user.id,
                    created_at=now,
                )
                self.session.add(entity)
            entity.ownership_share, entity.include_in_kfo, entity.comment = (
                decision.ownership_share,
                decision.include_in_kfo,
                decision.comment,
            )
            entity.updated_by, entity.updated_at = self.principal.user.id, now
            if row.has_investments:
                for name, amount in _amounts(row, decision).items():
                    if amount is not None and abs(amount) >= MONEY_LIMIT:
                        raise DomainValidationError(
                            "MONEY_OUT_OF_RANGE", "Сумма не помещается в numeric(18,2)", **_key_extra(key)
                        )
                    setattr(entity, name, amount)
            written.append((entity, before))
        if not written:
            return table
        await self.session.flush()
        result = await self.table(version_id)
        after = {(r.head_legal_entity.id, r.subsidiary_legal_entity.id): r for r in result.rows}
        for entity, before in written:
            key = entity.head_legal_entity_id, entity.subsidiary_legal_entity_id
            await audit.record(
                action=AuditAction.GOODWILL_EDIT,
                section=AuditSection.CONSOLIDATION,
                object_type="goodwill",
                object_id=entity.id,
                consolidation_version_id=version_id,
                before_value=before,
                after_value=after[key].model_dump(mode="json"),
            )
        return result

    async def settings(self, version_id: UUID, enabled: bool, audit: RequestAudit) -> GoodwillTableOut:
        version, _ = await lock_editable_version(self.session, self.scope, version_id, self.principal.user.id)
        if version.goodwill_enabled != enabled:
            before = version.goodwill_enabled
            version.goodwill_enabled = enabled
            await self.session.flush()
            await audit.record(
                action=AuditAction.GOODWILL_EDIT,
                section=AuditSection.CONSOLIDATION,
                object_type="consolidation_version",
                object_id=version_id,
                consolidation_version_id=version_id,
                changed_field="goodwill_enabled",
                before_value={"goodwill_enabled": before},
                after_value={"goodwill_enabled": enabled},
            )
        return await self.table(version_id)

    async def export(self, version_id: UUID, audit: RequestAudit, sign: GoodwillSign = "all") -> bytes:

        table = await self.table(version_id)
        content = await run_in_threadpool(build_workbook, table, sign)
        await audit.record(
            action=AuditAction.EXPORT,
            section=AuditSection.EXPORT,
            object_type="consolidation_version",
            object_id=version_id,
            consolidation_version_id=version_id,
            after_value={"report": "goodwill", "rows": sum(matches_sign(row, sign) for row in table.rows)},
        )
        return content


MONEY_LIMIT = Decimal("10000000000000000")


def _key_extra(key: tuple[UUID, UUID]) -> dict[str, str]:
    return {"head_legal_entity_id": str(key[0]), "subsidiary_legal_entity_id": str(key[1])}


def _check_duplicates(rows: list[GoodwillRowKeyIn]) -> None:
    seen: set[tuple[UUID, UUID]] = set()
    for item in rows:
        key = item.head_legal_entity_id, item.subsidiary_legal_entity_id
        if key in seen:
            raise DomainValidationError("DUPLICATE_ROW", "Пара встречается в пакете дважды", **_key_extra(key))
        seen.add(key)


def _decision(row: GoodwillRowOut, item: GoodwillRowKeyIn, key: tuple[UUID, UUID]) -> domain.Decision:
    """Проверенный ввод строки; ошибка несёт ключ пары."""
    share, comment = item.ownership_share, item.comment
    if not row.has_investments:
        # DOC-259025 §10 п. 9: отсутствующий ввод не затирает сохранённые данные неактивной пары.
        if share is None and row.ownership_share is not None:
            # numeric(7,4) читается с масштабом 4 («80.1200»): без хвостовых нулей для проверки двух знаков
            share = format(row.ownership_share.normalize(), "f")
        if comment is None:
            comment = row.comment
    try:
        return validate_decision(
            GoodwillRowIn(ownership_share=share, include_in_kfo=item.include_in_kfo, comment=comment)
        )
    except DomainValidationError as error:
        raise DomainValidationError(error.code, error.message, **_key_extra(key)) from error


def _amounts(row: GoodwillRowOut, decision: domain.Decision) -> dict[str, Decimal | None]:
    """Снимок сумм действующей строки при сохранении (поля `goodwill` DBML)."""
    assert row.financial_investment_amount is not None
    p = (
        domain.participation(row.charter_capital_amount, decision.ownership_share)
        if decision.include_in_kfo and row.charter_capital_amount is not None and decision.ownership_share is not None
        else None
    )
    return {
        "financial_investment_amount": row.financial_investment_amount,
        "charter_capital_amount": row.charter_capital_amount,
        "additional_capital_amount": row.additional_capital_amount,
        "participation_amount": p,
        "goodwill_amount": row.financial_investment_amount - p if p is not None else None,
    }