Исходный код soniks_client.jobs.file_manager.move_file

"""Перемещение и удаление файлов и директорий наблюдения после выгрузки."""

import shutil
from pathlib import Path

from core.configs import logger, settings




def delete_observation_directory(directory_path: Path) -> None:
    """Удалить директорию наблюдения со всем содержимым."""

    try:
        shutil.rmtree(directory_path)
        logger.info("Удалена директория: %s", directory_path)
    except Exception as e:
        logger.error("Ошибка при удалении директории %s: %s", directory_path, e)





def delete_data_file(file_path: Path) -> None:
    """Удалить файл декодированных данных."""

    try:
        file_path.unlink()
        logger.debug("Удален файл: %s", file_path)
    except Exception as e:
        logger.error("Ошибка при удалении файла %s: %s", file_path, e)





def delete_data_files(file_paths: list[Path]) -> None:
    """Удалить несколько файлов декодированных данных."""

    for file_path in file_paths:
        try:
            file_path.unlink()
            logger.debug("Удален файл: %s", file_path)
        except Exception as e:
            logger.error("Ошибка при удалении файла %s: %s", file_path, e)





def move_observation_directory_to_complete_directory(directory_path: Path) -> None:
    """Перенести директорию наблюдения в ``complete``."""

    complete_path = settings.paths.complete_path
    destination_directory = complete_path / directory_path.name

    if not destination_directory.exists():
        try:
            shutil.move(str(directory_path), str(destination_directory))
            logger.info(
                "Директория перемещена из %s в %s",
                directory_path,
                destination_directory,
            )

            return

        except Exception as e:
            logger.error(
                "Ошибка при перемещении директории %s: %s",
                destination_directory,
                e,
            )

            return

    for file_in_dir in directory_path.iterdir():
        target_file = destination_directory / file_in_dir.name

        try:
            if not target_file.exists():
                shutil.move(str(file_in_dir), str(target_file))
                logger.info(
                    "Загруженный файл перемещён из %s в %s",
                    file_in_dir,
                    target_file,
                )
            else:
                logger.info(
                    "Пропуск перемещения - загруженный файл %s уже существует в директории %s",
                    file_in_dir,
                    destination_directory,
                )

        except Exception as e:
            logger.error("Ошибка перемещения файла %s: %s", file_in_dir, e)

    try:
        shutil.rmtree(directory_path)
    except Exception as e:
        logger.error("Ошибка при удалении директории %s: %s", directory_path, e)





def move_data_file_to_complete_directory(file_path: Path) -> None:
    """Перенести выгруженный файл в ``complete``."""

    destination_directory = settings.paths.complete_path
    source_directory_name = file_path.parent.name

    destination_subdirectory = destination_directory / source_directory_name
    destination_subdirectory.mkdir(parents=True, exist_ok=True)
    destination_file_path = destination_subdirectory / file_path.name

    if destination_file_path.exists():
        logger.warning(
            "Файл %s уже существует в директории %s, поэтому он будет удален",
            file_path.name,
            destination_subdirectory,
        )

        try:
            file_path.unlink()
            logger.info("Удален файл: %s", file_path)
        except Exception as e:
            logger.error("Ошибка при удалении файла %s: %s", file_path, e)

        # Без выхода управление проваливалось в shutil.move на уже удалённом
        # файле — ошибка в лог на каждый такой файл.
        return

    try:
        shutil.move(str(file_path), str(destination_file_path))
        logger.info(
            "Загруженный файл перемещён из %s в %s",
            file_path,
            destination_file_path,
        )
    except Exception as e:
        logger.error("Ошибка при перемещении файла: %s", e)





def move_file_to_incomplete_directory(file_path: Path) -> bool:
    """Перенести невыгруженный файл в ``incomplete`` для повторной попытки.

    Returns:
        ``True``, если файл лёг в ``incomplete``. ``False`` означает, что файл
        выпал из очереди повторов: вызывающий не должен закрывать директорию
        наблюдения, иначе единственная копия данных будет удалена.
    """

    destination_directory = settings.paths.incomplete_path
    source_directory_name = file_path.parent.name

    destination_subdirectory = destination_directory / source_directory_name
    destination_subdirectory.mkdir(parents=True, exist_ok=True)
    destination_file_path = destination_subdirectory / file_path.name

    try:
        shutil.move(str(file_path), str(destination_file_path))
        logger.info(
            "Незагруженный файл перемещён из %s в %s",
            file_path,
            destination_file_path,
        )

        return True

    except Exception as e:
        logger.error("Ошибка при перемещении файла: %s", e)

        return False