ACCOUNT_RE: сегмент субсчёта — 1–2 цифры или 1–3 прописные буквы кириллицы/латиницы (без IGNORECASE); ранг — число сегментов («76.АВ» — сосед «76.04»). Остаточный ложноположительный «12.ООО» описан в docstring. Тесты: буквенные субсчета и их дети (outline и account_pattern), отказы «76.АВГД»/«76.ав»; допуск «60,Расчеты» назван намеренным (R-18); fixtures-свойство «нет строк за схемой», data_type строк = data_type шапки (T8.3); негатив R-13: «Итого» без закрепления вкладывается под последний счёт (T8.5). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
70 lines
3.0 KiB
Python
70 lines
3.0 KiB
Python
"""Разбор реального файла ОСВ до классифицированных строк — общий конвейер 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)
|