feat: общая функция статуса документа (detail не держит) и статуса конверта
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
parent
6067893db8
commit
a5d10a6b0e
97
arm_statements/status.py
Normal file
97
arm_statements/status.py
Normal file
@ -0,0 +1,97 @@
|
|||||||
|
"""Статус документа и конверта (спека «Шаг 8 — статус документа и конверта»).
|
||||||
|
|
||||||
|
Одна функция на сервис и ядро: вход не зависит от источника строк.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from collections.abc import Sequence
|
||||||
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
from arm_statements.model import (
|
||||||
|
DocStatus,
|
||||||
|
DocumentType,
|
||||||
|
EnvelopeStatus,
|
||||||
|
LineStatus,
|
||||||
|
PeriodKind,
|
||||||
|
Severity,
|
||||||
|
StatementDocument,
|
||||||
|
StatementWarning,
|
||||||
|
)
|
||||||
|
|
||||||
|
_ACCEPTABLE_LINES = frozenset({LineStatus.OK, LineStatus.DETAIL})
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class StatusLine:
|
||||||
|
line_status: LineStatus
|
||||||
|
needs_review: bool
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class StatusValue:
|
||||||
|
line_status: LineStatus
|
||||||
|
needs_review: bool
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass(frozen=True, slots=True)
|
||||||
|
class StatusInput:
|
||||||
|
document_type: DocumentType
|
||||||
|
layout_found: bool
|
||||||
|
has_organization_or_inn: bool
|
||||||
|
period_known: bool
|
||||||
|
needs_review_fields: tuple[str, ...]
|
||||||
|
lines: tuple[StatusLine, ...]
|
||||||
|
values: tuple[StatusValue, ...]
|
||||||
|
hard_failures: int
|
||||||
|
accepted: bool
|
||||||
|
warning_severities: tuple[Severity, ...]
|
||||||
|
|
||||||
|
|
||||||
|
def document_status(data: StatusInput) -> DocStatus:
|
||||||
|
"""`DISCLOSURE` — всегда `done`; иначе `done` только при выполнении всех условий спеки."""
|
||||||
|
if data.document_type is DocumentType.DISCLOSURE:
|
||||||
|
return DocStatus.DONE
|
||||||
|
done = (
|
||||||
|
data.layout_found
|
||||||
|
and data.has_organization_or_inn
|
||||||
|
and data.period_known
|
||||||
|
and all(line.line_status in _ACCEPTABLE_LINES for line in data.lines)
|
||||||
|
and not any(line.needs_review for line in data.lines)
|
||||||
|
# значения строк detail в статус не входят (Р-23)
|
||||||
|
and not any(v.needs_review for v in data.values if v.line_status is not LineStatus.DETAIL)
|
||||||
|
and not data.needs_review_fields
|
||||||
|
and (data.hard_failures == 0 or data.accepted)
|
||||||
|
and Severity.ERROR not in data.warning_severities
|
||||||
|
)
|
||||||
|
return DocStatus.DONE if done else DocStatus.NEEDS_CHECK
|
||||||
|
|
||||||
|
|
||||||
|
def envelope_status(
|
||||||
|
document_statuses: Sequence[DocStatus], file_warnings: Sequence[StatementWarning]
|
||||||
|
) -> EnvelopeStatus:
|
||||||
|
if not document_statuses:
|
||||||
|
return EnvelopeStatus.ERROR
|
||||||
|
if all(s is DocStatus.DONE for s in document_statuses) and all(
|
||||||
|
w.severity is not Severity.ERROR for w in file_warnings
|
||||||
|
):
|
||||||
|
return EnvelopeStatus.DONE
|
||||||
|
return EnvelopeStatus.NEEDS_CHECK
|
||||||
|
|
||||||
|
|
||||||
|
def status_input(document: StatementDocument) -> StatusInput:
|
||||||
|
"""Вход функции статуса из документа шагов 2–8."""
|
||||||
|
header = document.header
|
||||||
|
period_known = header.report_date.value is not None or any(
|
||||||
|
c.period_kind is PeriodKind.ABSOLUTE and c.period_date is not None for c in document.columns
|
||||||
|
)
|
||||||
|
return StatusInput(
|
||||||
|
document_type=document.document_type,
|
||||||
|
layout_found=document.lines != (),
|
||||||
|
has_organization_or_inn=header.organization_name.value is not None or header.inn.value is not None,
|
||||||
|
period_known=period_known,
|
||||||
|
needs_review_fields=header.needs_review_fields,
|
||||||
|
lines=tuple(StatusLine(line.line_status, line.needs_review) for line in document.lines),
|
||||||
|
values=tuple(StatusValue(line.line_status, v.needs_review) for line in document.lines for v in line.values),
|
||||||
|
hard_failures=0 if document.checks is None else len(document.checks.relations_failed),
|
||||||
|
accepted=False,
|
||||||
|
warning_severities=tuple(w.severity for w in document.warnings),
|
||||||
|
)
|
||||||
117
arm_statements/tests/test_status.py
Normal file
117
arm_statements/tests/test_status.py
Normal file
@ -0,0 +1,117 @@
|
|||||||
|
from dataclasses import replace
|
||||||
|
from datetime import date
|
||||||
|
from decimal import Decimal as D
|
||||||
|
|
||||||
|
from arm_statements.header import parse_header
|
||||||
|
from arm_statements.issues import make_warning
|
||||||
|
from arm_statements.loader import Rules
|
||||||
|
from arm_statements.model import (
|
||||||
|
DocStatus,
|
||||||
|
DocumentType,
|
||||||
|
EnvelopeStatus,
|
||||||
|
FormVariant,
|
||||||
|
LineStatus,
|
||||||
|
PeriodKind,
|
||||||
|
Severity,
|
||||||
|
StatementChecks,
|
||||||
|
StatementColumn,
|
||||||
|
StatementDocument,
|
||||||
|
StatementLine,
|
||||||
|
StatementValue,
|
||||||
|
)
|
||||||
|
from arm_statements.status import (
|
||||||
|
StatusInput,
|
||||||
|
StatusLine,
|
||||||
|
StatusValue,
|
||||||
|
document_status,
|
||||||
|
envelope_status,
|
||||||
|
status_input,
|
||||||
|
)
|
||||||
|
from arm_statements.tests.builders import hdr
|
||||||
|
|
||||||
|
OK, DETAIL, UNREADABLE = LineStatus.OK, LineStatus.DETAIL, LineStatus.UNREADABLE
|
||||||
|
DONE, CHECK = DocStatus.DONE, DocStatus.NEEDS_CHECK
|
||||||
|
BASE = StatusInput(
|
||||||
|
document_type=DocumentType.BALANCE,
|
||||||
|
layout_found=True,
|
||||||
|
has_organization_or_inn=True,
|
||||||
|
period_known=True,
|
||||||
|
needs_review_fields=(),
|
||||||
|
lines=(StatusLine(OK, False),),
|
||||||
|
values=(StatusValue(OK, False),),
|
||||||
|
hard_failures=0,
|
||||||
|
accepted=False,
|
||||||
|
warning_severities=(Severity.INFO,),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_clean_form_done() -> None:
|
||||||
|
assert document_status(BASE) is DONE
|
||||||
|
|
||||||
|
|
||||||
|
def test_unreadable_line_blocks() -> None:
|
||||||
|
assert document_status(replace(BASE, lines=(StatusLine(OK, False), StatusLine(UNREADABLE, False)))) is CHECK
|
||||||
|
|
||||||
|
|
||||||
|
def test_detail_value_does_not_block() -> None:
|
||||||
|
lines = (StatusLine(OK, False), StatusLine(DETAIL, False))
|
||||||
|
values = (StatusValue(OK, False), StatusValue(DETAIL, True))
|
||||||
|
assert document_status(replace(BASE, lines=lines, values=values)) is DONE
|
||||||
|
|
||||||
|
|
||||||
|
def test_blocking_conditions() -> None:
|
||||||
|
hard = 2
|
||||||
|
assert document_status(replace(BASE, values=(StatusValue(OK, True),))) is CHECK
|
||||||
|
assert document_status(replace(BASE, lines=(StatusLine(OK, True),))) is CHECK
|
||||||
|
assert document_status(replace(BASE, needs_review_fields=("report_date",))) is CHECK
|
||||||
|
assert document_status(replace(BASE, hard_failures=hard)) is CHECK
|
||||||
|
assert document_status(replace(BASE, hard_failures=hard, accepted=True)) is DONE
|
||||||
|
assert document_status(replace(BASE, warning_severities=(Severity.ERROR,))) is CHECK
|
||||||
|
assert document_status(replace(BASE, layout_found=False)) is CHECK
|
||||||
|
assert document_status(replace(BASE, has_organization_or_inn=False)) is CHECK
|
||||||
|
assert document_status(replace(BASE, period_known=False)) is CHECK
|
||||||
|
|
||||||
|
|
||||||
|
def test_disclosure_always_done() -> None:
|
||||||
|
disclosure = replace(BASE, document_type=DocumentType.DISCLOSURE, layout_found=False, lines=(), values=())
|
||||||
|
assert document_status(disclosure) is DONE
|
||||||
|
|
||||||
|
|
||||||
|
def test_envelope(rules: Rules) -> None:
|
||||||
|
assert envelope_status([DONE, CHECK], []) is EnvelopeStatus.NEEDS_CHECK
|
||||||
|
assert envelope_status([DONE], []) is EnvelopeStatus.DONE
|
||||||
|
lost = make_warning("TABLE_LOST", "таблица без колонки кодов", rules, locator="file:a.json;table:1")
|
||||||
|
assert envelope_status([DONE], [lost]) is EnvelopeStatus.NEEDS_CHECK
|
||||||
|
assert envelope_status([], []) is EnvelopeStatus.ERROR
|
||||||
|
|
||||||
|
|
||||||
|
def test_status_input_from_document(rules: Rules) -> None:
|
||||||
|
head = parse_header(hdr(organization_name="ООО «Пример»", units="тыс. руб."), rules)
|
||||||
|
column = StatementColumn(2, "На 30.09.2025", 100.0, date(2025, 9, 30), PeriodKind.ABSOLUTE)
|
||||||
|
value = StatementValue(2, D("5"), None, "5", 60.0, True, needs_review=True)
|
||||||
|
detail = StatementLine(1, "row:1", DETAIL, None, None, None, "в т. ч.", 100.0, "row:0", False, (value,))
|
||||||
|
line = StatementLine(
|
||||||
|
0, "row:0", OK, "1210", "1210", 100.0, "Запасы", 100.0, None, False, (replace(value, needs_review=False),)
|
||||||
|
)
|
||||||
|
doc = StatementDocument(
|
||||||
|
DocumentType.BALANCE,
|
||||||
|
FormVariant.FULL,
|
||||||
|
"file:t.json;table:0",
|
||||||
|
head,
|
||||||
|
(column,),
|
||||||
|
(line, detail),
|
||||||
|
StatementChecks((), ()),
|
||||||
|
DocStatus.DONE,
|
||||||
|
(),
|
||||||
|
)
|
||||||
|
data = status_input(doc)
|
||||||
|
assert (data.layout_found, data.has_organization_or_inn, data.period_known, data.hard_failures) == (
|
||||||
|
True,
|
||||||
|
True,
|
||||||
|
True,
|
||||||
|
0,
|
||||||
|
)
|
||||||
|
assert data.values == (StatusValue(OK, False), StatusValue(DETAIL, True))
|
||||||
|
assert document_status(data) is DONE
|
||||||
|
empty = replace(doc, columns=(), lines=(), checks=None)
|
||||||
|
assert (status_input(empty).layout_found, status_input(empty).period_known) == (False, False)
|
||||||
Reference in New Issue
Block a user