Срез 2a: парсер ОСВ из XLS/XLSX/XLSB — динамическое обнаружение структуры (§1), контракт v1.1, golden на 66 реальных файлах #6
54
api/src/parser/errors.py
Normal file
54
api/src/parser/errors.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""Ожидаемые сбои разбора: каждый несёт код контракта и превращается в `ParseOutcome(status=error)` (design D-6)."""
|
||||
|
||||
from typing import ClassVar
|
||||
|
||||
from src.contract import ErrorCode
|
||||
|
||||
__all__ = [
|
||||
"InvalidDocumentError",
|
||||
"InvalidFormatError",
|
||||
"ParseTimeoutError",
|
||||
"ParserError",
|
||||
"UnreadableError",
|
||||
"UnsupportedFormatError",
|
||||
]
|
||||
|
||||
|
||||
class ParserError(Exception):
|
||||
"""Базовая ошибка разбора: `code` — код контракта, `message` — одна строка для `error.message`."""
|
||||
|
||||
code: ClassVar[ErrorCode] = ErrorCode.RECOGNITION_FAILED
|
||||
|
||||
def __init__(self, message: str) -> None:
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
|
||||
|
||||
class UnreadableError(ParserError):
|
||||
"""Книга не открывается библиотекой чтения (повреждённый файл, обрезанный zip/BIFF)."""
|
||||
|
||||
code = ErrorCode.UNREADABLE
|
||||
|
||||
|
||||
class UnsupportedFormatError(ParserError):
|
||||
"""Сигнатура файла не XLS/XLSX/XLSB или превышен лимит размера."""
|
||||
|
||||
code = ErrorCode.UNSUPPORTED_FORMAT
|
||||
|
||||
|
||||
class InvalidFormatError(ParserError):
|
||||
"""Лист ОСВ найден, но нет заголовка таблицы или колонок сумм."""
|
||||
|
||||
code = ErrorCode.INVALID_FORMAT
|
||||
|
||||
|
||||
class InvalidDocumentError(ParserError):
|
||||
"""В книге нет ни одного листа ОСВ."""
|
||||
|
||||
code = ErrorCode.INVALID_DOCUMENT
|
||||
|
||||
|
||||
class ParseTimeoutError(ParserError):
|
||||
"""Превышено время разбора `PARSE_TIMEOUT_SECONDS`."""
|
||||
|
||||
code = ErrorCode.TIMEOUT
|
||||
138
api/src/parser/grid.py
Normal file
138
api/src/parser/grid.py
Normal file
@ -0,0 +1,138 @@
|
||||
"""Единая модель листа (design D-1): читатели xlsx/xls/xlsb приводят книгу к `Sheet`, детекторы работают только с ним.
|
||||
|
||||
Индексы строк и колонок — 1-based, как в Excel (`row_number` контракта). Это и `readers/` — единственные места
|
||||
парсера, где допустимы индексы и буквы колонок; буквы нужны только для диагностики (`column_letter`).
|
||||
"""
|
||||
|
||||
import datetime
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
from functools import cached_property
|
||||
from typing import Literal
|
||||
|
||||
__all__ = ["EMPTY_CELL", "Cell", "CellKind", "MergedRange", "Sheet", "column_letter"]
|
||||
|
||||
CellKind = Literal["empty", "text", "number", "date", "bool"]
|
||||
CellValue = str | Decimal | datetime.date | datetime.datetime | bool | None
|
||||
|
||||
_ALPHABET_SIZE = 26
|
||||
# Диапазоны больше этой площади не раскладываются в индекс «ячейка → диапазон», а проверяются перебором:
|
||||
# merged на весь столбец/лист не должен порождать миллионы записей
|
||||
_MAX_INDEXED_AREA = 10_000
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Cell:
|
||||
"""Значение ячейки как пришло из файла (первичка не мутируется) и признаки оформления для каскада иерархии."""
|
||||
|
||||
value: CellValue
|
||||
kind: CellKind
|
||||
bold: bool = False
|
||||
indent: int = 0 # alignment.indent, а при его отсутствии — число ведущих пробелов текста
|
||||
|
||||
@property
|
||||
def text(self) -> str:
|
||||
"""Строка для сопоставления: `str(value).strip()`; `None` и строка из пробелов дают ""."""
|
||||
if self.value is None:
|
||||
return ""
|
||||
return str(self.value).strip()
|
||||
|
||||
|
||||
EMPTY_CELL = Cell(None, "empty")
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class MergedRange:
|
||||
"""Объединённый диапазон, границы 1-based включительно; значение лежит в (top, left)."""
|
||||
|
||||
top: int
|
||||
left: int
|
||||
bottom: int
|
||||
right: int
|
||||
|
||||
@property
|
||||
def area(self) -> int:
|
||||
return (self.bottom - self.top + 1) * (self.right - self.left + 1)
|
||||
|
||||
def contains(self, row: int, col: int) -> bool:
|
||||
return self.top <= row <= self.bottom and self.left <= col <= self.right
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Sheet:
|
||||
"""Лист книги: только непустые ячейки, merged-диапазоны, outline level строк, габариты."""
|
||||
|
||||
name: str
|
||||
cells: Mapping[tuple[int, int], Cell] # (row, col) 1-based
|
||||
merged: tuple[MergedRange, ...]
|
||||
row_outline: Mapping[int, int] # только строки с уровнем > 0
|
||||
n_rows: int
|
||||
n_cols: int
|
||||
|
||||
def cell(self, row: int, col: int) -> Cell:
|
||||
return self.cells.get((row, col), EMPTY_CELL)
|
||||
|
||||
def text_at(self, row: int, col: int) -> str:
|
||||
"""Текст ячейки; внутри merged-диапазона — текст его origin (левой верхней ячейки)."""
|
||||
return self.cell(*self.merged_origin(row, col)).text
|
||||
|
||||
def merged_origin(self, row: int, col: int) -> tuple[int, int]:
|
||||
span = self.merged_span(row, col)
|
||||
if span is None:
|
||||
return row, col
|
||||
return span.top, span.left
|
||||
|
||||
def merged_span(self, row: int, col: int) -> MergedRange | None:
|
||||
span = self._merged_index.get((row, col))
|
||||
if span is not None:
|
||||
return span
|
||||
for large in self._large_merged:
|
||||
if large.contains(row, col):
|
||||
return large
|
||||
return None
|
||||
|
||||
def row_cells(self, row: int) -> list[tuple[int, Cell]]:
|
||||
"""Непустые ячейки строки `(col, cell)` по возрастанию col; ячейки из одних пробелов пропускаются."""
|
||||
return list(self._rows.get(row, ()))
|
||||
|
||||
def is_row_empty(self, row: int) -> bool:
|
||||
return row not in self._rows
|
||||
|
||||
def outline(self, row: int) -> int:
|
||||
return self.row_outline.get(row, 0)
|
||||
|
||||
# Индексы строятся лениво один раз: cached_property пишет в __dict__ экземпляра в обход frozen-__setattr__
|
||||
@cached_property
|
||||
def _rows(self) -> dict[int, tuple[tuple[int, Cell], ...]]:
|
||||
rows: dict[int, list[tuple[int, Cell]]] = {}
|
||||
for (row, col), cell in self.cells.items():
|
||||
if cell.text:
|
||||
rows.setdefault(row, []).append((col, cell))
|
||||
return {row: tuple(sorted(items, key=lambda item: item[0])) for row, items in rows.items()}
|
||||
|
||||
@cached_property
|
||||
def _merged_index(self) -> dict[tuple[int, int], MergedRange]:
|
||||
index: dict[tuple[int, int], MergedRange] = {}
|
||||
for span in self.merged:
|
||||
if span.area > _MAX_INDEXED_AREA:
|
||||
continue
|
||||
for row in range(span.top, span.bottom + 1):
|
||||
for col in range(span.left, span.right + 1):
|
||||
index[(row, col)] = span
|
||||
return index
|
||||
|
||||
@cached_property
|
||||
def _large_merged(self) -> tuple[MergedRange, ...]:
|
||||
return tuple(span for span in self.merged if span.area > _MAX_INDEXED_AREA)
|
||||
|
||||
|
||||
def column_letter(index: int) -> str:
|
||||
"""Буква колонки Excel по 1-based индексу (1 → "A", 27 → "AA"). Только для диагностики `columns`."""
|
||||
if index < 1:
|
||||
raise ValueError(f"индекс колонки 1-based, получено {index}")
|
||||
letters: list[str] = []
|
||||
while index:
|
||||
index, remainder = divmod(index - 1, _ALPHABET_SIZE)
|
||||
letters.append(chr(ord("A") + remainder))
|
||||
return "".join(reversed(letters))
|
||||
62
api/src/parser/readers/__init__.py
Normal file
62
api/src/parser/readers/__init__.py
Normal file
@ -0,0 +1,62 @@
|
||||
"""Чтение книги Excel в модель `Sheet`: формат определяется по сигнатуре байтов, а не по `file_type` запроса.
|
||||
|
||||
PK\\x03\\x04 (zip) → XLSX или XLSB (XLSB — если в архиве есть `xl/workbook.bin`); OLE2 (D0 CF 11 E0 …) → XLS (BIFF8).
|
||||
Несовпадение с `file_type` — предупреждение, чтение по сигнатуре. Иная сигнатура → `UnsupportedFormatError`;
|
||||
исключение библиотеки чтения → `UnreadableError` с причиной одной строкой.
|
||||
"""
|
||||
|
||||
import io
|
||||
import zipfile
|
||||
from collections.abc import Callable
|
||||
|
||||
from src.contract import FileType
|
||||
from src.parser.errors import ParserError, UnreadableError, UnsupportedFormatError
|
||||
from src.parser.grid import Sheet
|
||||
from src.parser.readers._common import one_line_reason
|
||||
from src.parser.readers.xls import read_xls
|
||||
from src.parser.readers.xlsb import OUTLINE_UNAVAILABLE_WARNING, read_xlsb
|
||||
from src.parser.readers.xlsx import read_xlsx
|
||||
|
||||
__all__ = ["read_book"]
|
||||
|
||||
ZIP_SIGNATURE = b"PK\x03\x04"
|
||||
OLE2_SIGNATURE = bytes.fromhex("D0CF11E0A1B11AE1")
|
||||
XLSB_WORKBOOK_PART = "xl/workbook.bin"
|
||||
|
||||
# Формат → (название содержимого для предупреждения, читатель)
|
||||
_READERS: dict[FileType, tuple[str, Callable[[bytes], tuple[Sheet, ...]]]] = {
|
||||
"XLSX": ("OOXML", read_xlsx),
|
||||
"XLSB": ("BIFF12", read_xlsb),
|
||||
"XLS": ("BIFF8", read_xls),
|
||||
}
|
||||
|
||||
|
||||
def read_book(data: bytes, file_type: FileType) -> tuple[tuple[Sheet, ...], list[str]]:
|
||||
"""Листы книги и предупреждения чтения (несовпадение `file_type`, ограничения xlsb)."""
|
||||
detected = _detect(data, file_type)
|
||||
content, reader = _READERS[detected]
|
||||
warnings: list[str] = []
|
||||
if detected != file_type:
|
||||
warnings.append(f"file_type={file_type}, но содержимое {content}: прочитано как {detected}")
|
||||
try:
|
||||
sheets = reader(data)
|
||||
except ParserError:
|
||||
raise
|
||||
except Exception as exc: # библиотеки чтения бросают что угодно: zipfile, KeyError, struct.error, XLRDError…
|
||||
raise UnreadableError(f"книга не читается ({detected}): {one_line_reason(exc)}") from exc
|
||||
if detected == "XLSB":
|
||||
warnings.append(OUTLINE_UNAVAILABLE_WARNING)
|
||||
return sheets, warnings
|
||||
|
||||
|
||||
def _detect(data: bytes, file_type: FileType) -> FileType:
|
||||
if data.startswith(OLE2_SIGNATURE):
|
||||
return "XLS"
|
||||
if data.startswith(ZIP_SIGNATURE):
|
||||
try:
|
||||
with zipfile.ZipFile(io.BytesIO(data)) as archive:
|
||||
names = set(archive.namelist())
|
||||
except (zipfile.BadZipFile, OSError, EOFError) as exc:
|
||||
raise UnreadableError(f"повреждённый zip-архив: {one_line_reason(exc)}") from exc
|
||||
return "XLSB" if XLSB_WORKBOOK_PART in names else "XLSX"
|
||||
raise UnsupportedFormatError(f"содержимое файла не XLS/XLSX/XLSB (file_type={file_type})")
|
||||
40
api/src/parser/readers/_common.py
Normal file
40
api/src/parser/readers/_common.py
Normal file
@ -0,0 +1,40 @@
|
||||
"""Общее для читателей: значение библиотеки → `Cell`, ошибка библиотеки → одна строка причины."""
|
||||
|
||||
import datetime
|
||||
from decimal import Decimal
|
||||
|
||||
from src.parser.grid import Cell
|
||||
|
||||
|
||||
def leading_spaces(text: str) -> int:
|
||||
"""Число ведущих пробельных символов (1С иногда делает отступ иерархии пробелами, а не alignment).
|
||||
|
||||
У строки из одних пробелов отступа нет: это пустая ячейка, а не сдвинутый текст.
|
||||
"""
|
||||
stripped = text.lstrip()
|
||||
return len(text) - len(stripped) if stripped else 0
|
||||
|
||||
|
||||
def make_cell(value: object, *, bold: bool = False, indent: int = 0) -> Cell | None:
|
||||
"""Ячейка из значения библиотеки чтения; `None` — пустая ячейка (в `Sheet.cells` не попадает).
|
||||
|
||||
Числа — `Decimal(str(value))` без квантования (его делает `numbers.py`); bool проверяется раньше int.
|
||||
Строка из пробелов остаётся `text` (значение как пришло), её `Cell.text` пуст. Время без даты — текстом.
|
||||
"""
|
||||
if value is None or value == "":
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return Cell(value, "bool", bold=bold, indent=indent)
|
||||
if isinstance(value, int | float):
|
||||
return Cell(Decimal(str(value)), "number", bold=bold, indent=indent)
|
||||
if isinstance(value, datetime.date): # datetime.datetime — подкласс date
|
||||
return Cell(value, "date", bold=bold, indent=indent)
|
||||
text = value if isinstance(value, str) else str(value)
|
||||
return Cell(text, "text", bold=bold, indent=indent or leading_spaces(text))
|
||||
|
||||
|
||||
def one_line_reason(exc: BaseException) -> str:
|
||||
"""Причина сбоя библиотеки одной строкой: `Тип: сообщение` без переводов строк."""
|
||||
detail = " ".join(str(exc).split())
|
||||
name = type(exc).__name__
|
||||
return f"{name}: {detail}" if detail else name
|
||||
68
api/src/parser/readers/xls.py
Normal file
68
api/src/parser/readers/xls.py
Normal file
@ -0,0 +1,68 @@
|
||||
"""Чтение XLS (BIFF8) через xlrd с `formatting_info`: merged, outline level, шрифт и отступ из XF-записей."""
|
||||
|
||||
import io
|
||||
from typing import Any
|
||||
|
||||
import xlrd
|
||||
|
||||
from src.parser.grid import Cell, MergedRange, Sheet
|
||||
from src.parser.readers._common import make_cell
|
||||
|
||||
__all__ = ["read_xls"]
|
||||
|
||||
|
||||
def read_xls(data: bytes) -> tuple[Sheet, ...]:
|
||||
# logfile: xlrd пишет замечания о структуре BIFF в stdout — в выводе сервиса им не место
|
||||
book = xlrd.open_workbook(file_contents=data, formatting_info=True, logfile=io.StringIO())
|
||||
try:
|
||||
return tuple(_sheet(book, sheet) for sheet in book.sheets())
|
||||
finally:
|
||||
book.release_resources()
|
||||
|
||||
|
||||
def _sheet(book: Any, sheet: Any) -> Sheet:
|
||||
cells: dict[tuple[int, int], Cell] = {}
|
||||
for rowx in range(sheet.nrows):
|
||||
types = sheet.row_types(rowx)
|
||||
values = sheet.row_values(rowx)
|
||||
for colx, (ctype, raw) in enumerate(zip(types, values, strict=True)):
|
||||
value = _value(book, ctype, raw)
|
||||
if value is None:
|
||||
continue
|
||||
xf = book.xf_list[sheet.cell_xf_index(rowx, colx)]
|
||||
cell = make_cell(
|
||||
value,
|
||||
bold=bool(book.font_list[xf.font_index].bold),
|
||||
indent=int(xf.alignment.indent_level or 0),
|
||||
)
|
||||
if cell is not None:
|
||||
cells[(rowx + 1, colx + 1)] = cell
|
||||
# xlrd: (rlo, rhi, clo, chi), 0-based, полуоткрытые → 1-based включительно
|
||||
merged = tuple(
|
||||
MergedRange(top=rlo + 1, left=clo + 1, bottom=rhi, right=chi) for rlo, rhi, clo, chi in sheet.merged_cells
|
||||
)
|
||||
row_outline = {rowx + 1: int(info.outline_level) for rowx, info in sheet.rowinfo_map.items() if info.outline_level}
|
||||
return Sheet(
|
||||
name=str(sheet.name),
|
||||
cells=cells,
|
||||
merged=merged,
|
||||
row_outline=row_outline,
|
||||
n_rows=int(sheet.nrows),
|
||||
n_cols=int(sheet.ncols),
|
||||
)
|
||||
|
||||
|
||||
def _value(book: Any, ctype: int, raw: Any) -> object:
|
||||
"""Значение ячейки xlrd по её типу; пустые и BLANK → None."""
|
||||
if ctype in (xlrd.XL_CELL_EMPTY, xlrd.XL_CELL_BLANK):
|
||||
return None
|
||||
if ctype == xlrd.XL_CELL_BOOLEAN:
|
||||
return bool(raw)
|
||||
if ctype == xlrd.XL_CELL_ERROR:
|
||||
return xlrd.error_text_from_code.get(raw, f"#ERR{raw}")
|
||||
if ctype == xlrd.XL_CELL_DATE:
|
||||
try:
|
||||
return xlrd.xldate_as_datetime(raw, book.datemode)
|
||||
except (xlrd.xldate.XLDateError, ValueError, OverflowError):
|
||||
return raw # серийный номер вне диапазона дат — остаётся числом, как в файле
|
||||
return raw
|
||||
37
api/src/parser/readers/xlsb.py
Normal file
37
api/src/parser/readers/xlsb.py
Normal file
@ -0,0 +1,37 @@
|
||||
"""Чтение XLSB (BIFF12) через pyxlsb: только значения; outline level, merged и шрифт библиотека не отдаёт."""
|
||||
|
||||
import io
|
||||
from typing import Any
|
||||
|
||||
import pyxlsb
|
||||
|
||||
from src.parser.grid import Cell, Sheet
|
||||
from src.parser.readers._common import make_cell
|
||||
|
||||
__all__ = ["OUTLINE_UNAVAILABLE_WARNING", "read_xlsb"]
|
||||
|
||||
OUTLINE_UNAVAILABLE_WARNING = "xlsb: outline level недоступен"
|
||||
|
||||
|
||||
def read_xlsb(data: bytes) -> tuple[Sheet, ...]:
|
||||
with pyxlsb.open_workbook(io.BytesIO(data)) as book:
|
||||
return tuple(_sheet(str(name), book) for name in book.sheets)
|
||||
|
||||
|
||||
def _sheet(name: str, book: Any) -> Sheet:
|
||||
cells: dict[tuple[int, int], Cell] = {}
|
||||
with book.get_sheet(name) as sheet:
|
||||
for row in sheet.rows(sparse=True):
|
||||
for source in row:
|
||||
cell = make_cell(source.v)
|
||||
if cell is not None:
|
||||
cells[(source.r + 1, source.c + 1)] = cell
|
||||
# Даты в xlsb — серийные числа без формата (pyxlsb не читает стили): остаются числами
|
||||
return Sheet(
|
||||
name=name,
|
||||
cells=cells,
|
||||
merged=(),
|
||||
row_outline={},
|
||||
n_rows=max((row for row, _ in cells), default=0),
|
||||
n_cols=max((col for _, col in cells), default=0),
|
||||
)
|
||||
44
api/src/parser/readers/xlsx.py
Normal file
44
api/src/parser/readers/xlsx.py
Normal file
@ -0,0 +1,44 @@
|
||||
"""Чтение XLSX через openpyxl (полное чтение: нужны merged, outline level и шрифт)."""
|
||||
|
||||
import io
|
||||
from typing import Any
|
||||
|
||||
import openpyxl
|
||||
|
||||
from src.parser.grid import Cell, MergedRange, Sheet
|
||||
from src.parser.readers._common import make_cell
|
||||
|
||||
__all__ = ["read_xlsx"]
|
||||
|
||||
|
||||
def read_xlsx(data: bytes) -> tuple[Sheet, ...]:
|
||||
book = openpyxl.load_workbook(io.BytesIO(data), data_only=True, read_only=False)
|
||||
try:
|
||||
return tuple(_sheet(worksheet) for worksheet in book.worksheets)
|
||||
finally:
|
||||
book.close()
|
||||
|
||||
|
||||
def _sheet(ws: Any) -> Sheet:
|
||||
cells: dict[tuple[int, int], Cell] = {}
|
||||
# `_cells` — разреженное хранилище openpyxl (версия закреплена): iter_rows создал бы пустые ячейки на весь
|
||||
# прямоугольник листа
|
||||
for source in ws._cells.values():
|
||||
cell = make_cell(source.value, bold=bool(source.font.bold), indent=int(source.alignment.indent or 0))
|
||||
if cell is not None:
|
||||
cells[(source.row, source.column)] = cell
|
||||
merged = tuple(
|
||||
MergedRange(top=span.min_row, left=span.min_col, bottom=span.max_row, right=span.max_col)
|
||||
for span in ws.merged_cells.ranges
|
||||
)
|
||||
row_outline = {
|
||||
row: int(dimension.outline_level) for row, dimension in ws.row_dimensions.items() if dimension.outline_level
|
||||
}
|
||||
return Sheet(
|
||||
name=str(ws.title),
|
||||
cells=cells,
|
||||
merged=merged,
|
||||
row_outline=row_outline,
|
||||
n_rows=int(ws.max_row),
|
||||
n_cols=int(ws.max_column),
|
||||
)
|
||||
49
api/tests/fixtures/test_readers_real.py
vendored
Normal file
49
api/tests/fixtures/test_readers_real.py
vendored
Normal file
@ -0,0 +1,49 @@
|
||||
"""Читатели на реальных ОСВ из ../fixtures: все 66 файлов читаются быстро, outline и merged на месте."""
|
||||
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import cast
|
||||
|
||||
import pytest
|
||||
|
||||
from src.contract import FileType
|
||||
from src.parser.readers import read_book
|
||||
|
||||
READ_SECONDS_LIMIT = 5.0
|
||||
NAMED_XLS = "osv/schastl/2024-Q3/Оборотносальдовая ведомость по счету 62 за 3 квартал 2024 г..xls"
|
||||
# Строка 15 (1-based) — контрагент «МАЛЕЧКИНСКАЯ ПТИЦЕФАБРИКА ООО» под 62 → 62.01 → «БП Аренда»
|
||||
NAMED_XLS_ROW = 15
|
||||
NAMED_XLS_OUTLINE = 3
|
||||
|
||||
|
||||
@pytest.mark.fixtures
|
||||
def test_real_file_is_read_quickly_with_outline_and_merged(osv_file: Path) -> None:
|
||||
data = osv_file.read_bytes()
|
||||
started = time.perf_counter()
|
||||
|
||||
sheets, warnings = read_book(data, cast(FileType, osv_file.suffix.lstrip(".").upper()))
|
||||
|
||||
assert time.perf_counter() - started < READ_SECONDS_LIMIT
|
||||
assert warnings == []
|
||||
assert sheets
|
||||
assert any(sheet.row_outline for sheet in sheets)
|
||||
assert all(level >= 1 for sheet in sheets for level in sheet.row_outline.values())
|
||||
assert any(sheet.merged for sheet in sheets)
|
||||
assert any(sheet.cells for sheet in sheets)
|
||||
|
||||
|
||||
@pytest.mark.fixtures
|
||||
def test_named_xls_outline_level(fixtures_dir: Path) -> None:
|
||||
sheets, _ = read_book((fixtures_dir / NAMED_XLS).read_bytes(), "XLS")
|
||||
|
||||
sheet = sheets[0]
|
||||
assert sheet.cell(NAMED_XLS_ROW, 1).text == "МАЛЕЧКИНСКАЯ ПТИЦЕФАБРИКА ООО"
|
||||
assert sheet.outline(NAMED_XLS_ROW) == NAMED_XLS_OUTLINE
|
||||
|
||||
|
||||
@pytest.mark.fixtures
|
||||
def test_biff8_content_with_xlsx_file_type_is_read_as_xls(fixtures_dir: Path) -> None:
|
||||
sheets, warnings = read_book((fixtures_dir / NAMED_XLS).read_bytes(), "XLSX")
|
||||
|
||||
assert warnings == ["file_type=XLSX, но содержимое BIFF8: прочитано как XLS"]
|
||||
assert sheets[0].outline(NAMED_XLS_ROW) == NAMED_XLS_OUTLINE
|
||||
93
api/tests/unit/parser/test_grid.py
Normal file
93
api/tests/unit/parser/test_grid.py
Normal file
@ -0,0 +1,93 @@
|
||||
"""Модель листа `Sheet`: merged-диапазоны, порядок ячеек строки, пустота, буквы колонок для диагностики."""
|
||||
|
||||
import datetime
|
||||
from decimal import Decimal
|
||||
|
||||
import pytest
|
||||
|
||||
from src.parser.grid import EMPTY_CELL, Cell, MergedRange, Sheet, column_letter
|
||||
|
||||
|
||||
def _sheet(
|
||||
cells: dict[tuple[int, int], Cell],
|
||||
merged: tuple[MergedRange, ...] = (),
|
||||
row_outline: dict[int, int] | None = None,
|
||||
) -> Sheet:
|
||||
n_rows = max((row for row, _ in cells), default=0)
|
||||
n_cols = max((col for _, col in cells), default=0)
|
||||
return Sheet(name="ОСВ", cells=cells, merged=merged, row_outline=row_outline or {}, n_rows=n_rows, n_cols=n_cols)
|
||||
|
||||
|
||||
def test_text_at_inside_merged_range_takes_origin_value() -> None:
|
||||
sheet = _sheet(
|
||||
{(1, 2): Cell("Сальдо на начало периода", "text"), (2, 2): Cell("Дебет", "text")},
|
||||
merged=(MergedRange(top=1, left=2, bottom=1, right=3),),
|
||||
)
|
||||
|
||||
assert sheet.text_at(1, 3) == "Сальдо на начало периода"
|
||||
assert sheet.text_at(1, 2) == "Сальдо на начало периода"
|
||||
assert sheet.cell(1, 3) is EMPTY_CELL
|
||||
assert sheet.merged_origin(1, 3) == (1, 2)
|
||||
assert sheet.merged_span(1, 3) == MergedRange(1, 2, 1, 3)
|
||||
assert sheet.text_at(2, 3) == ""
|
||||
assert sheet.merged_origin(2, 3) == (2, 3)
|
||||
assert sheet.merged_span(2, 3) is None
|
||||
|
||||
|
||||
def test_large_merged_range_is_resolved_without_cell_index() -> None:
|
||||
# Диапазон на весь столбец: индекс «ячейка → диапазон» для него не строится, поиск всё равно работает
|
||||
big = MergedRange(top=1, left=1, bottom=1_048_576, right=16_384)
|
||||
sheet = _sheet({(1, 1): Cell("Шапка", "text")}, merged=(big,))
|
||||
|
||||
assert sheet.merged_origin(500_000, 10_000) == (1, 1)
|
||||
assert sheet.text_at(1_048_576, 16_384) == "Шапка"
|
||||
|
||||
|
||||
def test_row_cells_are_ordered_by_column_and_skip_blank_text() -> None:
|
||||
sheet = _sheet(
|
||||
{
|
||||
(3, 5): Cell(Decimal("10.5"), "number"),
|
||||
(3, 1): Cell("60.01", "text"),
|
||||
(3, 3): Cell(" ", "text"),
|
||||
(4, 1): Cell("62", "text"),
|
||||
}
|
||||
)
|
||||
|
||||
assert sheet.row_cells(3) == [(1, Cell("60.01", "text")), (5, Cell(Decimal("10.5"), "number"))]
|
||||
assert sheet.row_cells(7) == []
|
||||
|
||||
|
||||
def test_row_of_spaces_is_empty() -> None:
|
||||
sheet = _sheet({(2, 1): Cell(" ", "text"), (2, 4): Cell("\xa0 ", "text"), (3, 1): Cell(False, "bool")})
|
||||
|
||||
assert sheet.is_row_empty(2)
|
||||
assert sheet.is_row_empty(10)
|
||||
assert not sheet.is_row_empty(3)
|
||||
|
||||
|
||||
def test_cell_text_strips_and_treats_spaces_as_empty() -> None:
|
||||
assert Cell(" ", "text").text == ""
|
||||
assert Cell(" Контрагент ", "text").text == "Контрагент"
|
||||
assert Cell(None, "empty").text == ""
|
||||
assert EMPTY_CELL.text == ""
|
||||
assert Cell(Decimal("1250.00"), "number").text == "1250.00"
|
||||
assert Cell(datetime.date(2024, 9, 30), "date").text == "2024-09-30"
|
||||
|
||||
|
||||
def test_outline_defaults_to_zero() -> None:
|
||||
sheet = _sheet({(1, 1): Cell("62", "text")}, row_outline={2: 1, 3: 2})
|
||||
|
||||
assert [sheet.outline(row) for row in (1, 2, 3)] == [0, 1, 2]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("index", "letter"),
|
||||
[(1, "A"), (26, "Z"), (27, "AA"), (52, "AZ"), (703, "AAA"), (16_384, "XFD")],
|
||||
)
|
||||
def test_column_letter(index: int, letter: str) -> None:
|
||||
assert column_letter(index) == letter
|
||||
|
||||
|
||||
def test_column_letter_rejects_non_positive_index() -> None:
|
||||
with pytest.raises(ValueError, match="1-based"):
|
||||
column_letter(0)
|
||||
181
api/tests/unit/parser/test_readers.py
Normal file
181
api/tests/unit/parser/test_readers.py
Normal file
@ -0,0 +1,181 @@
|
||||
"""Читатели книг: синтетическая книга openpyxl в памяти → `Sheet`; определение формата по сигнатуре; ошибки чтения."""
|
||||
|
||||
import datetime
|
||||
import io
|
||||
import zipfile
|
||||
from collections.abc import Iterator
|
||||
from decimal import Decimal
|
||||
from types import TracebackType
|
||||
from typing import NamedTuple
|
||||
|
||||
import openpyxl
|
||||
import pytest
|
||||
from openpyxl.styles import Alignment, Font
|
||||
|
||||
from src.contract import ErrorCode
|
||||
from src.parser.errors import ParserError, UnreadableError, UnsupportedFormatError
|
||||
from src.parser.grid import Cell, MergedRange
|
||||
from src.parser.readers import read_book
|
||||
from src.parser.readers import xlsb as xlsb_reader
|
||||
|
||||
|
||||
def _synthetic_xlsx() -> bytes:
|
||||
"""Две страницы: шапка с merged-заголовками, outline 0/1/2, bold, отступ, число, дата, bool, пустые строки."""
|
||||
book = openpyxl.Workbook()
|
||||
osv = book.active
|
||||
assert osv is not None
|
||||
osv.title = "ОСВ"
|
||||
osv["A1"] = "ООО Ромашка"
|
||||
osv["A1"].font = Font(bold=True)
|
||||
osv["A3"] = "Счет"
|
||||
osv.merge_cells("A3:A4")
|
||||
osv["B3"] = "Сальдо на начало периода"
|
||||
osv.merge_cells("B3:C3")
|
||||
osv["B4"] = "Дебет"
|
||||
osv["C4"] = "Кредит"
|
||||
osv["A5"] = "62"
|
||||
osv["A5"].font = Font(bold=True)
|
||||
osv["B5"] = 1500
|
||||
osv["A6"] = "62.01"
|
||||
osv["A6"].alignment = Alignment(indent=2)
|
||||
osv["B6"] = 1500.25
|
||||
osv.row_dimensions[6].outline_level = 1
|
||||
osv["A7"] = " Контрагент"
|
||||
osv["B7"] = datetime.datetime(2024, 9, 30)
|
||||
osv.row_dimensions[7].outline_level = 2
|
||||
osv["A9"] = " "
|
||||
osv["A10"] = "Итого"
|
||||
osv["C10"] = True
|
||||
other = book.create_sheet("Прочее")
|
||||
other["A1"] = "x"
|
||||
buffer = io.BytesIO()
|
||||
book.save(buffer)
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
def test_xlsx_reader_builds_sheet_model() -> None:
|
||||
sheets, warnings = read_book(_synthetic_xlsx(), "XLSX")
|
||||
|
||||
assert warnings == []
|
||||
assert [sheet.name for sheet in sheets] == ["ОСВ", "Прочее"]
|
||||
osv = sheets[0]
|
||||
assert osv.cell(1, 1) == Cell("ООО Ромашка", "text", bold=True)
|
||||
assert osv.cell(5, 1) == Cell("62", "text", bold=True)
|
||||
assert osv.cell(5, 2) == Cell(Decimal("1500"), "number")
|
||||
assert osv.cell(6, 1) == Cell("62.01", "text", indent=2)
|
||||
assert osv.cell(6, 2) == Cell(Decimal("1500.25"), "number")
|
||||
assert osv.cell(7, 1) == Cell(" Контрагент", "text", indent=3)
|
||||
assert osv.cell(7, 2) == Cell(datetime.datetime(2024, 9, 30), "date")
|
||||
assert osv.cell(10, 3) == Cell(True, "bool")
|
||||
assert osv.cell(9, 1).kind == "text"
|
||||
assert osv.is_row_empty(8)
|
||||
assert osv.is_row_empty(9)
|
||||
assert set(osv.merged) == {MergedRange(3, 1, 4, 1), MergedRange(3, 2, 3, 3)}
|
||||
assert osv.text_at(4, 1) == "Счет"
|
||||
assert osv.text_at(3, 3) == "Сальдо на начало периода"
|
||||
assert [osv.outline(row) for row in range(1, 11)] == [0, 0, 0, 0, 0, 1, 2, 0, 0, 0]
|
||||
assert (osv.n_rows, osv.n_cols) == (10, 3)
|
||||
assert all(cell.kind != "empty" for cell in osv.cells.values())
|
||||
assert sheets[1].row_cells(1) == [(1, Cell("x", "text"))]
|
||||
|
||||
|
||||
def test_file_type_mismatch_reads_by_signature_with_warning() -> None:
|
||||
sheets, warnings = read_book(_synthetic_xlsx(), "XLS")
|
||||
|
||||
assert [sheet.name for sheet in sheets] == ["ОСВ", "Прочее"]
|
||||
assert warnings == ["file_type=XLS, но содержимое OOXML: прочитано как XLSX"]
|
||||
|
||||
|
||||
def test_garbage_bytes_are_unsupported_format() -> None:
|
||||
with pytest.raises(UnsupportedFormatError) as caught:
|
||||
read_book(b"%PDF-1.7 not a workbook", "XLSX")
|
||||
|
||||
assert caught.value.code is ErrorCode.UNSUPPORTED_FORMAT
|
||||
assert isinstance(caught.value, ParserError)
|
||||
|
||||
|
||||
def test_truncated_zip_is_unreadable_with_one_line_reason() -> None:
|
||||
data = _synthetic_xlsx()
|
||||
|
||||
with pytest.raises(UnreadableError) as caught:
|
||||
read_book(data[: len(data) // 2], "XLSX")
|
||||
|
||||
assert caught.value.code is ErrorCode.UNREADABLE
|
||||
assert caught.value.message
|
||||
assert "\n" not in caught.value.message
|
||||
|
||||
|
||||
def test_truncated_biff_is_unreadable() -> None:
|
||||
with pytest.raises(UnreadableError):
|
||||
read_book(bytes.fromhex("D0CF11E0A1B11AE1") + b"\x00" * 64, "XLS")
|
||||
|
||||
|
||||
def _zip_with(name: str) -> bytes:
|
||||
buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(buffer, "w") as archive:
|
||||
archive.writestr(name, b"\x00")
|
||||
return buffer.getvalue()
|
||||
|
||||
|
||||
class _XlsbCell(NamedTuple):
|
||||
r: int
|
||||
c: int
|
||||
v: object
|
||||
|
||||
|
||||
class _FakeXlsbSheet:
|
||||
def __init__(self, rows: list[list[_XlsbCell]]) -> None:
|
||||
self._rows = rows
|
||||
|
||||
def __enter__(self) -> "_FakeXlsbSheet":
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc: object) -> None:
|
||||
return None
|
||||
|
||||
def rows(self, sparse: bool = False) -> Iterator[list[_XlsbCell]]:
|
||||
assert sparse
|
||||
yield from self._rows
|
||||
|
||||
|
||||
class _FakeXlsbBook:
|
||||
sheets = ("Лист1",)
|
||||
|
||||
def __enter__(self) -> "_FakeXlsbBook":
|
||||
return self
|
||||
|
||||
def __exit__(
|
||||
self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None
|
||||
) -> None:
|
||||
return None
|
||||
|
||||
def get_sheet(self, name: str) -> _FakeXlsbSheet:
|
||||
assert name == "Лист1"
|
||||
return _FakeXlsbSheet(
|
||||
[
|
||||
[_XlsbCell(0, 0, "Счет"), _XlsbCell(0, 1, None), _XlsbCell(0, 2, " ")],
|
||||
[_XlsbCell(2, 0, "62.01"), _XlsbCell(2, 1, 0.1), _XlsbCell(2, 2, False)],
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def test_xlsb_signature_is_read_by_pyxlsb_without_outline(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(xlsb_reader.pyxlsb, "open_workbook", lambda _stream: _FakeXlsbBook())
|
||||
|
||||
sheets, warnings = read_book(_zip_with("xl/workbook.bin"), "XLSB")
|
||||
|
||||
assert warnings == ["xlsb: outline level недоступен"]
|
||||
(sheet,) = sheets
|
||||
assert sheet.name == "Лист1"
|
||||
assert sheet.cell(1, 1) == Cell("Счет", "text")
|
||||
assert sheet.cell(1, 3) == Cell(" ", "text")
|
||||
assert sheet.cell(3, 2) == Cell(Decimal("0.1"), "number")
|
||||
assert sheet.cell(3, 3) == Cell(False, "bool")
|
||||
assert sheet.merged == ()
|
||||
assert sheet.row_outline == {}
|
||||
assert (sheet.n_rows, sheet.n_cols) == (3, 3)
|
||||
|
||||
|
||||
def test_zip_without_workbook_bin_goes_to_openpyxl_and_is_unreadable() -> None:
|
||||
with pytest.raises(UnreadableError):
|
||||
read_book(_zip_with("hello.txt"), "XLSX")
|
||||
@ -143,9 +143,9 @@ def read_book(data: bytes, file_type: FileType) -> tuple[tuple[Sheet, ...], list
|
||||
```
|
||||
Числа: читатели кладут `Decimal(str(value))` для `float`/`int` (без квантования — это делает `numbers.py`), bool отдельно (`kind="bool"`), даты — как есть. xlsx: `load_workbook(BytesIO(data), data_only=True, read_only=False)`; outline — `ws.row_dimensions[r].outline_level`; bold — `cell.font.bold`; indent — `cell.alignment.indent`; merged — `ws.merged_cells.ranges`. xls: `xlrd.open_workbook(file_contents=data, formatting_info=True)`; outline — `sheet.rowinfo_map[r].outline_level`; bold/indent — через `book.xf_list[cell.xf_index]` (`font.bold`, `alignment.indent_level`); merged — `sheet.merged_cells` (полуоткрытые → включительные). xlsb: `pyxlsb.open_workbook(BytesIO)`; outline/merged недоступны → пустые, предупреждение `"xlsb: outline level недоступен"`.
|
||||
|
||||
- [ ] **Step 1: Тесты.** `test_grid.py`: `text_at` в merged-диапазоне берёт origin; `row_cells` упорядочен; `column_letter(1/26/27/52/703)`; `Cell(" ").text == ""`. `test_readers.py`: синтетическая книга, собранная openpyxl в памяти (2 листа, merged заголовок, outline 0/1/2, bold, indent, число, текст, дата, пустая строка) → `read_book` даёт ожидаемую `Sheet`; та же книга, сохранённая как xls невозможна без Excel — для xls используется реальный файл в `test_readers_real.py`; подмена `file_type` (XLSX при BIFF8) → предупреждение; мусорные байты → `UnsupportedFormatError`; обрезанный zip → `UnreadableError`. `test_readers_real.py` (`fixtures`): все 66 файлов читаются < 5 с, у каждого ≥ 1 лист, есть строки с outline ≥ 1, merged > 0; `xls` `…62 за 3 квартал 2024 г..xls` — outline строки 15 (1-based) равен 3.
|
||||
- [ ] **Step 2:** падает → **Step 3:** реализация → **Step 4:** `pytest tests/unit/parser tests/fixtures/test_readers_real.py -v`, `make check` → PASS.
|
||||
- [ ] **Step 5: Commit** `feat(parser): модель листа Sheet и читатели xlsx/xls/xlsb`.
|
||||
- [x] **Step 1: Тесты.** `test_grid.py`: `text_at` в merged-диапазоне берёт origin; `row_cells` упорядочен; `column_letter(1/26/27/52/703)`; `Cell(" ").text == ""`. `test_readers.py`: синтетическая книга, собранная openpyxl в памяти (2 листа, merged заголовок, outline 0/1/2, bold, indent, число, текст, дата, пустая строка) → `read_book` даёт ожидаемую `Sheet`; та же книга, сохранённая как xls невозможна без Excel — для xls используется реальный файл в `test_readers_real.py`; подмена `file_type` (XLSX при BIFF8) → предупреждение; мусорные байты → `UnsupportedFormatError`; обрезанный zip → `UnreadableError`. `test_readers_real.py` (`fixtures`): все 66 файлов читаются < 5 с, у каждого ≥ 1 лист, есть строки с outline ≥ 1, merged > 0; `xls` `…62 за 3 квартал 2024 г..xls` — outline строки 15 (1-based) равен 3.
|
||||
- [x] **Step 2:** падает → **Step 3:** реализация → **Step 4:** `pytest tests/unit/parser tests/fixtures/test_readers_real.py -v`, `make check` → PASS.
|
||||
- [x] **Step 5: Commit** `feat(parser): модель листа Sheet и читатели xlsx/xls/xlsb`.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -13,7 +13,7 @@
|
||||
- [x] 1.2 Контракт v1.1: скопировать `../docs/contracts/recognition-result.schema.json` v1.1 в `api/contracts/`, `OsvRow.extra:
|
||||
dict[str, str | None] | None`, `CONTRACT_VERSION = "1.1"`; примеры docs проходят; тест идентичности зелёный; воркер пишет
|
||||
`result_schema_version = "1.1"` (`test_worker` обновлён). Проверка: `make check`.
|
||||
- [ ] 1.3 `grid.py` и читатели: `Cell`, `Sheet`, `MergedRange`, `column_letter`; `readers/xlsx.py` (openpyxl, полное чтение,
|
||||
- [x] 1.3 `grid.py` и читатели: `Cell`, `Sheet`, `MergedRange`, `column_letter`; `readers/xlsx.py` (openpyxl, полное чтение,
|
||||
`outline_level`, `alignment.indent`, bold), `readers/xls.py` (xlrd `formatting_info`, `rowinfo_map`, XF bold/indent),
|
||||
`readers/xlsb.py` (pyxlsb, без outline), `readers/__init__.py:read_book(data, file_type) -> tuple[Sheet, ...]` с
|
||||
определением по сигнатуре и предупреждением при несоответствии; ошибки → `UnreadableError`/`UnsupportedFormatError`.
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user