arm-core/api/src/schemas/statements.py

265 lines
9.0 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Схемы формы ББ/ОФР для экрана «Проверка распознавания» (spec documents-api «Форма ББ/ОФР для экрана проверки»;
design D-5, D-14).
Суммы — `Decimal` (в JSON — строка, без `float`), как `EntryOut`. Проверки §7 в ответе — вычисленные на лету
(`checks.failed` / `soft_failed` / `cross_form`), акцепт — из шапки.
"""
import datetime
import uuid
from collections.abc import Sequence
from decimal import Decimal
from typing import Annotated
from pydantic import BaseModel, ConfigDict, Field, StringConstraints
from src.domain.statement_checks import ChecksReport, FailedCheck
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter
from src.models.files import Document
from src.models.statements import (
FinancialStatementColumn,
FinancialStatementLine,
FinancialStatementMeta,
FinancialStatementValue,
)
class HeaderFieldOut(BaseModel):
"""Поле шапки, доступное для правки: значение, распознанное исходное и признаки проверки."""
value: str | None
original_value: str | None
confidence: Decimal | None = None
verified: bool | None
modified: bool | None
class StatementHeaderOut(BaseModel):
inn: HeaderFieldOut
report_date: HeaderFieldOut
units: HeaderFieldOut
organization_name: str | None
okud: str | None
okpo: str | None
okopf: str | None
okfs: str | None
address: str | None
class ColumnOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: uuid.UUID
column_index: int
label: str | None
period_date: datetime.date | None
is_current: bool | None
class ValueOut(BaseModel):
id: uuid.UUID
column_id: uuid.UUID
column_index: int
amount: Decimal | None
string_value: str | None
confidence: Decimal | None
is_recognized: bool | None
is_modified: bool
original_amount: Decimal | None
modified_by: uuid.UUID | None
modified_at: datetime.datetime | None
class LineOut(BaseModel):
id: uuid.UUID
line_code: str | None
line_code_confidence: Decimal | None
line_name: str | None
line_name_confidence: Decimal | None
values: list[ValueOut]
class FailedCheckOut(BaseModel):
"""Нарушенное соотношение §7; `column_index = -1` — межформенная проверка (колонки нет)."""
model_config = ConfigDict(from_attributes=True)
column_index: int
relation: str
expected: Decimal
actual: Decimal
class ChecksAcceptedOut(BaseModel):
by: uuid.UUID
at: datetime.datetime
comment: str
class ChecksOut(BaseModel):
failed: list[FailedCheckOut]
soft_failed: list[FailedCheckOut]
cross_form: FailedCheckOut | None
accepted: ChecksAcceptedOut | None
class StatementDocumentOut(BaseModel):
id: uuid.UUID
document_type: DocumentType
status: DocumentStatus
period_year: int | None
period_quarter: PeriodQuarter | None
reporting_date: datetime.date | None
counterparty_check: DocumentCheckResult | None
locator: str | None
file_id: uuid.UUID
class StatementOut(BaseModel):
document: StatementDocumentOut
edition: str | None
header: StatementHeaderOut
columns: list[ColumnOut]
lines: list[LineOut]
checks: ChecksOut
def _failed(check: FailedCheck) -> FailedCheckOut:
return FailedCheckOut(
column_index=check.column_index, relation=check.relation, expected=check.expected, actual=check.actual
)
def _value_out(value: FinancialStatementValue, column_index: dict[uuid.UUID, int]) -> ValueOut:
return ValueOut(
id=value.id,
column_id=value.column_id,
column_index=column_index[value.column_id],
amount=value.amount,
string_value=value.string_value,
confidence=value.confidence,
is_recognized=value.is_recognized,
is_modified=value.is_modified,
original_amount=value.original_amount,
modified_by=value.modified_by,
modified_at=value.modified_at,
)
def _accepted(meta: FinancialStatementMeta) -> ChecksAcceptedOut | None:
if meta.checks_accepted_at is None or meta.checks_accepted_by is None:
return None
return ChecksAcceptedOut(
by=meta.checks_accepted_by, at=meta.checks_accepted_at, comment=meta.checks_accepted_comment or ""
)
def statement_out( # noqa: PLR0913, PLR0917 - сборка ответа из частей формы
document: Document,
meta: FinancialStatementMeta,
columns: Sequence[FinancialStatementColumn],
lines: Sequence[tuple[FinancialStatementLine, Sequence[FinancialStatementValue]]],
report: ChecksReport,
cross: FailedCheck | None,
*,
edition: str | None,
) -> StatementOut:
"""Ответ экрана проверки: документ, шапка, колонки, строки со значениями и проверки."""
column_index = {column.id: column.column_index for column in columns}
return StatementOut(
document=StatementDocumentOut(
id=document.id,
document_type=document.document_type,
status=document.status,
period_year=document.period_year,
period_quarter=document.period_quarter,
reporting_date=document.reporting_date,
counterparty_check=document.counterparty_check,
locator=document.locator,
file_id=document.stored_file_id,
),
edition=edition,
header=StatementHeaderOut(
inn=HeaderFieldOut(
value=meta.inn,
original_value=meta.inn_original_value,
confidence=meta.inn_confidence,
verified=meta.inn_verified,
modified=meta.inn_modified,
),
report_date=HeaderFieldOut(
value=meta.report_date.isoformat() if meta.report_date is not None else None,
original_value=meta.report_date_original_value,
verified=meta.report_date_verified,
modified=meta.report_date_modified,
),
units=HeaderFieldOut(
value=meta.units,
original_value=meta.units_original_value,
verified=meta.units_verified,
modified=meta.units_modified,
),
organization_name=meta.organization_name,
okud=meta.okud,
okpo=meta.okpo,
okopf=meta.okopf,
okfs=meta.okfs,
address=meta.address,
),
columns=[ColumnOut.model_validate(column) for column in columns],
lines=[
LineOut(
id=line.id,
line_code=line.line_code,
line_code_confidence=line.line_code_confidence,
line_name=line.line_name,
line_name_confidence=line.line_name_confidence,
values=[_value_out(value, column_index) for value in values],
)
for line, values in lines
],
checks=ChecksOut(
failed=[_failed(check) for check in report.failed],
soft_failed=[_failed(check) for check in report.soft_failed],
cross_form=_failed(cross) if cross is not None else None,
accepted=_accepted(meta),
),
)
class CellPatch(BaseModel):
"""Новая сумма ячейки: не более двух знаков и 16 цифр до запятой (`numeric(18,2)`), `null` — очистить."""
model_config = ConfigDict(extra="forbid")
line_id: uuid.UUID
column_id: uuid.UUID
amount: Decimal | None = Field(max_digits=18, decimal_places=2)
class HeaderPatch(BaseModel):
"""Отсутствующее поле не меняется; `report_date: null` — очистить дату (`model_fields_set`); `inn: null` — 422
`INVALID_INN` (ИНН нельзя очистить: иначе обходилась бы проверка контрагента).
`units: null` — «не менять» (единицы измерения очищать нельзя); непустая строка после обрезки пробелов, до 50
символов (пробельная строка — 422)."""
model_config = ConfigDict(extra="forbid")
inn: str | None = Field(None, description="10 или 12 цифр; null не допускается (422 INVALID_INN)")
report_date: datetime.date | None = None
units: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=50)] | None = None
class StatementPatch(BaseModel):
model_config = ConfigDict(extra="forbid")
header: HeaderPatch | None = None
values: list[CellPatch] = Field(default_factory=list, max_length=5000)
class ChecksAcceptIn(BaseModel):
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
comment: str = Field(min_length=1, max_length=2000)