RSHB/arm-core
9
0
Fork 0
Code Issues 25 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(statements): GET достраивает форму строками справочника (#80)

Browse Source 
GET …/statement добавляет строки справочника редакции, которых нет в форме
(id = null, origin = template), и недостающие ячейки существующих строк
(id = null, все значения пусты). В БД ничего не пишется. Добавлено поле
LineOut.origin (source | template).

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-09 12:00:26 +04:00
parent 438752f035
commit 3a543b5cd1
3 changed files with 230 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
api/src/schemas/statements.py
View File

@ -9,11 +9,12 @@ import datetime
import uuid
from collections.abc import Sequence
from decimal import Decimal
from typing import Annotated
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, Field, StringConstraints
from src.domain.statement_checks import ChecksReport, FailedCheck
from src.domain.statement_lines import TemplateLine
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter
from src.models.files import Document
from src.models.statements import (
@ -56,8 +57,11 @@ class ColumnOut(BaseModel):
is_current: bool | None
LineOrigin = Literal["source", "template"]
class ValueOut(BaseModel):
id: uuid.UUID
id: uuid.UUID | None # None — ячейка достроена в ответе, в БД её нет
column_id: uuid.UUID
column_index: int
amount: Decimal | None
@ -71,7 +75,8 @@ class ValueOut(BaseModel):
class LineOut(BaseModel):
id: uuid.UUID
id: uuid.UUID | None # None — строка достроена из справочника редакции, в БД её нет
origin: LineOrigin
line_code: str | None
line_code_confidence: Decimal | None
line_name: str | None
@ -146,6 +151,71 @@ def _value_out(value: FinancialStatementValue, column_index: dict[uuid.UUID, int
)
def line_origin(line: FinancialStatementLine) -> LineOrigin:
"""`template` — строка создана правкой аналитика (маппинг всегда пишет `line_code_original_value`)."""
return "template" if line.line_code_original_value is None else "source"
def _empty_value(column: FinancialStatementColumn) -> ValueOut:
"""Ячейка, которой нет в БД (design D-3): всё пусто, не распознана и не правлена."""
return ValueOut(
id=None,
column_id=column.id,
column_index=column.column_index,
amount=None,
string_value=None,
confidence=None,
is_recognized=None,
is_modified=False,
original_amount=None,
modified_by=None,
modified_at=None,
)
def _lines_out(
columns: Sequence[FinancialStatementColumn],
lines: Sequence[tuple[FinancialStatementLine, Sequence[FinancialStatementValue]]],
template: Sequence[TemplateLine],
) -> list[LineOut]:
"""Строки БД с ячейками по всем колонкам + строки справочника, которых нет в форме (design D-3), по коду."""
column_index = {column.id: column.column_index for column in columns}
out: list[LineOut] = []
for line, values in lines:
present = {value.column_id for value in values}
cells = [_value_out(value, column_index) for value in values]
cells += [_empty_value(column) for column in columns if column.id not in present]
cells.sort(key=lambda cell: cell.column_index)
out.append(
LineOut(
id=line.id,
origin=line_origin(line),
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=cells,
)
)
codes = {line.line_code for line, _ in lines}
for item in template:
if item.code in codes:
continue
out.append(
LineOut(
id=None,
origin="template",
line_code=item.code,
line_code_confidence=None,
line_name=item.name,
line_name_confidence=None,
values=[_empty_value(column) for column in columns],
)
)
out.sort(key=lambda line: (line.line_code is None, line.line_code or ""))
return out
def _accepted(meta: FinancialStatementMeta) -> ChecksAcceptedOut | None:
if meta.checks_accepted_at is None or meta.checks_accepted_by is None:
return None
@ -163,9 +233,10 @@ def statement_out( # noqa: PLR0913, PLR0917 - сборка ответа из ч
cross: FailedCheck | None,
*,
edition: str | None,
template: Sequence[TemplateLine],
) -> StatementOut:
"""Ответ экрана проверки: документ, шапка, колонки, строки со значениями и проверки."""
column_index = {column.id: column.column_index for column in columns}
"""Ответ экрана проверки: документ, шапка, колонки, строки со значениями и проверки; строки достроены по
справочнику редакции (D-3)."""
return StatementOut(
document=StatementDocumentOut(
id=document.id,
@ -207,17 +278,7 @@ def statement_out( # noqa: PLR0913, PLR0917 - сборка ответа из ч
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
],
lines=_lines_out(columns, lines, template),
checks=ChecksOut(
failed=[_failed(check) for check in report.failed],
soft_failed=[_failed(check) for check in report.soft_failed],

10
api/src/services/statements.py
View File

@ -16,6 +16,7 @@ from src.common.exceptions import ConflictError, DomainValidationError, NotFound
from src.domain.osv_mapping import counterparty_check
from src.domain.policy import VisibilityScope
from src.domain.statement_checks import ChecksReport, FailedCheck, cross_form_check, load_check_rules, run_checks
from src.domain.statement_lines import load_line_catalog
from src.domain.statement_mapping import edition_for, quarter_of, statement_period, statement_status, values_ok
from src.models.directories import LegalEntity
from src.models.enums import (
@ -123,12 +124,13 @@ class StatementService:
return await self.build_out(document, meta)
async def build_out(self, document: Document, meta: FinancialStatementMeta) -> schemas.StatementOut:
"""Ответ формы по текущему состоянию БД (используется и ответами правки/акцепта)."""
"""Ответ формы по текущему состоянию БД, достроенный по справочнику редакции (design D-3; используется и
ответами правки/акцепта). В БД ничего не пишет."""
columns, lines = await repo.load_form(self._session, meta.id)
report, cross = await self.checks_report(document, meta)
return schemas.statement_out(
document, meta, columns, lines, report, cross, edition=edition_for(document.reporting_date)
)
edition = edition_for(document.reporting_date)
template = load_line_catalog().lines(edition, document.document_type.value)
return schemas.statement_out(document, meta, columns, lines, report, cross, edition=edition, template=template)
async def hard_failures(self, document: Document, meta: FinancialStatementMeta) -> tuple[FailedCheck, ...]:
"""Нарушенные жёсткие проверки §7; при акцепте — пустой кортеж (используется и подтверждением формы)."""

150
api/tests/test_api_statements.py
View File

@ -9,7 +9,7 @@ from http import HTTPStatus
from pathlib import Path
import pytest
from sqlalchemy import select, update
from sqlalchemy import delete, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from src.domain.statement_checks import CROSS_FORM_RELATION
@ -42,6 +42,8 @@ ROSTOV = "Ростовский региональный филиал"
MEZH_BALANCE = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма1_бух_09.json"
MEZH_INCOME = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма2_бух_09.json"
MEZH_BALANCE_LINES = 37
MEZH_BALANCE_TEMPLATE_ONLY = {"1215"} # коды fsbu4, которых нет в файле Межениновской (файл по форме 66н)
MEZH_INCOME_TEMPLATE_ONLY = {"2420"}
EVENTS_AFTER_BREAK_AND_ACCEPT = 2
@ -124,7 +126,11 @@ async def test_get_statement(seeded: AsyncSession, fixtures_dir: Path, auth_clie
body = response.json()
assert body["edition"] == "fsbu4"
assert body["document"]["document_type"] == "BALANCE" and body["document"]["period_quarter"] == "Q3"
assert len(body["lines"]) == MEZH_BALANCE_LINES
source = [line for line in body["lines"] if line["origin"] == "source"]
template = [line for line in body["lines"] if line["origin"] == "template"]
assert len(source) == MEZH_BALANCE_LINES and all(line["id"] for line in source)
assert {line["line_code"] for line in template} == MEZH_BALANCE_TEMPLATE_ONLY
assert all(line["id"] is None for line in template)
codes = [line["line_code"] for line in body["lines"]]
assert codes == sorted(codes)
columns = len(body["columns"])
@ -154,7 +160,15 @@ async def test_get_statement_shape_and_decimal_strings(
assert {"organization_name", "okud", "okpo", "okopf", "okfs", "address"} <= set(body["header"])
assert set(body["columns"][0]) == {"id", "column_index", "label", "period_date", "is_current"}
line = next(line for line in body["lines"] if any(v["amount"] is not None for v in line["values"]))
assert set(line) >= {"id", "line_code", "line_code_confidence", "line_name", "line_name_confidence", "values"}
assert set(line) == {
"id",
"origin",
"line_code",
"line_code_confidence",
"line_name",
"line_name_confidence",
"values",
}
value = next(v for v in line["values"] if v["amount"] is not None)
assert isinstance(value["amount"], str)
assert set(value) == {
@ -183,6 +197,136 @@ async def test_get_income_statement(seeded: AsyncSession, fixtures_dir: Path, au
assert body["lines"]
async def _meta(session: AsyncSession, document: Document) -> FinancialStatementMeta:
return (
await session.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == document.id))
).scalar_one()
async def drop_line(session: AsyncSession, document: Document, line_code: str) -> dict[int, Decimal | None]:
"""Удалить строку формы в БД теста (как будто источник её потерял); вернуть её суммы по `column_index`."""
meta = await _meta(session, document)
line = (
await session.execute(
select(FinancialStatementLine).where(
FinancialStatementLine.meta_id == meta.id, FinancialStatementLine.line_code == line_code
)
)
).scalar_one()
rows = (
await session.execute(
select(FinancialStatementColumn.column_index, FinancialStatementValue.amount)
.join(FinancialStatementColumn, FinancialStatementColumn.id == FinancialStatementValue.column_id)
.where(FinancialStatementValue.line_id == line.id)
)
).all()
await session.execute(delete(FinancialStatementValue).where(FinancialStatementValue.line_id == line.id))
await session.execute(delete(FinancialStatementLine).where(FinancialStatementLine.id == line.id))
await session.flush()
return dict(rows)
async def drop_value(session: AsyncSession, document: Document, line_code: str, column_index: int) -> None:
"""Удалить одну ячейку строки в БД теста."""
meta = await _meta(session, document)
line_id = (
await session.execute(
select(FinancialStatementLine.id).where(
FinancialStatementLine.meta_id == meta.id, FinancialStatementLine.line_code == line_code
)
)
).scalar_one()
column_id = (
await session.execute(
select(FinancialStatementColumn.id).where(
FinancialStatementColumn.meta_id == meta.id, FinancialStatementColumn.column_index == column_index
)
)
).scalar_one()
await session.execute(
delete(FinancialStatementValue).where(
FinancialStatementValue.line_id == line_id, FinancialStatementValue.column_id == column_id
)
)
await session.flush()
async def db_line_count(session: AsyncSession, document: Document) -> int:
meta = await _meta(session, document)
ids = await session.execute(select(FinancialStatementLine.id).where(FinancialStatementLine.meta_id == meta.id))
return len(ids.all())
async def test_get_income_statement_template_lines(
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
) -> None:
document = await form(seeded, fixtures_dir, MEZH_INCOME)
client = await auth_client(KA_RF)
body = (await client.get(f"/documents/{document.id}/statement")).json()
template = {line["line_code"] for line in body["lines"] if line["origin"] == "template"}
assert template == MEZH_INCOME_TEMPLATE_ONLY
async def test_get_completes_lost_line_without_writing(
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
) -> None:
"""Строка, потерянная источником (1700), достроена из справочника; статус и проверки — как без достройки."""
document = await form(seeded, fixtures_dir)
await drop_line(seeded, document, "1700")
lines_in_db = await db_line_count(seeded, document)
client = await auth_client(KA_RF)
body = (await client.get(f"/documents/{document.id}/statement")).json()
line = next(line for line in body["lines"] if line["line_code"] == "1700")
assert line["id"] is None and line["origin"] == "template" and line["line_name"] == "БАЛАНС"
assert line["line_code_confidence"] is None and line["line_name_confidence"] is None
assert len(line["values"]) == len(body["columns"])
assert all(
v["id"] is None
and v["amount"] is None
and v["is_recognized"] is None
and v["is_modified"] is False
and v["original_amount"] is None
and v["confidence"] is None
for v in line["values"]
)
assert [v["column_index"] for v in line["values"]] == sorted(c["column_index"] for c in body["columns"])
relations = {(f["relation"], f["column_index"]) for f in body["checks"]["failed"]}
# цель 1700 не проверяется; «1600 = 1700» (цель 1600 есть, 1700 = 0) нарушено — как и без достройки
assert not any(relation.startswith("1700 =") for relation, _ in relations)
assert {index for relation, index in relations if relation == "1600 = 1700"} == {0, 1, 2}
assert body["document"]["status"] == document.status.value # статус пересчитывается только правкой
assert await db_line_count(seeded, document) == lines_in_db
async def test_get_completes_missing_cell(
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
) -> None:
document = await form(seeded, fixtures_dir)
await drop_value(seeded, document, "1150", 1)
client = await auth_client(KA_RF)
body = (await client.get(f"/documents/{document.id}/statement")).json()
line = next(line for line in body["lines"] if line["line_code"] == "1150")
assert line["origin"] == "source" and len(line["values"]) == len(body["columns"])
missing = next(v for v in line["values"] if v["column_index"] == 1)
assert missing["id"] is None and missing["amount"] is None
assert [v["column_index"] for v in line["values"]] == sorted(v["column_index"] for v in line["values"])
async def test_get_without_edition_completes_cells_only(
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
) -> None:
document = await form(seeded, fixtures_dir)
await drop_value(seeded, document, "1150", 1)
document.reporting_date = None
await seeded.flush()
client = await auth_client(KA_RF)
body = (await client.get(f"/documents/{document.id}/statement")).json()
assert body["edition"] is None
assert len(body["lines"]) == MEZH_BALANCE_LINES
assert all(line["origin"] == "source" for line in body["lines"])
assert all(len(line["values"]) == len(body["columns"]) for line in body["lines"])
async def test_statement_of_osv_is_conflict(
seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
) -> None: