Исходный код EJIO.utils.system.file_utils

## /utils/system/file_utils.py

import datetime
import os
import platform
from pathlib import Path
from contextlib import contextmanager
from collections.abc import Callable, Generator
from loguru import logger

__all__ = [
    'change_cwd',
    'generate_timestamped_filename',
    'save_atomic',
    'rotate_and_save',
]


def _make_file_hidden(file_path: Path) -> None:
    """
    Внутренний хелпер: делает файл скрытым на уровне операционной системы.

    :param file_path: Путь к файлу.
    """
    try:
        if platform.system().lower() == "windows":
            import ctypes
            kernel32 = ctypes.windll.kernel32
            set_attr_func = getattr(kernel32, 'SetFileAttributesW', None)

            if set_attr_func is not None:
                set_attr_func.argtypes = [ctypes.c_wchar_p, ctypes.c_ulong]
                set_attr_func.restype = ctypes.c_bool
                # Константа 0x02 — это FILE_ATTRIBUTE_HIDDEN в Win32 API
                set_attr_func(str(file_path), 0x02)
    except Exception as e:
        logger.debug(f"Failed to set the stealth attribute to the file {file_path}: {e}")




def save_atomic(target_path: str | Path, write_callback: Callable[[Path], None]) -> None:
    """
    Выполняет транзакционную (атомарную) запись в файл через скрытый временный файл.
    Гарантирует, что оригинальный файл не будет поврежден в случае сбоя программы или ОС.

    :param target_path: Целевой путь к файлу (куда нужно сохранить итог).
    :param write_callback: Функция-коллбэк, принимающая Path временного файла, куда нужно писать данные.
    """
    dest = Path(target_path).resolve()
    # Создаем временное имя. На Linux/Mac точка на конце сделает его скрытым по умолчанию.
    temp_name = f".tmp_{generate_timestamped_filename(dest.stem, dest.suffix)}"
    temp_path = dest.parent.joinpath(temp_name)

    try:
        # 1. Обеспечиваем существование родительской папки
        dest.parent.mkdir(parents=True, exist_ok=True)

        # 2. Создаем пустой временный файл
        temp_path.touch(exist_ok=True)

        # 3. Передаем временный файл в callback пользователя для записи данных
        write_callback(temp_path)

        # 4. Выставляем файлу скрытость
        _make_file_hidden(temp_path)

        # 5. АТОМАРНАЯ СМЕНА (ТРАНЗАКЦИЯ):
        # Переименовываем временный файл в целевой. На уровне ядра ОС
        # это атомарная операция замены указателей. Старый файл затрется безболезненно.
        os.replace(temp_path, dest)

    except Exception as e:
        logger.error(f"Critical failure of atomic write to file {dest.name}: {e}")
        # В случае падения аккуратно убираем временный мусор, если он успел создаться
        if temp_path.exists():
            try:
                os.remove(temp_path)
            except OSError:
                pass
        raise





def rotate_and_save(
        target_path: str | Path,
        max_backups: int,
        write_callback: Callable[[Path], None]
) -> None:
    """
    Управляет ротацией бэкапов файла по индексам (file.1.ext, file.2.ext)
    и атомарно сохраняет новое состояние под основным именем.

    :param target_path: Путь к основному файлу (например, 'config.json').
    :param max_backups: Максимальное количество удерживаемых файлов бэкапа (>= 1).
    :param write_callback: Функция-коллбэк для безопасной записи нового файла.
    """
    dest = Path(target_path).resolve()
    if max_backups < 1:
        raise ValueError(f"Number of backups must be no less than 1, got {max_backups} instead.")

    # Если основного файла еще нет — просто делаем атомарную запись
    if not dest.exists():
        save_atomic(dest, write_callback)
        logger.info(f"Backup rotation and atomic file update successfully performed: {dest.name}")
        return

    # --- ФАЗА РОТАЦИИ БЭКАПОВ ---
    # Обход делаем С КОНЦА (от старых к новым), чтобы не перезаписать файлы до сдвига
    for i in range(max_backups, 0, -1):
        # Формируем имя текущего индекса бэкапа (например, config.1.json)
        current_backup = dest.parent.joinpath(f"{dest.stem}.{i}{dest.suffix}")

        if current_backup.exists():
            if i == max_backups:
                # Самый старый бэкап за пределами лимита просто удаляем
                os.remove(current_backup)
            else:
                # Остальные сдвигаем на 1 шаг вверх (config.1.json -> config.2.json)
                next_backup = dest.parent.joinpath(f"{dest.stem}.{i + 1}{dest.suffix}")
                os.replace(current_backup, next_backup)

    # Основной файл сдвигаем в самый первый бэкап (config.json -> config.1.json)
    first_backup = dest.parent.joinpath(f"{dest.stem}.1{dest.suffix}")
    os.replace(dest, first_backup)

    # --- ФАЗА СОХРАНЕНИЯ ---
    # Записываем новое состояние под оригинальным именем через нашу атомарную транзакцию
    save_atomic(dest, write_callback)
    logger.info(f"Backup rotation and atomic file update successfully performed: {dest.name}")





def generate_timestamped_filename(base_name: str, extension: str, prefix_date: bool = False) -> str:
    """
    Генерирует имя файла, интегрируя текущую дату и время в безопасном формате ISO.

    Пример: generate_timestamped_filename("report", "json")
    Результат: "report_20260621_153045.json"

    :param base_name: Базовое имя файла (например, 'backup', 'log').
    :param extension: Расширение файла без точки (например, 'txt', 'csv').
    :param prefix_date: Если True, дата-время встанет в начало: '20260621_153045_report.json'.
    :return: Строка с уникальным именем файла.
    """
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    clean_ext = extension.lstrip('.')

    if prefix_date:
        return f"{timestamp}_{base_name}.{clean_ext}"
    return f"{base_name}_{timestamp}.{clean_ext}"





@contextmanager
def change_cwd(target_path: str | Path) -> Generator[Path, None, None]:
    """
    Контекстный менеджер для временной смены текущей рабочей директории (CWD).
    По завершении блока кода или при возникновении исключения гарантированно
    возвращает рабочую папку в исходное состояние.

    Пример использования:
    with change_cwd("./resources"):
        # Код выполняется внутри папки resources
        open("theme.json", "r")

    :param target_path: Путь к директории.
    :yields: Путь к директории.
    """
    original_cwd = Path(os.getcwd()).resolve()
    resolved_target = Path(target_path).resolve()

    if not resolved_target.exists():
        raise FileNotFoundError(f"Cannot change directory. Path '{resolved_target}' does not exist.")
    if not resolved_target.is_dir():
        raise NotADirectoryError(f"Cannot change directory. Path '{resolved_target}' is not a directory.")

    try:
        os.chdir(resolved_target)
        logger.debug(f"The working directory is temporarily changed to: '{resolved_target}'")
        yield resolved_target
    finally:
        os.chdir(original_cwd)
        logger.debug(f"The working directory was successfully restored to its original: '{original_cwd}'")