onco-cola-utils 0.5.5

Мои общие утилиты для работы с Excel

Onco-Cola's utils

onco-cola-utils — это набор утилит для Python, предназначенный для упрощения работы с данными, особенно с файлами Excel. Библиотека включает инструменты для чтения, записи и обработки данных, а также вспомогательные модули для логирования, создания типизированных констант и форматированного вывода.

Основные возможности

• ReaderController: Мощный класс для управления файлами Excel (XLS/XLSX). Позволяет читать данные, находить нужный лист, обновлять и сохранять файлы, а также добавлять уникальные идентификаторы к строкам.
• KeyParamClass: Метакласс для создания неизменяемых, enum-подобных классов. Идеально подходит для определения констант, статусов или других предопределенных значений в коде.
• pretty_print: Утилита для красивого вывода сложных объектов (dataclasses, словарей, списков), что упрощает отладку.
• logger: Преднастроенный логгер на базе loguru с цветовой кодировкой для разных уровней логирования (DEBUG, INFO, SUCCESS, ERROR).
• value_to_bool: Функция для преобразования различных значений в булев тип по расширенным правилам.
• DefaultState: Класс, предназначенный для типизированного ответа из любой функции
• JsonReader: Класс, предназначенный для чтения/записи json-файлов

Установка

Для установки зависимостей выполните команду:

pip install onco-cola-utils

Важно: Перед запуском скриптов, работающих с Excel-файлами, убедитесь, что эти файлы не открыты в других программах, чтобы избежать ошибок доступа (PermissionError).

---

API-документация

ReaderController

Класс ReaderController предоставляет полный набор инструментов для взаимодействия с Excel-файлами. Он позволяет читать данные из листов, управлять их содержимым, проверять доступность файлов и сохранять изменения. Класс разработан с учетом удобства использования и автоматизации рутинных операций с табличными данными.

Инициализация

ReaderController(file_path: Path, file_output: Path, is_new: bool = False, skip_rows: int = 0)

• file_path (Path): Путь к входному Excel-файлу.
• file_output (Path): Путь к выходному Excel-файлу. Может совпадать с file_path.
• is_new (bool, по умолчанию False): Если True, файл считается новым и не проверяется на существование при инициализации.
• skip_rows (int, по умолчанию 0): Количество строк, которые нужно пропустить при чтении данных. (На данный момент не полностью реализовано в методах чтения).

Пример:

from onco_cola_utils.reader_controller.core import ReaderController
from pathlib import Path

input_excel = Path("data/input.xlsx")
output_excel = Path("data/output.xlsx")

reader = ReaderController(file_path=input_excel, file_output=output_excel)

Методы

read_data(sheet_name: Optional[str] = None) -> None

Читает данные из указанного листа Excel-файла и сохраняет их во внутренний датафрейм (_dataframe). Все значения преобразуются в строки, а числа с плавающей точкой, являющиеся целыми, форматируются без десятичной части (например, 1.0 становится 1).

• sheet_name (str | None, по умолчанию None): Имя листа для чтения. Если None, читается первый лист.

Пример:

reader.read_data(sheet_name="Sheet1")
# Или для первого листа:
reader.read_data()

get_data(sheet_name: Optional[str] = None) -> list[dict]

Лениво загружает данные из Excel-файла, если они еще не были загружены. Возвращает текущий внутренний датафрейм.

• sheet_name (str | None, по умолчанию None): Имя листа для чтения, если данные еще не загружены.

Возвращает:

• list[dict]: Список словарей, где каждый словарь представляет строку данных, а ключи — имена столбцов.

Пример:

data = reader.get_data()
for row in data:
    print(row)

perfect_data(data_list: DFType) -> dict

Фильтрует входной список данных, исключая строки, в которых значение в столбце ColumnStrings.DATA_ENTITY_TOBE соответствует одному из "нулевых" значений, определенных в System.NULLED.

• data_list (DFType): Список словарей, представляющих данные.

Возвращает:

• dict: Словарь, где ключами являются local_id строк, а значениями — отфильтрованные данные.

Пример:

clean_data = reader.perfect_data(reader.get_data())

local_idfy(data_list: DFType) -> dict

Присваивает уникальные идентификаторы (local_id) строкам данных на основе уже существующих local_id в данных. Этот метод создает внутреннее представление данных, индексированное по local_id.

• data_list (DFType): Список словарей, содержащих данные, которые нужно индексировать.

Возвращает:

• dict: Словарь, где ключами являются local_id, а значениями — соответствующие строки данных.

Пример:

indexed_data = reader.local_idfy(reader.get_data())

cycle_right_sheet() -> None

Автоматически определяет и выбирает правильный лист в Excel-файле. Метод итерируется по листам, удаляя первый лист и пересохраняя файл, пока не найдет лист, содержащий обязательные поля (source_name и url). Это полезно, если файл содержит служебные или пустые листы в начале.

Пример:

# Обычно вызывается автоматически при проверке local_id или чтении данных
reader.cycle_right_sheet()

check_local_id(find_it: bool = True) -> bool

Проверяет наличие столбца local_id в данных. Перед проверкой вызывает cycle_right_sheet() для определения корректного листа.

• find_it (bool, по умолчанию True): Если True, метод проверяет именно наличие local_id. Если False, метод просто убеждается, что лист корректен и содержит данные.

Возвращает:

• bool: True, если local_id найден (или лист корректен, если find_it=False), иначе False.

Пример:

if not reader.check_local_id():
    print("Столбец local_id не найден, будет добавлен.")
    reader.process_local_idfying()

process_local_idfying(field: str = ColumnStrings.DATA_LOCAL_ID) -> None

Добавляет или обновляет столбец local_id в данных, присваивая последовательные целочисленные значения (начиная с 1). После обработки данные сохраняются обратно в файл.

• field (str, по умолчанию ColumnStrings.DATA_LOCAL_ID): Имя столбца, который будет использоваться для local_id.

Пример:

reader.process_local_idfying()

update_dataframe_from_updated_dataframe(updated_dataframe: dict, updated_fields: list[str], field_id: str = "ID") -> Optional[bool]

Обновляет основной датафрейм (_dataframe) на основе предоставленных обновленных данных. Этот метод используется для слияния результатов обработки (например, ответа от ИИ-модели) с исходными данными.

• updated_dataframe (dict): Словарь, где ключи — это ID строк, а значения — словари с обновленными полями.
• updated_fields (list[str]): Список имен полей, которые должны быть обновлены.
• field_id (str, по умолчанию "ID"): Имя поля, используемого в качестве идентификатора строки в updated_dataframe.

Возвращает:

• Optional[bool]: True в случае успешного обновления, None если updated_dataframe или updated_fields пусты.

Пример:

ai_results = {
    1: {"category_asis": "Electronics", "remark": "1"},
    2: {"category_asis": "Books", "remark": None},
}
reader.update_dataframe_from_updated_dataframe(
    updated_dataframe=ai_results,
    updated_fields=["category_asis", "remark"]
)

update_file(same_file: bool = True) -> bool

Сохраняет текущее состояние внутреннего датафрейма (_dataframe) обратно в Excel-файл.

• same_file (bool, по умолчанию True): Если True, данные сохраняются в file_path. Если False, данные сохраняются в file_output.

Возвращает:

• bool: True в случае успешного сохранения.

Пример:

reader.update_file() # Сохранить в исходный файл
reader.update_file(same_file=False) # Сохранить в выходной файл

save_to_csv(same_file: bool = True) -> bool

Сохраняет текущее состояние внутреннего датафрейма в CSV-файл. Расширение файла должно быть .csv в file_path или file_output.

• same_file (bool, по умолчанию True): Если True, данные сохраняются в file_path. Если False, данные сохраняются в file_output.

Возвращает:

• bool: True в случае успешного сохранения.

Пример:

# Предполагается, что reader._file_output установлен на Path("data/output.csv")
reader.save_to_csv(same_file=False)

rename(new_name: str, same_file: bool = True) -> bool

Переименовывает входной или выходной файл Excel.

• new_name (str): Новое базовое имя файла (без расширения).
• same_file (bool, по умолчанию True): Если True, переименовывается file_path. Если False, переименовывается file_output.

Возвращает:

• bool: True в случае успешного переименования.

Пример:

reader.rename("processed_data")

get_asis_fields() -> list[str]

Возвращает список всех имен столбцов из текущего датафрейма, которые содержат подстроку _asis.

Возвращает:

• list[str]: Отсортированный список имен столбцов.

Пример:

asis_columns = reader.get_asis_fields()
print(asis_columns)

get_tobe_fields() -> list[str]

Возвращает список всех имен столбцов из текущего датафрейма, которые содержат подстроку _tobe.

Возвращает:

• list[str]: Отсортированный список имен столбцов.

Пример:

tobe_columns = reader.get_tobe_fields()
print(tobe_columns)

idfy_to_dataframe(idfy_data: IdfyGoods) -> list[dict]

Преобразует данные из формата IdfyGoods (словарь, индексированный по ID) обратно в список словарей, пригодный для использования с pandas.DataFrame.

• idfy_data (IdfyGoods): Словарь данных, индексированный по ID.

Возвращает:

• list[dict]: Список словарей, представляющих данные.

Пример:

# Предположим, у нас есть данные в формате IdfyGoods
# indexed_data = reader.local_idfy(reader.get_data())
# df_format_data = reader.idfy_to_dataframe(indexed_data)

Свойства

• rows_total (int): Общее количество строк в данных, включая заголовки (если применимо).
• rows_data (int): Количество строк с данными (без заголовков).
• get_sheet_names() -> list[str]: Возвращает список имен всех листов в Excel-файле.

KeyParamClass

KeyParamClass — это базовый класс, который позволяет создавать неизменяемые, типобезопасные константы, похожие на Enum, но с дополнительными атрибутами name (короткое строковое имя) и desc (подробное описание). Это удобно для определения фиксированных наборов значений, таких как статусы, типы или категории, где помимо самого значения требуется и его человекочитаемое описание.

Использование

Для создания набора констант, унаследуйте свой класс от KeyParamClass и определите константы как кортежи (name: str, desc: str).

from onco_cola_utils.key_param_class.core import KeyParamClass

class Status(KeyParamClass):
    PENDING = ("pending", "Ожидает обработки")
    ACTIVE = ("active", "Активен")
    COMPLETED = ("completed", "Завершено")
    CANCELLED = ("cancelled", "Отменено")

# Доступ к константам
print(Status.PENDING)          # Вывод: pending (возвращает .name по умолчанию)
print(Status.ACTIVE.name)      # Вывод: active
print(Status.COMPLETED.desc)   # Вывод: Завершено

# Сравнение
print(Status.PENDING == "pending") # Вывод: True
print(Status.ACTIVE == Status.ACTIVE) # Вывод: True

# Итерация по всем константам
for status in Status:
    print(f"Имя: {status.name}, Описание: {status.desc}")

# Получение константы по имени или атрибуту
print(Status.from_name("active")) # Вывод: <Status.ACTIVE: active (Активен)>
print(Status.get("COMPLETED"))    # Вывод: <Status.COMPLETED: completed (Завершено)>

# Получение всех констант в виде списка кортежей (name, desc)
print(Status.choices) # Вывод: [("pending", "Ожидает обработки"), ("active", "Активен"), ...]

# Получение всех констант в виде словаря
print(Status.dict()) # Вывод: {"PENDING": {"name": "pending", "desc": "Ожидает обработки"}, ...}

Особенности

• Неизменяемость: После создания экземпляры KeyParamClass и сам класс являются неизменяемыми. Попытка изменить их атрибуты вызовет AttributeError.
• Проверка дубликатов: Метакласс проверяет уникальность name (кода) при определении констант. Дубликаты desc (описания) вызовут предупреждение, но не ошибку.
• Enum-like API: Поддерживает методы, схожие с Enum, такие как итерация, доступ по ключу (Status["PENDING"]), get(), from_name(), choices и dict().

---

pretty_print

Модуль pretty_print предоставляет функцию для форматированного вывода сложных структур данных, таких как dataclasses, словари и списки. Это значительно улучшает читаемость вывода в консоли и упрощает отладку.

Функция pretty_print

pretty_print(obj, indent: int = 4, title: str = 'PRETTY_PRINT', m2d: bool = False, outputter=log)

• obj: Объект, который нужно вывести. Может быть dataclass, словарем, списком или другим типом.
• indent (int, по умолчанию 4): Количество пробелов для каждого уровня отступа.
• title (str, по умолчанию 'PRETTY_PRINT'): Заголовок, который будет выведен перед форматированным объектом.
• m2d (bool, по умолчанию False): Если True, функция попытается преобразовать объект в словарь с помощью model_to_dict перед форматированием. Это полезно для объектов ORM или других моделей, которые могут быть преобразованы в словари.
• outputter (callable, по умолчанию log): Функция, которая будет использоваться для вывода строк. По умолчанию используется функция log из встроенного модуля logger.

Пример:

from dataclasses import dataclass
from onco_cola_utils.pretty_print.core import pretty_print

@dataclass
class User:
    name: str
    age: int
    contacts: dict

user_data = User(
    name="Alice",
    age=30,
    contacts={
        "email": "[e-mail]",
        "phone": "123-456-7890"
    }
)

pretty_print(user_data, title="User Profile")

# Вывод:
# User Profile
# User(
#     name='Alice',
#     age=30,
#     contacts={
#         'email': '[e-mail]',
#         'phone': '123-456-7890',
#     },
# )

pretty_print({"item1": 10, "item2": ["a", "b"]}, title="My Dictionary")

# Вывод:
# My Dictionary
# dict(
#     'item1': 10,
#     'item2': [
#         'a',
#         'b',
#     ],
# )

---

logger

Модуль logger предоставляет удобный и настраиваемый интерфейс для логирования событий в приложении, используя библиотеку loguru. Он преднастроен для вывода сообщений разных уровней с цветовой кодировкой в стандартный поток ошибок (sys.stderr).

Уровни логирования и функции

Модуль предоставляет следующие функции для логирования:

Уровень	Функция	Описание
DEBUG	log	Для отладочной информации.
INFO	loginf	Для общей информации о ходе выполнения.
SUCCESS	logsuc	Для сообщений об успешном выполнении операций.
ERROR	logerr	Для сообщений об ошибках.

Пример:

from onco_cola_utils.logger.core import log, loginf, logsuc, logerr

log("Это отладочное сообщение.")
loginf("Приложение запущено.")
logsuc("Операция успешно завершена!")
logerr("Произошла критическая ошибка.")

---

value_to_bool

Функция value_to_bool предоставляет надежный способ преобразования различных типов значений (строк, чисел, булевых) в булево значение, используя расширенный набор правил.

Функция value_to_bool

value_to_bool(value: Any) -> bool

• value: Любое значение, которое нужно преобразовать в булево.

Правила преобразования:

• True: True, 1, "true", "1", "yes", "on", "+", "ok", "да", "вкл", "y", "t", "enable", "enabled" (регистронезависимо).
• False: False, 0, None, "" (пустая строка), " " (строка из пробелов), "false", "0", "no", "off", "-", "нет", "выкл", "n", "f", "disable", "disabled", "null", "none" (регистронезависимо).
• Числа с плавающей точкой: 0.0 -> False, любое другое число -> True.
• Любое другое значение, которое не соответствует правилам True или False, будет преобразовано в False.

Пример:

from onco_cola_utils.value_to_bool import value_to_bool

print(value_to_bool("true"))    # True
print(value_to_bool(1))         # True
print(value_to_bool("да"))      # True
print(value_to_bool("false"))   # False
print(value_to_bool(0))         # False
print(value_to_bool(None))      # False
print(value_to_bool(""))        # False
print(value_to_bool("random_text")) # False

---

Конфигурация

Библиотека использует несколько конфигурационных файлов для определения системных констант и строковых значений столбцов. Эти константы централизуют часто используемые значения и имена полей, обеспечивая единообразие и упрощая поддержку кода.

System (в src/onco_cola_utils/configs/system.py)

Класс System содержит общие системные константы, используемые в приложении.

• NULLED: Список значений, которые считаются "нулевыми" или пустыми при обработке данных (например, ["", "0", "none"]).
• FULL_SKIP: Список значений, при наличии которых строка данных должна быть полностью пропущена.
• ID: Имя поля по умолчанию, используемое в качестве идентификатора строки (по умолчанию "ID").

Пример использования:

from onco_cola_utils.configs.system import System

if data_item.get(ColumnStrings.DATA_ENTITY_TOBE) in System.NULLED:
    # Обработка нулевых значений
    pass

ColumnStrings (в src/onco_cola_utils/configs/column_strings.py)

Класс ColumnStrings определяет стандартизированные имена для различных столбцов, используемых в Excel-файлах и внутренних структурах данных. Это помогает избежать "магических строк" и обеспечивает консистентность в именовании полей.

• DATA_LOCAL_ID: Имя столбца для локального идентификатора строки (по умолчанию "local_id").
• DATA_SOURCE_NAME: Имя столбца, содержащего исходное имя или описание данных (по умолчанию "source_name").
• DATA_URL: Имя столбца для URL-адресов (по умолчанию "url").
• RMK: Имя столбца для заметок или пометок (по умолчанию "remark").
• DATA_ENTITY_TOBE: Имя столбца, используемого для проверки на "нулевое" значение (по умолчанию "category_tobe").

Пример использования:

from onco_cola_utils.configs.column_strings import ColumnStrings

local_id_field = ColumnStrings.DATA_LOCAL_ID
print(f"Имя поля для локального ID: {local_id_field}")

---

DefaultState

Универсальный класс-обертка для стандартизированного представления состояния операций в приложениях. Реализует паттерн "состояние результата" с автоматическим отслеживанием временных меток и гибкой типизацией данных.

Особенности

• Генерализованный ответ: Единый формат для успешных и ошибочных результатов операций
• Автоматическое временное отслеживание: Мета-информация о создании и обновлении состояния
• Гибкая типизация: Поддержка generic-типов для строгой типизации поля data
• Булева семантика: Возможность использования в условных выражениях через __bool__
• Несколько способов инициализации: Из словаря, ключевых аргументов или пустая инициализация
• Цепочка вызовов: Методы обновления возвращают self для fluent-интерфейса

Быстрый старт

from datetime import datetime
from typing import Optional
from your_module import DefaultState

# Создание пустого состояния
state = DefaultState()
print(state.result)  # False
print(state.status)  # INIT

# Обновление состояния
state.update(result=True, status="SUCCESS", detail="Operation completed")
print(bool(state))  # True

# Использование с типизированными данными
user_data = {"id": 1, "name": "John"}
user_state = DefaultState[dict]()
user_state.success(data=user_data, status="USER_FOUND")

# Проверка результата
if user_state:
    print(f"User: {user_state.data['name']}")
    print(f"Created: {user_state.created_dt}")

API Reference

DefaultState Class

Универсальный контейнер состояния операции.

__init__(init_data: Optional[dict] = None, **kwargs) -> None

Инициализирует объект состояния. Поддерживает три способа инициализации:

Параметры:

• init_data (Optional[dict]): Словарь с начальными значениями полей
• **kwargs: Ключевые аргументы для прямого присвоения полям

Примеры:

# Из словаря
state1 = DefaultState({'result': True, 'status': 'OK'})

# Из ключевых аргументов  
state2 = DefaultState(result=False, status='ERROR', detail='Not found')

# Пустая инициализация
state3 = DefaultState()

update(result: bool = False, status: Optional[AllowedTypes] = None, detail: Optional[AllowedTypes] = None, data: Optional[T] = None) -> DefaultState[T]

Обновляет указанные поля состояния и временную метку.

Параметры:

• result (bool): Результат операции (по умолчанию False)
• status (Optional[AllowedTypes]): Статус операции
• detail (Optional[AllowedTypes]): Детальное описание
• data (Optional[T]): Произвольные данные

Возвращает:

• DefaultState[T]: self для цепочки вызовов

Пример:

state.update(result=True, status="PROCESSING", data={"progress": 50})

success(status: Optional[AllowedTypes] = None, detail: Optional[AllowedTypes] = None, data: Optional[T] = None) -> DefaultState[T]

Специализированный метод для установки успешного результата.

Параметры:

• status (Optional[AllowedTypes]): Статус успешной операции
• detail (Optional[AllowedTypes]): Детальное описание
• data (Optional[T]): Произвольные данные

Возвращает:

• DefaultState[T]: self для цепочки вызовов

Пример:

state.success(status="COMPLETED", detail="User created", data=user_id)

insert(data: dict) -> DefaultState[T]

Алиас для update() с передачей словаря (обратная совместимость).

Параметры:

• data (dict): Словарь с данными для обновления

Возвращает:

• DefaultState[T]: self для цепочки вызовов

Пример:

state.insert({"result": True, "status": "UPDATED"})

__bool__() -> bool

Реализует булев контекст. Возвращает значение result.

Возвращает:

• bool: True если операция успешна

Пример:

if state:
    # Действия при успешном результате
    process_data(state.data)

Свойства

result: Optional[bool]

Основной результат операции (True/False).

status: Optional[AllowedTypes]

Краткий статус операции.

detail: Optional[AllowedTypes]

Детальное описание результата или ошибки.

data: Optional[T]

Произвольные данные, возвращаемые операцией.

created_at: int

Тimestamp создания состояния (короткий доступ к meta.created_at).

updated_at: int

Timestamp последнего обновления (короткий доступ к meta.updated_at).

created_dt: datetime

Объект datetime создания состояния.

updated_dt: datetime

Объект datetime последнего обновления.

MetaState Class

Вспомогательный класс для хранения мета-информации.

created_at: int

Timestamp создания состояния.

updated_at: int

Timestamp последнего обновления.

Конфигурация

Класс не требует внешней конфигурации. Все настройки задаются через параметры инициализации или методы обновления.

Рекомендации

• Используйте success() для положительных сценариев.
• Не используйте data для передачи метаданных — лучше вынести в отдельные поля.
• При работе с JSON — убедитесь, что data сериализуемо.

---

💡 Совет: Этот класс отлично подходит для шаблонизации ответов в микросервисах, ETL-процессах, AI-обработчиках и CLI-утилитах.

---

JsonReader

Умный читатель и валидатор JSON-файлов с поддержкой потокового чтения и расширенной мета-информацией. Предоставляет безопасный интерфейс для работы с JSON-файлами любого размера.

Особенности

• 🔍 Валидация синтаксиса — проверка корректности JSON перед загрузкой
• 📊 Мета-информация — размер, хеш, даты создания/изменения, количество строк
• ⚡ Два режима работы — обычный (до ~50 МБ) и потоковый (для больших файлов)
• 🔄 Ленивая загрузка — данные читаются только при первом обращении
• 📝 Статические методы записи — сериализация с поддержкой Pydantic, Path и сложных структур
• 🔒 Атомарная запись — предотвращение частичной записи при сбоях

Быстрый старт

Базовое использование

from json_reader import JsonReader

# Создание читателя
reader = JsonReader("data.json")

# Проверка валидности
if reader.is_valid:
    print(f"Файл валиден, размер: {reader.size_mb} МБ")
    
    # Ленивая загрузка данных
    data = reader.data
    print(f"Тип данных: {reader.data_type}")
    print(f"Количество элементов: {reader.length}")
else:
    print(f"Ошибка: {reader.validation_error}")

# Полная информация о файле
print(reader.info())

Потоковый режим для больших файлов

# Для файлов больше 50 МБ
reader = JsonReader("large_data.json", use_stream=True)

if reader.is_valid:
    # Итеративная обработка без полной загрузки в память
    for item in reader.iter_items():
        process_item(item)

Запись JSON-файлов

data = {"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}

# Красивая запись с форматированием
JsonReader.write_pretty(data, "output.json")

# Компактная запись
JsonReader.write_compact(data, "compact_output.json")

# Кастомная запись
JsonReader.write(data, "custom.json", indent=4, ensure_ascii=True)

API Reference

Конструктор

JsonReader(file_path: Union[str, Path], use_stream: bool = False)

Создает экземпляр читателя JSON-файла.

Параметры:

• file_path — путь к JSON-файлу
• use_stream — использовать потоковую валидацию (для файлов >50 МБ)

Исключения:

• ValueError — если use_stream=True, но ijson не установлен
• FileNotFoundError — если файл не существует

Свойства файла

path: Path

Полный путь к файлу.

name: str

Имя файла с расширением.

stem: str

Имя файла без расширения.

size_bytes: int

Размер файла в байтах.

size_mb: float

Размер файла в мегабайтах.

created_at: Optional[datetime]

Дата создания файла.

modified_at: Optional[datetime]

Дата последнего изменения.

sha256: str

SHA-256 хеш содержимого файла.

Валидация и данные

is_valid: bool

Проверяет валидность JSON-файла.

validation_error: Optional[str]

Сообщение об ошибке валидации.

data: Any

Лениво загружает и возвращает содержимое JSON.

Исключения:

• ValueError — при ошибках парсинга JSON
• OSError — при ошибках чтения файла

data_type: Optional[str]

Тип корневого элемента ('dict', 'list', 'str' и т.д.).

length: Optional[int]

Длина корневого элемента (для list/dict/str).

Методы

validate_structure(required_keys: Optional[List[str]] = None) -> bool

Проверяет наличие обязательных ключей в JSON-объекте.

Параметры:

• required_keys — список обязательных ключей

Исключения:

• ValueError — если файл невалиден или корень не объект

# Проверка структуры
if reader.validate_structure(["users", "metadata"]):
    print("Структура соответствует требованиям")

iter_items(prefix: str = '') -> Generator[Any, None, None]

Потоковое чтение элементов (требует use_stream=True).

# Для массивов
for item in reader.iter_items():
    print(item)

# Для объектов
for key, value in reader.iter_items():
    print(f"{key}: {value}")

info() -> Dict[str, Any]

Возвращает полную мета-информацию о файле.

info = reader.info()
print(f"""
Path: {info['path']}
Valid: {info['valid']}
Type: {info['type']}
Size: {info['size_mb']} MB
SHA256: {info['sha256']}
""")

Статические методы записи

JsonReader.write(data, file_path, indent=2, ensure_ascii=False, create_parents=True, atomic=True)

Универсальный метод записи JSON с поддержкой сложных структур.

Поддерживает:

• Pydantic модели (v1/v2)
• Объекты Path
• Словари с int-ключами
• Множества (set)

JsonReader.write_pretty(data, file_path)

Форматированная запись с отступами.

JsonReader.write_compact(data, file_path)

Компактная запись без форматирования.

Примеры использования

Мониторинг изменений файла

reader1 = JsonReader("config.json")
original_hash = reader1.sha256

# После некоторых операций...
reader2 = JsonReader("config.json")
if reader2.sha256 != original_hash:
    print("Файл был изменен!")

Пакетная обработка JSON-файлов

from pathlib import Path

json_files = Path("data/").glob("*.json")

for file_path in json_files:
    reader = JsonReader(file_path)
    if reader.is_valid and reader.data_type == "list":
        print(f"{file_path.name}: {reader.length} элементов")

Валидация конфигурационных файлов

def validate_config(file_path: str) -> bool:
    reader = JsonReader(file_path)
    
    if not reader.is_valid:
        print(f"Invalid JSON: {reader.validation_error}")
        return False
    
    try:
        return reader.validate_structure(["version", "settings", "users"])
    except ValueError as e:
        print(f"Structure error: {e}")
        return False

Особенности работы

Режимы валидации

• Обычный режим (use_stream=False) — полная загрузка в память, быстрая проверка
• Потоковый режим (use_stream=True) — постепенная проверка без полной загрузки

Производительность

Размер файла	Рекомендуемый режим	Требования
До 50 МБ	Обычный	Стандартная библиотека
50+ МБ	Потоковый	pip install ijson

Обработка ошибок

Все методы корректно обрабатывают:

• Отсутствующие файлы
• Невалидный JSON
• Проблемы с кодировкой
• Ошибки доступа

---

---

Зависимости

Проект onco-cola-tools использует следующие основные библиотеки:

• pandas (>=1.5): Для работы с табличными данными и Excel-файлами.
• openpyxl (>=3.0): Бэкенд для чтения и записи .xlsx файлов в pandas.
• loguru (^0.7.3): Для расширенного и удобного логирования.
• deprecated (>=1.2): Для маркировки устаревших функций и методов.
• build (>=1.3.0): Для сборки пакета.
• twine (>=6.1.0): Для публикации пакета.
• tiktoken (>=0.11.0): Для работы с токенами (вероятно, для интеграции с LLM).

Все зависимости указаны в файле pyproject.toml.

---

Лицензия

Данный проект распространяется под лицензией MIT. Подробности см. в файле LICENSE.

---