From 29198ddc447e60938405a9d9484a43da46f48dd7 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 18 Oct 2026 14:01:56 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D1=88=D0=B0=D0=B3=206=20=D0=B8=20?= =?UTF-8?q?=D0=BA=D0=BE=D0=BD=D0=B2=D0=B5=D0=B9=D0=B5=D1=80=20=D0=BA=D0=B0?= =?UTF-8?q?=D0=BD=D0=B4=D0=B8=D0=B4=D0=B0=D1=82=D0=B0=20=E2=80=94=20=D1=82?= =?UTF-8?q?=D0=B8=D0=BF=20=D0=BF=D0=BE=20=D0=BA=D0=BE=D0=B4=D0=B0=D0=BC,?= =?UTF-8?q?=20=D0=B2=D0=B0=D1=80=D0=B8=D0=B0=D0=BD=D1=82=20=D1=84=D0=BE?= =?UTF-8?q?=D1=80=D0=BC=D1=8B,=20=D0=BF=D0=BE=D1=8F=D1=81=D0=BD=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F,=20=D0=BF=D1=80=D0=B5=D0=B4=D1=83=D0=BF?= =?UTF-8?q?=D1=80=D0=B5=D0=B6=D0=B4=D0=B5=D0=BD=D0=B8=D1=8F=20=D1=84=D0=B0?= =?UTF-8?q?=D0=B9=D0=BB=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit normalize_candidate / normalize; __init__ экспортирует уже реализованный публичный API. DISCLOSURE: шапка не проверяется, предупреждений шапки нет, статус done (P1-6). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W --- arm_statements/__init__.py | 118 +++++++++++++- arm_statements/pipeline.py | 215 ++++++++++++++++++++++++++ arm_statements/tests/test_pipeline.py | 196 +++++++++++++++++++++++ 3 files changed, 527 insertions(+), 2 deletions(-) create mode 100644 arm_statements/pipeline.py create mode 100644 arm_statements/tests/test_pipeline.py diff --git a/arm_statements/__init__.py b/arm_statements/__init__.py index 6339e26..5b2bcf4 100644 --- a/arm_statements/__init__.py +++ b/arm_statements/__init__.py @@ -1,5 +1,119 @@ -"""Общий код нормализации форм ББ/ОФР.""" +"""Общий код нормализации форм ББ/ОФР. +Публичный API — только имена `__all__` (раздел «Публичный API пакета» плана 1). Имена задач 18–19 +(`recompute`, `Stored*`, `to_contract`, `warning_to_contract`, `status_input_from_contract`) добавляются ими. +""" + +from arm_statements.amounts import parse_amount +from arm_statements.codes import parse_code, sheet_joins +from arm_statements.columns import amount_columns +from arm_statements.dates import parse_date, resolve_relative +from arm_statements.header import continuation_type, is_continuation, parse_header, type_hint +from arm_statements.issues import WarningCode, make_warning +from arm_statements.layout import code_column +from arm_statements.loader import Rules, default_rules, load_rules +from arm_statements.model import ( + HEADER_FIELDS, + CandidateResult, + Cell, + CellValue, + DateField, + DocStatus, + DocumentType, + Edition, + EnvelopeStatus, + Evidence, + FailedRelation, + FormCandidate, + FormVariant, + HeaderCells, + HeaderField, + LineStatus, + NormalizationResult, + PeriodKind, + Severity, + SignForm, + Source, + StatementChecks, + StatementColumn, + StatementDocument, + StatementHeader, + StatementLine, + StatementValue, + StatementWarning, +) +from arm_statements.pipeline import normalize, normalize_candidate +from arm_statements.serialize import candidate_from_dict, to_jsonable +from arm_statements.status import ( + StatusInput, + StatusLine, + StatusValue, + document_status, + envelope_status, + status_input, +) +from arm_statements.structure import choose_edition +from arm_statements.text import fix_lookalikes, normalize_caption, normalize_label from arm_statements.version import statements_version -__all__ = ["statements_version"] +__all__ = [ + "HEADER_FIELDS", + "CandidateResult", + "Cell", + "CellValue", + "DateField", + "DocStatus", + "DocumentType", + "Edition", + "EnvelopeStatus", + "Evidence", + "FailedRelation", + "FormCandidate", + "FormVariant", + "HeaderCells", + "HeaderField", + "LineStatus", + "NormalizationResult", + "PeriodKind", + "Rules", + "Severity", + "SignForm", + "Source", + "StatementChecks", + "StatementColumn", + "StatementDocument", + "StatementHeader", + "StatementLine", + "StatementValue", + "StatementWarning", + "StatusInput", + "StatusLine", + "StatusValue", + "WarningCode", + "amount_columns", + "candidate_from_dict", + "choose_edition", + "code_column", + "continuation_type", + "default_rules", + "document_status", + "envelope_status", + "fix_lookalikes", + "is_continuation", + "load_rules", + "make_warning", + "normalize", + "normalize_candidate", + "normalize_caption", + "normalize_label", + "parse_amount", + "parse_code", + "parse_date", + "parse_header", + "resolve_relative", + "sheet_joins", + "statements_version", + "status_input", + "to_jsonable", + "type_hint", +] diff --git a/arm_statements/pipeline.py b/arm_statements/pipeline.py new file mode 100644 index 0000000..aadcd0a --- /dev/null +++ b/arm_statements/pipeline.py @@ -0,0 +1,215 @@ +"""Шаг 6 и конвейер кандидата: шаги 2–8 одного кандидата и нормализация набора кандидатов.""" + +from collections.abc import Sequence +from dataclasses import replace + +from arm_statements.checks import checks_from +from arm_statements.codes import document_type_of, majority_section +from arm_statements.columns import DocumentDates +from arm_statements.evaluation import EvalLine, EvalValue +from arm_statements.header import parse_header +from arm_statements.issues import WarningCode, make_warning +from arm_statements.lines import build_lines, units_warnings +from arm_statements.loader import Rules, default_rules +from arm_statements.model import ( + CandidateResult, + DocStatus, + DocumentType, + FormCandidate, + FormVariant, + LineStatus, + NormalizationResult, + Source, + StatementDocument, + StatementHeader, + StatementLine, + StatementWarning, +) +from arm_statements.review import evaluate, review_header +from arm_statements.rows import RawLine, collect_raw_lines +from arm_statements.segments import Layout, find_layout +from arm_statements.status import document_status, status_input +from arm_statements.text import has_digit, is_disguised_digit + +_SHEET_PREFIX = "sheet:" + + +def _eval_lines(lines: Sequence[StatementLine]) -> list[EvalLine]: + """Строки сервиса в общий вид шагов 6–8: ключ — `ref`, колонка — `str(column_index)`.""" + return [ + EvalLine( + key=line.ref, + order=line.row_index, + line_status=line.line_status, + line_code=line.line_code, + code_confidence=line.line_code_confidence, + code_fixed=False, + values=tuple( + EvalValue(str(v.column_index), v.amount, v.text, v.confidence, v.is_recognized, fixed=False) + for v in line.values + ), + ) + for line in lines + ] + + +def _form_name(candidate: FormCandidate) -> str | None: + title = candidate.evidence.title + if title is None or not title.text.strip(): + return None + return title.text + + +def _has_content(candidate: FormCandidate, rules: Rules) -> bool: + """Содержимое таблицы: ячейка с цифрой или «цифрой под видом буквы».""" + return any( + cell is not None and (has_digit(cell.text) or is_disguised_digit(cell.text, rules.lookalikes)) + for row in candidate.rows + for cell in row + ) + + +def _lost_table(candidate: FormCandidate, rules: Rules) -> StatementWarning: + """Документа нет (нет кодов и `type_hint`): предупреждение файла по источнику и содержимому.""" + if candidate.source is Source.EXCEL: + sheet = candidate.locator.removeprefix(_SHEET_PREFIX).split(",")[0] + message = f"лист «{sheet}» не содержит формы ББ/ОФР" + return make_warning(WarningCode.SHEET_OTHER, message, rules, locator=f"{_SHEET_PREFIX}{sheet}") + if _has_content(candidate, rules): + message = "таблица с данными не распознана как форма: нет столбца кодов и типа формы" + return make_warning(WarningCode.TABLE_LOST, message, rules, locator=candidate.locator) + return make_warning(WarningCode.TABLE_SKIPPED, "таблица без данных пропущена", rules, locator=candidate.locator) + + +def _without_lines( # noqa: PLR0913 -- сборка документа без строк: поля заданы схемой + candidate: FormCandidate, + document_type: DocumentType, + header: StatementHeader, + warnings: Sequence[StatementWarning], + *, + dates: DocumentDates | None, + rules: Rules, +) -> StatementDocument: + """Документ без строк: `DISCLOSURE` (`dates=None` — шапку не проверяем, P1-6) или без столбца кодов (P1-19).""" + collected = list(warnings) + if dates is not None: + header, header_warnings = review_header(header, dates, rules) + collected.extend(header_warnings) + document = StatementDocument( + document_type=document_type, + form_variant=None if document_type is DocumentType.DISCLOSURE else FormVariant.FULL, + locator=candidate.locator, + header=header, + columns=(), + lines=(), + checks=None, + status=DocStatus.DONE, + warnings=tuple(collected), + ) + return replace(document, status=document_status(status_input(document))) + + +def _lines_document( # noqa: PLR0913 -- сборка документа со строками: поля заданы схемой + candidate: FormCandidate, + document_type: DocumentType, + header: StatementHeader, + layout: Layout, + *, + raw: Sequence[RawLine], + row_warnings: Sequence[StatementWarning], + conflict: Sequence[StatementWarning], + rules: Rules, +) -> StatementDocument: + lines, range_warnings = build_lines(raw, candidate.source, rules) + units = units_warnings(lines, header, candidate.warnings, rules) + column_keys = [str(column.column_index) for column in layout.columns] + evaluation = evaluate(_eval_lines(lines), column_keys, document_type, layout.dates.effective, rules, accepted=False) + checks = checks_from(evaluation.outcomes, {key: int(key) for key in column_keys}) + header, header_warnings = review_header(header, dates=layout.dates, rules=rules) + reviewed = tuple( + replace( + line, + needs_review=evaluation.line_needs_review.get(line.ref, False), + values=tuple( + replace(v, needs_review=evaluation.value_needs_review.get((line.ref, str(v.column_index)), False)) + for v in line.values + ), + ) + for line in lines + ) + document = StatementDocument( + document_type=document_type, + form_variant=evaluation.variant, + locator=candidate.locator, + header=header, + columns=layout.columns, + lines=reviewed, + checks=checks, + status=DocStatus.DONE, + warnings=( + *candidate.warnings, + *layout.warnings, + *row_warnings, + *range_warnings, + *units, + *conflict, + *header_warnings, + ), + ) + return replace(document, status=document_status(status_input(document))) + + +def normalize_candidate(candidate: FormCandidate, rules: Rules | None = None) -> CandidateResult: + """Шаги 2–8 кандидата; документа нет — предупреждение файла (`TABLE_LOST`, `TABLE_SKIPPED`, `SHEET_OTHER`).""" + rules = rules if rules is not None else default_rules() + header = replace(parse_header(candidate.header, rules), form_name=_form_name(candidate)) + layout = find_layout(candidate, header, rules) + hint = candidate.type_hint + + if not layout.code_found: + if hint is None: + return CandidateResult(None, (_lost_table(candidate, rules),)) + base = [*candidate.warnings, *layout.warnings] + if hint is DocumentType.DISCLOSURE: + return CandidateResult( + _without_lines(candidate, hint, header, candidate.warnings, dates=None, rules=rules), () + ) + message = "столбец кодов строк не найден" + base.append(make_warning(WarningCode.CODE_COLUMN_NOT_FOUND, message, rules)) + return CandidateResult(_without_lines(candidate, hint, header, base, dates=layout.dates, rules=rules), ()) + + raw, row_warnings = collect_raw_lines(candidate, layout, rules) + ok_codes = [item.code for item in raw if item.line_status is LineStatus.OK and item.code is not None] + section = majority_section(ok_codes, rules) + if section is not None: + document_type = document_type_of(section) + elif hint is not None: + document_type = hint + else: + return CandidateResult(None, (_lost_table(candidate, rules),)) + + conflict: list[StatementWarning] = [] + if hint is not None and hint is not document_type: + message = f"тип по кодам строк ({document_type}) противоречит типу по заголовку ({hint})" + conflict.append(make_warning(WarningCode.TYPE_CONFLICT, message, rules)) + + if document_type is DocumentType.DISCLOSURE: + warnings = [*candidate.warnings, *conflict] + return CandidateResult(_without_lines(candidate, document_type, header, warnings, dates=None, rules=rules), ()) + document = _lines_document( + candidate, document_type, header, layout, raw=raw, row_warnings=row_warnings, conflict=conflict, rules=rules + ) + return CandidateResult(document, ()) + + +def normalize(candidates: Sequence[FormCandidate], rules: Rules | None = None) -> NormalizationResult: + """Кандидаты по порядку; исключения не перехватываются (изоляцию кандидатов делает хост).""" + rules = rules if rules is not None else default_rules() + documents: list[StatementDocument] = [] + file_warnings: list[StatementWarning] = [] + for candidate in candidates: + result = normalize_candidate(candidate, rules) + if result.document is not None: + documents.append(result.document) + file_warnings.extend(result.file_warnings) + return NormalizationResult(tuple(documents), tuple(file_warnings)) diff --git a/arm_statements/tests/test_pipeline.py b/arm_statements/tests/test_pipeline.py new file mode 100644 index 0000000..45af8d9 --- /dev/null +++ b/arm_statements/tests/test_pipeline.py @@ -0,0 +1,196 @@ +from decimal import Decimal as D + +from arm_statements import document_status, normalize, normalize_candidate, status_input +from arm_statements.issues import make_warning +from arm_statements.loader import Rules +from arm_statements.model import DocStatus, DocumentType, FormVariant, Severity +from arm_statements.tests.builders import excel, hdr, ocr + +HEAD = ["Наименование показателя", "Код", "На 30.09.2025"] +FULL_CONFIDENCE = 100.0 +FULL_HEADER = hdr(organization_name="ООО «Пример»", inn="7700000000", report_date="30.09.2025", units="тыс. руб.") + + +def bal_rows() -> list[list[str]]: + return [ + HEAD, + ["Запасы", "1210", "60"], + ["Дебиторы", "1230", "40"], + ["Итого II", "1200", "100"], + ["НМА", "1110", "50"], + ["Итого I", "1100", "50"], + ["БАЛАНС", "1600", "150"], + ["Капитал", "1310", "150"], + ["Итого III", "1300", "150"], + ["БАЛАНС", "1700", "150"], + ] + + +def test_clean_ocr_balance_done(rules: Rules) -> None: + cand = ocr(bal_rows(), header=FULL_HEADER, type_hint=DocumentType.BALANCE, title="Бухгалтерский баланс") + result = normalize_candidate(cand, rules) + doc = result.document + assert doc is not None and result.file_warnings == () + assert doc.header.form_name == "Бухгалтерский баланс" + assert document_status(status_input(doc)) is doc.status + assert (doc.document_type, doc.form_variant, doc.status) == (DocumentType.BALANCE, FormVariant.FULL, DocStatus.DONE) + assert doc.checks is not None and doc.checks.relations_failed == () + assert doc.header.inn.value == "7700000000" + + +def test_disclosure_hint_with_balance_codes(rules: Rules) -> None: + doc = normalize_candidate(ocr(bal_rows(), header=FULL_HEADER, type_hint=DocumentType.DISCLOSURE), rules).document + assert doc is not None and doc.document_type is DocumentType.BALANCE + assert [(w.code, w.severity) for w in doc.warnings if w.code == "TYPE_CONFLICT"] == [ + ("TYPE_CONFLICT", Severity.ERROR) + ] + + +def test_balance_hint_with_income_codes(rules: Rules) -> None: + rows = [ + HEAD, + ["Выручка", "2110", "100"], + ["Себестоимость", "2120", "(60)"], + ["Валовая", "2100", "40"], + ["Коммерческие", "2210", "(10)"], + ["Прибыль от продаж", "2200", "30"], + ] + doc = normalize_candidate(ocr(rows, header=FULL_HEADER, type_hint=DocumentType.BALANCE), rules).document + assert doc is not None and doc.document_type is DocumentType.INCOME + assert "TYPE_CONFLICT" in [w.code for w in doc.warnings] + + +def test_excel_other_sheet(rules: Rules) -> None: + result = normalize_candidate(excel([("Примечания", [["Пояснительный текст без цифр"]])]), rules) + assert result.document is None + assert [(w.code, w.severity, w.locator) for w in result.file_warnings] == [ + ("SHEET_OTHER", Severity.INFO, "sheet:Примечания") + ] + + +def test_ocr_lost_and_skipped_tables(rules: Rules) -> None: + lost = normalize_candidate(ocr([["Итого", "", "15 000"]], locator="file:a.json;table:1"), rules) + assert [(w.code, w.locator) for w in lost.file_warnings] == [("TABLE_LOST", "file:a.json;table:1")] + skipped = normalize_candidate(ocr([["Руководитель", "Иванов"]]), rules) + assert [w.code for w in skipped.file_warnings] == ["TABLE_SKIPPED"] and skipped.document is None + + +def test_hint_without_code_column(rules: Rules) -> None: + doc = normalize_candidate( + ocr([["Итого", "15 000"]], type_hint=DocumentType.BALANCE, header=FULL_HEADER), rules + ).document + assert doc is not None and (doc.lines, doc.columns, doc.checks) == ((), (), None) + assert doc.form_variant is FormVariant.FULL + assert doc.status is DocStatus.NEEDS_CHECK and "CODE_COLUMN_NOT_FOUND" in [w.code for w in doc.warnings] + disc = normalize_candidate(ocr([["Текст"]], type_hint=DocumentType.DISCLOSURE), rules).document + assert disc is not None and (disc.status, disc.form_variant, disc.lines) == (DocStatus.DONE, None, ()) + assert disc.header.form_name is None + + +def test_disclosure_by_codes(rules: Rules) -> None: + rows = [HEAD, ["Пояснение", "5101", "1"], ["Пояснение", "5102", "2"], ["Пояснение", "5100", "3"]] + doc = normalize_candidate(ocr(rows, header=FULL_HEADER), rules).document + assert doc is not None and (doc.document_type, doc.lines, doc.status) == ( + DocumentType.DISCLOSURE, + (), + DocStatus.DONE, + ) + + +def test_simplified_balance(rules: Rules) -> None: + rows = [ + HEAD, + ["Матактивы", "1150", "50"], + ["Запасы", "1210", "50"], + ["БАЛАНС", "1600", "100"], + ["Капитал", "1300", "100"], + ["БАЛАНС", "1700", "100"], + ] + doc = normalize_candidate(ocr(rows, header=FULL_HEADER), rules).document + assert doc is not None and doc.form_variant is FormVariant.SIMPLIFIED + assert doc.checks is not None and doc.checks.relations_failed == () + + +def test_income_without_inn_still_done(rules: Rules) -> None: + rows: list[list[str | D]] = [ + ["Наименование", "Код", "За 9 месяцев 2025 г."], + ["Выручка", "2110", D("100")], + ["Себестоимость", "2120", D("-60")], + ["Валовая прибыль", "2100", D("40")], + ["Прибыль от продаж", "2200", D("40")], + ["Прибыль до налогообложения", "2300", D("40")], + ["Чистая прибыль", "2400", D("40")], + ] + head = hdr(organization_name="ООО «Пример»", units="тыс. руб.") + doc = normalize_candidate(excel([("ОФР", rows)], header=head), rules).document + assert doc is not None and doc.status is DocStatus.DONE + assert [(w.code, w.severity) for w in doc.warnings] == [("INN_NOT_FOUND", Severity.INFO)] + + +def test_normalize_collects(rules: Rules) -> None: + result = normalize([ocr(bal_rows(), header=FULL_HEADER), ocr([["Руководитель", "Иванов"]])], rules) + assert len(result.documents) == 1 and [w.code for w in result.file_warnings] == ["TABLE_SKIPPED"] + + +# --- P1-6: у DISCLOSURE предупреждений шапки нет, статус DONE (требование ревью T15.1) --- + + +def test_disclosure_by_hint_empty_header_has_no_warnings(rules: Rules) -> None: + doc = normalize_candidate(ocr([["Текст"]], type_hint=DocumentType.DISCLOSURE), rules).document + assert doc is not None + assert (doc.warnings, doc.status) == ((), DocStatus.DONE) + + +def test_disclosure_by_codes_empty_header_has_no_warnings(rules: Rules) -> None: + rows = [HEAD, ["Пояснение", "5101", "1"], ["Пояснение", "5102", "2"], ["Пояснение", "5100", "3"]] + doc = normalize_candidate(ocr(rows), rules).document + assert doc is not None and doc.document_type is DocumentType.DISCLOSURE + assert (doc.warnings, doc.status) == ((), DocStatus.DONE) + + +# --- ROW_SPLIT: предупреждение документа без locator (место — locator документа, спека «Шаг 8 — предупреждения») --- + + +def test_row_split_has_no_locator(rules: Rules) -> None: + rows = bal_rows() + rows[1] = ["Запасы и дебиторы", "1210\n1230", "60\n40"] + doc = normalize_candidate(ocr(rows, header=FULL_HEADER, type_hint=DocumentType.BALANCE), rules).document + assert doc is not None + split = [w for w in doc.warnings if w.code == "ROW_SPLIT"] + assert len(split) == 1 and split[0].locator is None and split[0].severity is Severity.INFO + + +# --- строка «в тысячах рублей» над таблицей и уверенность даты отчёта у Excel --- + + +def test_units_row_above_table_is_not_header_and_excel_date_is_confident(rules: Rules) -> None: + rows = [["Бухгалтерский баланс"], ["в тысячах рублей"], *bal_rows()] + cand = excel([("Баланс", rows)], header=FULL_HEADER, type_hint=DocumentType.BALANCE) + doc = normalize_candidate(cand, rules).document + assert doc is not None + assert doc.header.report_date.confidence == FULL_CONFIDENCE and doc.header.report_date.value is not None + assert [c.column_index for c in doc.columns] == [2] + assert (doc.status, [w.code for w in doc.warnings]) == (DocStatus.DONE, []) + + +# --- порядок предупреждений документа: адаптера, раскладки, строк, единиц, TYPE_CONFLICT, шапки --- + + +def test_document_warnings_order(rules: Rules) -> None: + adapter = make_warning("NUMBER_FORMAT_UNPARSED", "формат", rules) + rows = [["Наименование", "Код", "Итого"], *bal_rows()[1:]] + rows[1] = ["Запасы и дебиторы", "1210\n1230", "60,5\n40,5"] + cand = excel([("Баланс", rows)], header=hdr(units="тыс. руб."), type_hint=DocumentType.INCOME, warnings=[adapter]) + doc = normalize_candidate(cand, rules).document + assert doc is not None + codes = [w.code for w in doc.warnings] + expected = [ + "NUMBER_FORMAT_UNPARSED", + "COLUMN_WITHOUT_DATE", + "ROW_SPLIT", + "FRACTIONAL_IN_THOUSANDS", + "TYPE_CONFLICT", + "NO_ORGANIZATION_NO_INN", + ] + assert [c for c in codes if c in expected] == [c for c in expected if c in codes] + assert codes[0] == "NUMBER_FORMAT_UNPARSED" and codes.index("TYPE_CONFLICT") < codes.index("NO_ORGANIZATION_NO_INN")