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

"""Правка строк формы ББ/ОФР — часть `lines` запроса `PATCH …/statement` (spec arm-core «PATCH формы — строки…»;
P3-10, P3-16, P3-24). Вызывается первой в `StatementService.patch`: проверка элементов идёт до первой записи,
поэтому `LINE_CODE_TAKEN` и 422 оставляют форму нетронутой.
"""

import re
import uuid
from collections.abc import Sequence
from decimal import Decimal
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from src.common.exceptions import ConflictError, DomainError, DomainValidationError
from src.models.enums import StatementLineStatus
from src.models.files import Document
from src.models.statements import FinancialStatementLine, FinancialStatementMeta
from src.repositories import statements as repo
from src.schemas.statements import LinePatch

LINE_CODE_TAKEN = "LINE_CODE_TAKEN"
LINE_CODE_REQUIRED = "LINE_CODE_REQUIRED"
INVALID_LINE_CODE = "INVALID_LINE_CODE"
INVALID_LINE_PATCH = "INVALID_LINE_PATCH"
UNKNOWN_LINE = "UNKNOWN_LINE"
ASSIGNED_CONFIDENCE = Decimal("100.00")  # `numeric(5,2)`: ответ отдаёт то, что лежит в сессии, без перечитывания
LINE_CODE_RE = re.compile(r"\d{4,5}", re.ASCII)


def _bad(code: str, message: str, item: LinePatch, *, conflict: bool = False) -> DomainError:
    """Ошибка элемента `lines[]`: в `detail` — `line_id` элемента и его `line_code` или `null` (P3-24)."""
    error = ConflictError if conflict else DomainValidationError
    return error(code, message, line_id=str(item.line_id), line_code=item.line_code)


def _snapshot(line: FinancialStatementLine) -> dict[str, Any]:
    return {
        "line_code": line.line_code,
        "line_status": line.line_status.value,
        "parent_row_index": line.parent_row_index,
        "is_confirmed": line.is_confirmed,
    }


def _check(items: Sequence[LinePatch], by_id: dict[uuid.UUID, FinancialStatementLine]) -> None:
    """Проверки элементов по порядку запроса (первый неверный — в ответ): действие, повтор `line_id`, строка формы,
    `line_status`, формат кода, допустимость действия для строки."""
    seen: set[uuid.UUID] = set()
    for item in items:
        actions = [item.line_code is not None, item.line_status is not None, item.confirmed is not None]
        if sum(actions) != 1:
            raise _bad(INVALID_LINE_PATCH, "в элементе должно быть ровно одно действие", item)
        if item.line_id in seen:
            raise _bad(INVALID_LINE_PATCH, "строка указана в запросе дважды", item)
        seen.add(item.line_id)
        line = by_id.get(item.line_id)
        if line is None:
            raise _bad(UNKNOWN_LINE, "строка не принадлежит форме", item)
        if item.line_status == "ok":
            raise _bad(LINE_CODE_REQUIRED, "в статус ok строку переводит только назначение кода", item)
        if item.line_code is not None and not LINE_CODE_RE.fullmatch(item.line_code):
            raise _bad(INVALID_LINE_CODE, "код строки — 4 или 5 цифр", item)
        if item.line_status == "detail" and line.row_index is None:
            raise _bad(INVALID_LINE_PATCH, "строку, созданную правкой, нельзя отметить расшифровкой", item)
        if item.confirmed is not None and line.line_status is not StatementLineStatus.ok:
            raise _bad(INVALID_LINE_PATCH, "подтверждать можно только код строки со статусом ok", item)


def _apply_item(line: FinancialStatementLine, item: LinePatch) -> bool:
    """Применить действие элемента к строке; `True` — изменены данные (код, статус), а не подтверждение."""
    if item.line_code is not None:
        if line.line_status is StatementLineStatus.ok and line.line_code == item.line_code:
            return False  # тот же код у строки `ok` — менять нечего
        line.line_code, line.line_status = item.line_code, StatementLineStatus.ok
        line.line_code_confidence, line.line_code_modified = ASSIGNED_CONFIDENCE, True
        line.is_confirmed = False  # M-4: подтверждение относилось к прежнему коду
        line.parent_row_index = None  # M-4: у строки `ok` родителя нет (был у `detail`)
        return True
    if item.line_status == "detail":
        if line.line_status is StatementLineStatus.detail:
            return False
        line.line_code, line.line_status = None, StatementLineStatus.detail
        line.is_confirmed = False  # M-4: у расшифровки кода нет — подтверждать нечего
        return True
    if item.confirmed is not None:
        line.is_confirmed = item.confirmed
    return False


async def line_parents(session: AsyncSession, meta: FinancialStatementMeta) -> dict[uuid.UUID, int | None]:
    """`parent_row_index` строк формы — снимок до правки для `merge_parent_changes` (T2.1)."""
    return {line.id: line.parent_row_index for line in await repo.form_lines(session, meta.id)}


async def merge_parent_changes(
    session: AsyncSession,
    meta: FinancialStatementMeta,
    old: dict[uuid.UUID, int | None],
    before: dict[str, Any],
    after: dict[str, Any],
) -> None:
    """После пересчёта (он назначает родителей расшифровок) довнести в `before/after` итоговую смену
    `parent_row_index`: у строк правки значение заменяется итоговым, у остальных строк появляется запись
    `line:<id>` с одним ключом (T2.1)."""
    for line in await repo.form_lines(session, meta.id):
        if line.id not in old or old[line.id] == line.parent_row_index:
            continue
        key = f"line:{line.id}"
        if key in after and after[key] is None:
            continue
        before.setdefault(key, {})["parent_row_index"] = old[line.id]
        after.setdefault(key, {})["parent_row_index"] = line.parent_row_index


async def apply_line_edits(  # noqa: PLR0913 - контракт задачи A13: документ, шапка, элементы, пользователь, аудит
    session: AsyncSession,
    *,
    document: Document,
    meta: FinancialStatementMeta,
    items: Sequence[LinePatch],
    user_id: uuid.UUID | None,  # noqa: ARG001 - у строки нет полей автора правки; интерфейс сохранён для части Б
    before: dict[str, Any],
    after: dict[str, Any],
) -> bool:
    """Применить `lines[]`; ключи `line:<id>` ложатся в `before/after` единственного события аудита. Возвращает `True`,
    если менялись данные (назначение кода, отметка расшифровкой) — тогда акцепт проверок сбрасывается; подтверждение
    кода акцепт не сбрасывает (Р-25)."""
    if not items:
        return False
    lines = await repo.form_lines(session, meta.id)
    by_id = {line.id: line for line in lines}
    _check(items, by_id)
    codes = {item.line_code for item in items if item.line_code is not None}
    templates = {
        line.id: (line, has_amount) for line, has_amount in await repo.template_lines_by_code(session, meta.id, codes)
    }
    for item in items:
        for template, has_amount in templates.values():
            if has_amount and template.line_code == item.line_code and template.id != item.line_id:
                raise _bad(LINE_CODE_TAKEN, "код занят строкой, созданной правкой, с суммами", item, conflict=True)
    # запись: пустые шаблонные строки назначаемых кодов уходят, их id — из `document_issue.line_ids`
    targets = {item.line_id for item in items}
    deleted: list[uuid.UUID] = []
    old = {line.id: _snapshot(line) for line in lines}
    for template, _ in templates.values():
        if template.id in targets or template.line_code not in codes:
            continue
        await repo.delete_line(session, template.id)
        await repo.remove_line_from_issues(session, document.id, template.id)
        lines = [line for line in lines if line.id != template.id]
        deleted.append(template.id)  # T13.2: удаление пустой шаблонной строки — в аудит как `line:<id> -> null`
    data_edited = any([_apply_item(by_id[item.line_id], item) for item in items])
    await session.flush()
    for line in lines:
        now = _snapshot(line)
        if now != old[line.id]:
            before[f"line:{line.id}"], after[f"line:{line.id}"] = old[line.id], now
    for line_id in deleted:
        before[f"line:{line_id}"], after[f"line:{line_id}"] = old[line_id], None
    return data_edited