"""Конвейер форм (design D-3…D-8): листы книги по порядку → блоки форм → шапка, таблица, строки → продолжения → соотношения §7 → `statementDocument`. `ok` документа: жёсткие соотношения сошлись, дубликатов нет, колонка кодов найдена, есть колонка-период, в шапке есть организация или ИНН, документ не одинокое продолжение. DISCLOSURE на статус не влияет. """ from collections.abc import Sequence from src.contract.result import StatementColumn, StatementDocument, StatementHeader from src.parser.deadline import Deadline from src.parser.detect import SheetKind from src.parser.grid import Sheet from src.parser.rules import Rules from src.parser.statements.checks import run_statement_checks from src.parser.statements.columns import NO_CODE_COLUMN, LayoutNotFoundError, TableLayout, find_layout from src.parser.statements.detect import FormBlock, continuation_form, find_forms from src.parser.statements.header import parse_header from src.parser.statements.lines import collect_lines from src.parser.statements.merge import Draft, can_join, join __all__ = ["CONTINUATION_WARNING", "LOCATOR_PREFIX", "LOCATOR_SEPARATOR", "StatementParser"] LOCATOR_PREFIX = "sheet:" LOCATOR_SEPARATOR = "," CONTINUATION_WARNING = "страница-продолжение без шапки: загрузите форму одним файлом" SKIPPED_COLUMN_WARNING = "колонка без периода пропущена" def _line(sheet: Sheet, text: str) -> str: return f"sheet={sheet.name} {text}" class StatementParser: def __init__(self, rules: Rules) -> None: self._rules = rules def parse_sheets( self, sheets: Sequence[tuple[Sheet, SheetKind]], deadline: Deadline ) -> tuple[list[StatementDocument], list[str], bool]: drafts: list[Draft] = [] warnings: list[str] = [] dropped = False for sheet, kind in sheets: deadline.check(f"лист «{sheet.name}»: форма") if kind is SheetKind.STATEMENT: blocks, block_warnings = find_forms(sheet, self._rules) warnings.extend(_line(sheet, text) for text in block_warnings) for block in blocks: drafts.append(self._block_draft(sheet, block, warnings)) else: dropped = not self._continuation(sheet, drafts, warnings) or dropped documents = [self._document(draft) for draft in drafts] for draft in drafts: warnings.extend(draft.warnings) ok = not dropped and all( self._is_ok(draft, document) for draft, document in zip(drafts, documents, strict=True) ) return documents, warnings, ok # --- блоки --- def _block_draft(self, sheet: Sheet, block: FormBlock, warnings: list[str]) -> Draft: if block.kind == "DISCLOSURE": header, _ = parse_header(sheet, self._rules, block.first_row, block.last_row, block) return Draft(kind=block.kind, header=header, periods=(), sheet_names=[sheet.name]) try: layout = find_layout(sheet, self._rules, block.first_row, block.last_row) except LayoutNotFoundError: header, header_warnings = parse_header(sheet, self._rules, block.first_row, block.last_row, block) warnings.extend(_line(sheet, text) for text in [NO_CODE_COLUMN, *header_warnings]) return Draft(kind=block.kind, header=header, periods=(), sheet_names=[sheet.name], has_code_column=False) header_last = (layout.header_row or layout.first_data_row) - 1 header, header_warnings = parse_header(sheet, self._rules, block.first_row, header_last, block) draft = Draft(kind=block.kind, header=header, periods=layout.periods, sheet_names=[sheet.name]) self._fill(sheet, draft, layout, layout.first_data_row, block.last_row, header_warnings, warnings) return draft def _continuation(self, sheet: Sheet, drafts: list[Draft], warnings: list[str]) -> bool: """`False` — лист потерян (нет колонки кодов): строки не попали ни в один документ, статус не `done`.""" try: layout = find_layout(sheet, self._rules, 1, sheet.n_rows) except LayoutNotFoundError: warnings.append(_line(sheet, NO_CODE_COLUMN)) return False open_draft = next((draft for draft in reversed(drafts) if draft.kind != "DISCLOSURE"), None) if open_draft is not None and can_join(open_draft, layout): result = collect_lines( sheet, self._rules, layout, layout.first_data_row, sheet.n_rows, units=open_draft.header.units ) join(open_draft, sheet.name, result.lines, result.duplicates) open_draft.fractional_values += result.fractional_values open_draft.numeric_values += result.numeric_values warnings.extend(_line(sheet, text) for text in result.warnings) return True kind, _ = continuation_form(sheet, self._rules) header, _ = parse_header(sheet, self._rules, 1, (layout.header_row or layout.first_data_row) - 1, None) draft = Draft( kind=kind, header=header, periods=layout.periods, sheet_names=[sheet.name], lonely_continuation=True ) warnings.append(_line(sheet, CONTINUATION_WARNING)) self._fill(sheet, draft, layout, layout.first_data_row, sheet.n_rows, [], warnings) drafts.append(draft) return True def _fill( # noqa: PLR0913, PLR0917 - все части документа передаются явно self, sheet: Sheet, draft: Draft, layout: TableLayout, first_row: int, last_row: int, header_warnings: list[str], warnings: list[str], ) -> None: result = collect_lines(sheet, self._rules, layout, first_row, last_row, units=draft.header.units) draft.lines = result.lines draft.duplicates = result.duplicates draft.fractional_values, draft.numeric_values = result.fractional_values, result.numeric_values texts = [*header_warnings, *result.warnings] if layout.skipped_labels: texts.append(f"{SKIPPED_COLUMN_WARNING}: {', '.join(layout.skipped_labels)}") warnings.extend(_line(sheet, text) for text in texts) # --- документ --- def _document(self, draft: Draft) -> StatementDocument: kind = draft.kind or "OTHER" columns = [ StatementColumn(column_index=index, label=column.label, period_date=column.period_date) for index, column in enumerate(draft.periods) ] locator = LOCATOR_PREFIX + LOCATOR_SEPARATOR.join(draft.sheet_names) if kind == "DISCLOSURE" or not draft.has_code_column: return StatementDocument( document_type=kind, locator=locator, header=draft.header, columns=columns if draft.has_code_column else [], lines=[], ) checks, check_warnings = run_statement_checks( draft.lines, len(columns), kind if kind != "OTHER" else "BALANCE", self._rules ) checks = checks.model_copy(update={"duplicate_line_codes": list(dict.fromkeys(draft.duplicates))}) draft.warnings.extend(f"sheet={draft.sheet_names[0]} {text}" for text in check_warnings) return StatementDocument( document_type=kind, locator=locator, header=draft.header, columns=columns, lines=draft.lines, checks=checks ) @staticmethod def _is_ok(draft: Draft, document: StatementDocument) -> bool: if document.document_type == "DISCLOSURE": return True header: StatementHeader = document.header return ( draft.has_code_column and not draft.lonely_continuation and bool(document.columns) and bool(header.organization_name or header.inn) and document.checks is not None and not document.checks.relations_failed and not document.checks.duplicate_line_codes )