"""Чтение книги 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` с причиной одной строкой. Лимиты фазы чтения (`_limits.py`, ruling R-23) проверяются до библиотеки чтения: распакованный размер zip (`max_uncompressed_bytes`, по умолчанию `ZIP_UNCOMPRESSED_LIMIT_FACTOR × PARSE_MAX_BYTES` из настроек по умолчанию) и площадь объединённых ячеек xlsx; после чтения — площадь объединённых ячеек любого формата. """ import io import zipfile from collections.abc import Callable from src.contract import FileType from src.core.config import Settings 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._limits import ( ZIP_UNCOMPRESSED_LIMIT_FACTOR, check_merged_area, check_xlsx_merged_area, check_zip_uncompressed_size, ) 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__ = ["default_uncompressed_limit", "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 default_uncompressed_limit() -> int: """Лимит распакованного zip без явного параметра: `ZIP_UNCOMPRESSED_LIMIT_FACTOR × PARSE_MAX_BYTES` по умолчанию.""" max_bytes: int = Settings.model_fields["PARSE_MAX_BYTES"].default return ZIP_UNCOMPRESSED_LIMIT_FACTOR * max_bytes def read_book( data: bytes, file_type: FileType, *, max_uncompressed_bytes: int | None = None ) -> tuple[tuple[Sheet, ...], list[str]]: """Листы книги и предупреждения чтения (несовпадение `file_type`, ограничения xlsb). `max_uncompressed_bytes` — лимит распакованного zip (xlsx/xlsb); `OsvParser` передаёт `ZIP_UNCOMPRESSED_LIMIT_FACTOR × max_bytes`. Превышение лимитов чтения — `UnsupportedFormatError`. """ limit = default_uncompressed_limit() if max_uncompressed_bytes is None else max_uncompressed_bytes detected = _detect(data, file_type, limit) 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 check_merged_area(sheets) if detected == "XLSB": warnings.append(OUTLINE_UNAVAILABLE_WARNING) return sheets, warnings def _detect(data: bytes, file_type: FileType, uncompressed_limit: int) -> FileType: """Формат по сигнатуре; для zip — заодно лимиты распакованного размера и (xlsx) объединённых ячеек.""" if data.startswith(OLE2_SIGNATURE): return "XLS" if data.startswith(ZIP_SIGNATURE): try: with zipfile.ZipFile(io.BytesIO(data)) as archive: check_zip_uncompressed_size(archive, uncompressed_limit) detected: FileType = "XLSB" if XLSB_WORKBOOK_PART in set(archive.namelist()) else "XLSX" if detected == "XLSX": check_xlsx_merged_area(archive) except ParserError: raise except Exception as exc: # zipfile, zlib.error, NotImplementedError (метод сжатия), RuntimeError (пароль)… raise UnreadableError(f"повреждённый zip-архив: {one_line_reason(exc)}") from exc return detected raise UnsupportedFormatError(f"содержимое файла не XLS/XLSX/XLSB (file_type={file_type})")