"""Разбор реального файла ОСВ до классифицированных строк — общий конвейер fixtures-тестов иерархии и классификации. Тот же порядок, что у парсера: лист ОСВ → заголовок таблицы → шапка и схема группировок → строки → каскад иерархии → классификация. Результат кэшируется по пути: несколько тестов одного файла разбирают его один раз; `seconds` — время первого (полного) разбора, включая чтение книги. """ import time from dataclasses import dataclass from functools import cache from pathlib import Path from typing import cast from src.contract import FileType from src.contract.result import OsvGroup from src.parser.classify import ClassifiedRow, classify_rows, collect_body_rows from src.parser.deadline import Deadline from src.parser.detect import SheetKind, TableHeader, classify_sheet, find_table_header from src.parser.grid import Sheet from src.parser.header import DocHeader, build_group_schema, parse_doc_header from src.parser.hierarchy import BodyRow, Levels, resolve_levels from src.parser.readers import read_book from src.parser.rules import load_rules PARSE_TIMEOUT_SECONDS = 120 @dataclass(frozen=True) class RealParse: sheet: Sheet header: TableHeader doc: DocHeader schema: list[OsvGroup] body: list[BodyRow] total_row: int | None levels: Levels rows: list[ClassifiedRow] warnings: list[str] seconds: float @property def data_rows(self) -> list[ClassifiedRow]: """Строки данных без «Итого».""" return [row for row in self.rows if row.group_type != "TOTAL"] def by_text(self, text: str) -> list[ClassifiedRow]: return [row for row in self.rows if row.body.text == text] def row(self, number: int) -> ClassifiedRow: return next(row for row in self.rows if row.body.row == number) @cache def parse_real(path: Path) -> RealParse: rules = load_rules() started = time.monotonic() sheets, _ = read_book(path.read_bytes(), cast(FileType, path.suffix.lstrip(".").upper())) osv = [sheet for sheet in sheets if classify_sheet(sheet, rules) is SheetKind.OSV] assert len(osv) == 1, path.name sheet = osv[0] deadline = Deadline(PARSE_TIMEOUT_SECONDS) header = find_table_header(sheet, rules) doc = parse_doc_header(sheet, header, rules) schema, _ = build_group_schema(sheet, header, rules) body, total_row, warnings = collect_body_rows(sheet, header, rules, deadline) levels = resolve_levels(body, total_row, deadline) rows = classify_rows(body, levels, schema, header, sheet, doc.data_type, total_row, rules) seconds = time.monotonic() - started return RealParse(sheet, header, doc, schema, body, total_row, levels, rows, warnings, seconds)