import os from loguru import logger from base.base import BaseDirectoryMonitor, BaseFileManager, Settings class FileManager(BaseFileManager): """ Менеджер файлов для работы с библиотекой путей. Основные функции: - Переотправка (replot) списка найденных файлов. - Добавление нового пути (например, выбранного пользователем). - Переключение режимов работы (создание отчёта, онлайн-мониторинг, работа с трейсами клиента). - Обновление настроек мониторинга. - Поиск файлов в директории с трейсами: поиск файлов .dat и .txt. """ def replot_all(self) -> None: """ Переотправляет все пути из библиотеки. Если текущий режим не равен 3, уведомляет медиатора со списком путей. Если режим равен 3, открывает директорию с трейсами. """ if self._paths_library is not None: if self._mode != 3: self._mediator.notify(self, list(self._paths_library)) else: self.open_raw_traces_dir(list(self._paths_library)[0]) def open_custom_file(self, path: str) -> None: """ Добавляет указанный путь в библиотеку и уведомляет медиатора. :param path: Путь к файлу, выбранному пользователем. """ self._paths_library.add(path) # Передаем уведомление с указанным путем (если путь пустой, передаем пустую строку) self._mediator.notify(self, [path if path else '']) def set_mode(self, num: int) -> None: """ Устанавливает режим работы менеджера файлов. Режимы: 1 - Режим создания отчёта. 2 - Режим онлайн-мониторинга папки. 3 - Режим работы с трейсами клиента. :param num: Целое число, определяющее режим работы. """ match num: case 1: # Режим создания отчёта self._monitor.stop() self._paths_library.clear() self._paths_library.add('') self._mediator.notify(self, list(self._paths_library)) self._mode = 1 case 2: # Режим онлайн-мониторинга папки self._monitor.init_state() self._monitor.start() self._mode = 2 case 3: # Режим работы с трейсами клиента self._monitor.stop() self._paths_library.clear() self._mode = 3 def update_monitor_settings(self, settings: Settings) -> None: """ Обновляет настройки мониторинга директории. Из объекта настроек извлекается путь к директории и интервал обновления. Выполняется проверка существования директории и корректности интервала обновления. При необходимости приостанавливается или возобновляется монитор. :param settings: Объект Settings с системными настройками. """ directory_path = settings.system['trace_storage_path'][0] update_time = settings.system['monitor_update_period'][0] if not os.path.exists(directory_path): logger.warning(f"Путь {directory_path} не существует.") # Можно раскомментировать raise, если требуется прерывание работы # raise FileNotFoundError(f"Путь {directory_path} не существует.") if update_time <= 0.01: logger.warning(f"Интервал между проверками папки слишком мал: {update_time}") if self._monitor.isActive: self._monitor.pause() self._monitor._directory_path = directory_path self._monitor._update_time = update_time if self._monitor.isActive: self._monitor.start() def add_new_paths(self, paths: list) -> None: """ Добавляет новые пути файлов в библиотеку, если они ранее не присутствовали, и уведомляет медиатора о новых путях. :param paths: Список новых путей. """ paths_set = set(paths) new_paths = self._paths_library.difference(paths_set) self._paths_library.update(new_paths) self._mediator.notify(list(new_paths)) def open_raw_traces_dir(self, path: str) -> None: """ Открывает указанную директорию, ищет в ней файлы с расширениями .dat и .txt, и при их наличии инициирует предварительный рендеринг через медиатора. :param path: Путь к директории с трейсами. """ self._paths_library.clear() self._paths_library.add(path) dat_file, txt_file = None, None # Перебираем содержимое директории for entry in os.listdir(path): full_path = os.path.join(path, entry) if os.path.isfile(full_path): _, ext = os.path.splitext(entry) ext = ext.lower() if ext == '.dat' and dat_file is None: dat_file = full_path elif ext == '.txt' and txt_file is None: txt_file = full_path if dat_file and txt_file: break if dat_file and txt_file: # Уведомляем медиатора для запуска предварительного рендеринга трейс-контента self._mediator.prerender_TCW(self, [dat_file, txt_file]) class DirectoryMonitor(BaseDirectoryMonitor): """ Мониторинг директории для поиска CSV-файлов. Инициализирует состояние, подключает таймер обновления и отслеживает появление новых файлов. При обнаружении новых файлов уведомляет менеджер файлов. """ def init_state(self): """ Инициализирует состояние мониторинга директории. Проверяет, существует ли директория. Если директория не существует, генерирует ошибку. Собирает список файлов с расширением .csv и подключает метод мониторинга к таймеру. """ if not os.path.exists(self._directory_path): logger.error(f"Путь {self._directory_path} не существует.") raise FileNotFoundError(f"Путь {self._directory_path} не существует.") self._files = [ os.path.join(self._directory_path, file) for file in os.listdir(self._directory_path) if file.lower().endswith('.csv') ] self.update_timer.timeout.connect(self._monitor) logger.info("Monitor initiated!") def _monitor(self): """ Метод, вызываемый по таймеру, для проверки появления новых CSV-файлов. Если обнаружены новые файлы, уведомляет менеджер файлов и обновляет список файлов. Если в директории отсутствуют файлы, сбрасывает список. """ current_files = [ os.path.join(self._directory_path, file) for file in os.listdir(self._directory_path) if file.lower().endswith('.csv') ] # Определяем файлы, которых еще не было в предыдущем списке new_files = sorted(list(filter(lambda x: x not in self._files, current_files))) if new_files: logger.info(f"New files detected: {new_files}") self._file_manager.add_new_paths(new_files) self._files = current_files if not current_files: self._files = []