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

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

from collections import defaultdict
from pathlib import Path

from core.configs import logger, settings
from core.types import FilePathsToSend




def get_paths_observation_files(observation_id: str) -> tuple[Path, FilePathsToSend]:
    """Собрать пути непустых файлов наблюдения, сгруппировав их по префиксу.
    
    Returns:
        Директория наблюдения и словарь «префикс — список путей».
    """

    output_path = settings.paths.output_path
    observation_dir_path = output_path / observation_id

    prefixes_of_files = (
        settings.observation.get_prefixes_of_files_available_for_upload()
    )

    file_paths = _collect_file_paths_by_prefix(
        observation_dir_path,
        prefixes_of_files,
    )

    return observation_dir_path, file_paths



def _collect_file_paths_by_prefix(
    observation_dir_path: Path,
    prefixes: list[str],
) -> FilePathsToSend:
    prefix_files = defaultdict(list)

    for prefix in prefixes:
        pattern = f"{prefix}*"

        for file_path in observation_dir_path.glob(pattern):
            if not file_path.is_file():
                logger.info("Путь: %s не является файлом", file_path)

                continue

            try:
                if file_path.stat().st_size == 0:
                    continue
            except OSError as e:
                logger.info("Не удалось получить размер файла %s: %s", file_path, e)

                continue

            prefix_files[prefix].append(file_path)

    return prefix_files