From 03a7ae1a596deb95f202d6fcdddac10a64406288 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 18 Oct 2026 13:13:31 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D1=82=D0=B8=D0=BF=D1=8B=20=D1=88=D0=B0?= =?UTF-8?q?=D0=B3=D0=B0=201=20=D0=B8=20=D0=B2=D1=8B=D1=85=D0=BE=D0=B4?= =?UTF-8?q?=D0=B0=20=D1=88=D0=B0=D0=B3=D0=BE=D0=B2=202=E2=80=938,=20=D1=81?= =?UTF-8?q?=D0=B5=D1=80=D0=B8=D0=B0=D0=BB=D0=B8=D0=B7=D0=B0=D1=86=D0=B8?= =?UTF-8?q?=D1=8F=20=D0=BA=D0=B0=D0=BD=D0=B4=D0=B8=D0=B4=D0=B0=D1=82=D0=B0?= =?UTF-8?q?=20=D0=B2=20json?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W --- arm_statements/model.py | 246 +++++++++++++++++++++++++++++ arm_statements/serialize.py | 108 +++++++++++++ arm_statements/tests/builders.py | 144 +++++++++++++++++ arm_statements/tests/test_model.py | 37 +++++ 4 files changed, 535 insertions(+) create mode 100644 arm_statements/model.py create mode 100644 arm_statements/serialize.py create mode 100644 arm_statements/tests/builders.py create mode 100644 arm_statements/tests/test_model.py diff --git a/arm_statements/model.py b/arm_statements/model.py new file mode 100644 index 0000000..8b6e451 --- /dev/null +++ b/arm_statements/model.py @@ -0,0 +1,246 @@ +"""Типы входа (кандидат формы) и выхода (нормализованный документ) пакета.""" + +from dataclasses import dataclass +from datetime import date +from decimal import Decimal +from enum import StrEnum + + +class Source(StrEnum): + OCR = "ocr" + EXCEL = "excel" + + +class DocumentType(StrEnum): + BALANCE = "BALANCE" + INCOME = "INCOME" + DISCLOSURE = "DISCLOSURE" + + +class FormVariant(StrEnum): + FULL = "full" + SIMPLIFIED = "simplified" + + +class Edition(StrEnum): + RAS_66N = "66n" + FSBU4 = "fsbu4" + + +class PeriodKind(StrEnum): + ABSOLUTE = "absolute" + CURRENT = "relative:current" + PREV_YEAR_END = "relative:prev_year_end" + PREV_PREV_YEAR_END = "relative:prev_prev_year_end" + PREV_PERIOD = "relative:prev_period" + NONE = "none" + + +class LineStatus(StrEnum): + OK = "ok" + UNREADABLE = "unreadable" + MISSING = "missing" + DETAIL = "detail" + + +class Severity(StrEnum): + ERROR = "error" + INFO = "info" + + +class DocStatus(StrEnum): + DONE = "done" + NEEDS_CHECK = "needs_check" + + +class EnvelopeStatus(StrEnum): + DONE = "done" + NEEDS_CHECK = "needs_check" + ERROR = "error" + + +class SignForm(StrEnum): + PARENTHESES = "parentheses" + MINUS = "minus" + UNSIGNED = "unsigned" + + +class RowRole(StrEnum): + TITLE = "title" + HEADER = "header" + NUMBERING = "numbering" + DATA = "data" + + +CellValue = Decimal | date | bool | None + + +@dataclass(frozen=True, slots=True) +class Cell: + text: str + value: CellValue + confidence: float + ref: str + + +@dataclass(frozen=True, slots=True) +class Evidence: + okud: Cell | None = None + title: Cell | None = None + + +@dataclass(frozen=True, slots=True) +class HeaderCells: + organization_name: Cell | None = None + inn: Cell | None = None + report_date: Cell | None = None + units: Cell | None = None + okud: Cell | None = None + okpo: Cell | None = None + okopf: Cell | None = None + okfs: Cell | None = None + okei: Cell | None = None + address: Cell | None = None + + +HEADER_FIELDS: tuple[str, ...] = ( + "organization_name", + "inn", + "report_date", + "units", + "okud", + "okpo", + "okopf", + "okfs", + "okei", + "address", +) + + +@dataclass(frozen=True, slots=True) +class StatementWarning: + code: str + severity: Severity + message: str + line_refs: tuple[str, ...] = () + column_index: int | None = None + field: str | None = None + locator: str | None = None + + +@dataclass(frozen=True, slots=True) +class FormCandidate: + source: Source + locator: str + type_hint: DocumentType | None + evidence: Evidence + header: HeaderCells + rows: tuple[tuple[Cell | None, ...], ...] + segment_breaks: tuple[int, ...] = () + warnings: tuple[StatementWarning, ...] = () + + +@dataclass(frozen=True, slots=True) +class HeaderField: + text: str | None + confidence: float | None + value: str | None + + +@dataclass(frozen=True, slots=True) +class DateField: + text: str | None + confidence: float | None + value: date | None + + +@dataclass(frozen=True, slots=True) +class StatementHeader: + organization_name: HeaderField + inn: HeaderField + report_date: DateField + units: HeaderField + okud: HeaderField + okpo: HeaderField + okopf: HeaderField + okfs: HeaderField + okei: HeaderField + address: HeaderField + form_name: str | None = None + needs_review_fields: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class StatementColumn: + column_index: int + label: str + label_confidence: float | None + period_date: date | None + period_kind: PeriodKind + + +@dataclass(frozen=True, slots=True) +class StatementValue: + column_index: int + amount: Decimal | None + string_value: str | None + text: str | None + confidence: float + is_recognized: bool + needs_review: bool = False + + +@dataclass(frozen=True, slots=True) +class StatementLine: + row_index: int + ref: str + line_status: LineStatus + line_code: str | None + line_code_text: str | None + line_code_confidence: float | None + line_name: str | None + line_name_confidence: float | None + parent_ref: str | None + needs_review: bool + values: tuple[StatementValue, ...] + + +@dataclass(frozen=True, slots=True) +class FailedRelation: + column_index: int | None + column_key: str + relation: str + codes: tuple[str, ...] + expected: Decimal | None + actual: Decimal | None + + +@dataclass(frozen=True, slots=True) +class StatementChecks: + relations_failed: tuple[FailedRelation, ...] + soft_relations_failed: tuple[FailedRelation, ...] + + +@dataclass(frozen=True, slots=True) +class StatementDocument: + document_type: DocumentType + form_variant: FormVariant | None + locator: str + header: StatementHeader + columns: tuple[StatementColumn, ...] + lines: tuple[StatementLine, ...] + checks: StatementChecks | None + status: DocStatus + warnings: tuple[StatementWarning, ...] + + +@dataclass(frozen=True, slots=True) +class CandidateResult: + document: StatementDocument | None + file_warnings: tuple[StatementWarning, ...] + + +@dataclass(frozen=True, slots=True) +class NormalizationResult: + documents: tuple[StatementDocument, ...] + file_warnings: tuple[StatementWarning, ...] diff --git a/arm_statements/serialize.py b/arm_statements/serialize.py new file mode 100644 index 0000000..03ebf8a --- /dev/null +++ b/arm_statements/serialize.py @@ -0,0 +1,108 @@ +"""Сериализация в json-совместимые значения и обратно (для golden-тестов и обмена).""" + +from collections.abc import Mapping +from dataclasses import fields, is_dataclass +from datetime import date +from decimal import Decimal +from enum import Enum +from typing import Any + +from arm_statements.model import ( + Cell, + CellValue, + DocumentType, + Evidence, + FormCandidate, + HeaderCells, + Severity, + Source, + StatementWarning, +) + + +def _value_to_jsonable(value: CellValue) -> Any: + """Тег типа значения ячейки; bool проверяется раньше прочего (не подкласс Decimal/date, но явно).""" + if value is None: + return None + if isinstance(value, bool): + return {"bool": value} + if isinstance(value, Decimal): + return {"decimal": str(value)} + return {"date": value.isoformat()} + + +def _scalar_to_jsonable(obj: object) -> Any: + if isinstance(obj, Enum): + return obj.value + if isinstance(obj, Decimal): + return str(obj) + if isinstance(obj, date): + return obj.isoformat() + return obj + + +def to_jsonable(obj: object) -> Any: + """Dataclass пакета → json-совместимое значение.""" + if isinstance(obj, Cell): + return { + "text": obj.text, + "value": _value_to_jsonable(obj.value), + "confidence": obj.confidence, + "ref": obj.ref, + } + if is_dataclass(obj) and not isinstance(obj, type): + return {f.name: to_jsonable(getattr(obj, f.name)) for f in fields(obj)} + if isinstance(obj, tuple | list): + return [to_jsonable(item) for item in obj] + return _scalar_to_jsonable(obj) + + +def _value_from_dict(raw: Mapping[str, Any] | None) -> CellValue: + if raw is None: + return None + if "decimal" in raw: + return Decimal(raw["decimal"]) + if "date" in raw: + return date.fromisoformat(raw["date"]) + if "bool" in raw: + return bool(raw["bool"]) + raise ValueError(f"неизвестный тег значения ячейки: {sorted(raw)}") + + +def _cell_from_dict(raw: Mapping[str, Any] | None) -> Cell | None: + if raw is None: + return None + return Cell( + text=raw["text"], + value=_value_from_dict(raw["value"]), + confidence=float(raw["confidence"]), + ref=raw["ref"], + ) + + +def _warning_from_dict(raw: Mapping[str, Any]) -> StatementWarning: + return StatementWarning( + code=raw["code"], + severity=Severity(raw["severity"]), + message=raw["message"], + line_refs=tuple(raw.get("line_refs", ())), + column_index=raw.get("column_index"), + field=raw.get("field"), + locator=raw.get("locator"), + ) + + +def candidate_from_dict(data: Mapping[str, Any]) -> FormCandidate: + """Обратная к `to_jsonable` для кандидата.""" + type_hint = data["type_hint"] + evidence = data["evidence"] + return FormCandidate( + source=Source(data["source"]), + locator=data["locator"], + type_hint=None if type_hint is None else DocumentType(type_hint), + evidence=Evidence(okud=_cell_from_dict(evidence["okud"]), title=_cell_from_dict(evidence["title"])), + header=HeaderCells(**{name: _cell_from_dict(raw) for name, raw in data["header"].items()}), + rows=tuple(tuple(_cell_from_dict(raw) for raw in row) for row in data["rows"]), + segment_breaks=tuple(data.get("segment_breaks", ())), + warnings=tuple(_warning_from_dict(raw) for raw in data.get("warnings", ())), + ) diff --git a/arm_statements/tests/builders.py b/arm_statements/tests/builders.py new file mode 100644 index 0000000..476b2f5 --- /dev/null +++ b/arm_statements/tests/builders.py @@ -0,0 +1,144 @@ +"""Сборщики тестовых кандидатов (только для тестов пакета).""" + +from collections.abc import Mapping, Sequence +from decimal import Decimal + +from arm_statements.model import ( + HEADER_FIELDS, + Cell, + DocumentType, + Evidence, + FormCandidate, + HeaderCells, + Source, + StatementWarning, +) + +OcrInput = str | tuple[str, float] | None +ExcelInput = str | Decimal | None +PlaceInput = str | Decimal | tuple[str, float] + +_FULL_CONFIDENCE = 100.0 +_LATIN_LETTERS = 26 + + +def _text_conf(item: str | tuple[str, float]) -> tuple[str, float]: + if isinstance(item, tuple): + return item + return item, _FULL_CONFIDENCE + + +def hdr(**fields: str | tuple[str, float]) -> HeaderCells: + """Шапка из текстов; ref — `hdr:<поле>`.""" + unknown = set(fields) - set(HEADER_FIELDS) + if unknown: + raise TypeError(f"неизвестные поля шапки: {sorted(unknown)}") + cells: dict[str, Cell] = {} + for name, item in fields.items(): + text, conf = _text_conf(item) + cells[name] = Cell(text, None, conf, f"hdr:{name}") + return HeaderCells(**cells) + + +def _evidence(okud: str | None, title: str | None) -> Evidence: + return Evidence( + okud=None if okud is None else Cell(okud, None, _FULL_CONFIDENCE, "evidence:okud"), + title=None if title is None else Cell(title, None, _FULL_CONFIDENCE, "evidence:title"), + ) + + +def ocr( # noqa: PLR0913 - сигнатура задана планом (Task 2) + rows: Sequence[Sequence[OcrInput]], + *, + type_hint: DocumentType | None = None, + header: HeaderCells | None = None, + title: str | None = None, + okud: str | None = None, + locator: str = "file:t.json;table:0", +) -> FormCandidate: + """Кандидат OCR: строка r → ячейки с ref `row:`.""" + grid: list[tuple[Cell | None, ...]] = [] + for r, row in enumerate(rows): + cells: list[Cell | None] = [] + for item in row: + if item is None: + cells.append(None) + else: + text, conf = _text_conf(item) + cells.append(Cell(text, None, conf, f"row:{r}")) + grid.append(tuple(cells)) + return FormCandidate( + source=Source.OCR, + locator=locator, + type_hint=type_hint, + evidence=_evidence(okud, title), + header=header if header is not None else HeaderCells(), + rows=tuple(grid), + ) + + +def _column_letters(index: int) -> str: + """0 → A, 25 → Z, 26 → AA.""" + letters = "" + n = index + 1 + while n > 0: + n, rem = divmod(n - 1, _LATIN_LETTERS) + letters = chr(ord("A") + rem) + letters + return letters + + +def _sheet_prefix(name: str) -> str: + """Имя листа в ссылке по правилу Excel: кавычки, если не только буквы/цифры/_ или начинается с цифры.""" + plain = all(ch.isalpha() or ch.isdigit() or ch == "_" for ch in name) and not name[:1].isdigit() + if plain: + return name + return "'" + name.replace("'", "''") + "'" + + +def _thousands(value: Decimal) -> str: + return f"{value:,}".replace(",", " ") + + +def excel( # noqa: PLR0913 - сигнатура задана планом (Task 2) + sheets: Sequence[tuple[str, Sequence[Sequence[ExcelInput]]]], + *, + type_hint: DocumentType | None = None, + header: HeaderCells | None = None, + title: str | None = None, + okud: str | None = None, + warnings: Sequence[StatementWarning] = (), +) -> FormCandidate: + """Кандидат Excel: листы подряд в одной сетке, ref `<лист>!`.""" + grid: list[tuple[Cell | None, ...]] = [] + breaks: list[int] = [] + for sheet_no, (name, sheet_rows) in enumerate(sheets): + if sheet_no > 0: + breaks.append(len(grid)) + prefix = _sheet_prefix(name) + for r, row in enumerate(sheet_rows): + cells: list[Cell | None] = [] + for c, item in enumerate(row): + ref = f"{prefix}!{_column_letters(c)}{r + 1}" + if item is None: + cells.append(None) + elif isinstance(item, Decimal): + cells.append(Cell(_thousands(item), item, _FULL_CONFIDENCE, ref)) + else: + cells.append(Cell(item, None, _FULL_CONFIDENCE, ref)) + grid.append(tuple(cells)) + return FormCandidate( + source=Source.EXCEL, + locator="sheet:" + ",".join(name for name, _ in sheets), + type_hint=type_hint, + evidence=_evidence(okud, title), + header=header if header is not None else HeaderCells(), + rows=tuple(grid), + segment_breaks=tuple(breaks), + warnings=tuple(warnings), + ) + + +def place(cells: Mapping[int, PlaceInput], width: int | None = None) -> list[PlaceInput | None]: + """Строка с ячейками в заданных столбцах, остальные None.""" + size = width if width is not None else (max(cells) + 1 if cells else 0) + return [cells.get(i) for i in range(size)] diff --git a/arm_statements/tests/test_model.py b/arm_statements/tests/test_model.py new file mode 100644 index 0000000..945eeb1 --- /dev/null +++ b/arm_statements/tests/test_model.py @@ -0,0 +1,37 @@ +import json +from datetime import date +from decimal import Decimal as D + +from arm_statements.model import Cell, DocumentType, Evidence, FormCandidate, HeaderCells, Source +from arm_statements.serialize import candidate_from_dict, to_jsonable +from arm_statements.tests.builders import excel, ocr + + +def test_candidate_roundtrip_through_json() -> None: + cand = FormCandidate( + source=Source.EXCEL, + locator="sheet:Баланс", + type_hint=DocumentType.BALANCE, + evidence=Evidence(okud=Cell("0710001", None, 100.0, "Баланс!F3"), title=None), + header=HeaderCells(report_date=Cell("30.09.2025", date(2025, 9, 30), 100.0, "Баланс!C5")), + rows=((Cell("106 311", D("106311"), 100.0, "Баланс!D27"), None, Cell("да", True, 100.0, "Баланс!F27")),), + ) + data = json.loads(json.dumps(to_jsonable(cand), ensure_ascii=False)) + assert data["rows"][0][0]["value"] == {"decimal": "106311"} + assert data["header"]["report_date"]["value"] == {"date": "2025-09-30"} + assert candidate_from_dict(data) == cand + + +def test_excel_builder_refs_follow_excel_quoting() -> None: + cand = excel([("Баланс", [["", "", "", "1110"]]), ("Бухгалтерский баланс. Лист 1", [["", "", "1150"]])]) + assert cand.rows[0][3] is not None and cand.rows[0][3].ref == "Баланс!D1" + assert cand.rows[1][2] is not None and cand.rows[1][2].ref == "'Бухгалтерский баланс. Лист 1'!C1" + assert cand.segment_breaks == (1,) + assert cand.locator == "sheet:Баланс,Бухгалтерский баланс. Лист 1" + + +def test_ocr_builder_cells() -> None: + cand = ocr([["Выручка", ("[190 :", 25.0), None]]) + cell = cand.rows[0][1] + assert cell == Cell("[190 :", None, 25.0, "row:0") + assert cand.rows[0][2] is None and cand.source is Source.OCR