Введение
Часто нам кажется, что компьютер, на котором мы работаем, загроможден файлами. У вас едва хватает времени, чтобы упорядочить эти файлы, и вам даже сложно найти конкретный файл именно тогда, когда он вам нужен. Даже файловый проводник имеет функцию поиска файлов, но кто запоминает имена файлов дольше. РЖУ НЕ МОГУ!!!
В этой статье я поделюсь пошаговым руководством по коду того, как вы можете всего за несколько минут создать собственную службу автоматизированного органайзера файлов. Давайте углубимся в это.
Дизайн рабочего процесса
Дизайн довольно простой. Служба мониторинга, которая следит за изменением папки. Если он обнаружит изменение, он проверит расширение файла и отправит файл в соответствующую папку. Вот схема, чтобы было легче понять.
Компоненты
Давайте попробуем разобрать его и попытаться понять, что именно нам нужно для построения этого сервиса.
- 1. Конфигурация — состоит из файлов с разными расширениями и типами файлов в формате пары «ключ-значение». Еще одна информация, которую нам нужно взять из конфигурации, — это папка, изменения в которой служба будет отслеживать.
- 2. Служба мониторинга. Для этого мы будем использовать watchdog, который представляет собойпакет для мониторинга изменений в файловой системе. Событие при создании запускается, как только сторожевой таймер обнаруживает любое изменение в папке.
- 3. Ведение журнала — для сохранения информации в файле журнала для проверки успешного или неудачного запуска событий и перемещения файлов.
Код
Начните с создания виртуальной среды.
python3 -m venv app_env !pip install watchdog
Наконец, вот код.
- config.json:
{ "extension_to_folder": { ".jpg": "image", ".jpeg": "image", ".png": "image", ".gif": "image", ".doc": "document", ".docx": "document", ".pdf": "document", ".txt": "document", ".xls": "excel", ".xlsx": "excel", ".csv": "excel", ".mp4": "video", ".mkv": "video", ".avi": "video", ".flv": "video", ".mov": "video", ".mp3": "audio", ".ogg": "audio", ".flac": "audio", ".wav": "audio", ".zip": "archive", ".tar": "archive", ".gz": "archive", ".tar.gz": "archive", ".7z": "archive", ".rar": "archive" }, "watch_folder_path": "FOLDER_PATH_TO_WATCH" }
Измените значение FOLDER_PATH_TO_WATCH на путь к папке, за файлами которой вы хотите, чтобы служба следила.
2. app.py:
import os import shutil import time import logging from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler import json import sys print(f'ENV: {sys.prefix}') with open('config.json', 'rb') as config: config_data = json.load(config) extension_to_folder = config_data["extension_to_folder"] class NewFileHandler(FileSystemEventHandler): def __init__(self, dst_folder, logger): self.dst_folder = dst_folder self.logger = logger def create_if_not_exists(self, dst): if not os.path.exists(dst): os.makedirs(dst) def on_created(self, event): if not event.is_directory: filename, file_extension = os.path.splitext(event.src_path) logging.info(f"Got new file - {filename}") file_extension = file_extension.lower() if file_extension in extension_to_folder: file_type = extension_to_folder[file_extension] logging.info(f"File has been classfied as - {file_type}") base_folder = os.path.join(self.dst_folder, file_type) self.create_if_not_exists(base_folder) dst = os.path.join(base_folder, os.path.basename(event.src_path)) try: shutil.move(event.src_path, dst) self.logger.info(f"Moved {event.src_path} to {dst}") except: self.logger.warning(f"Failed to move file - {event.src_path}") else: self.logger.info(f"{event.src_path} has an unsupported extension") others_folder = 'others' dst = os.path.join(self.dst_folder, others_folder) self.create_if_not_exists(dst) shutil.move(event.src_path, dst) self.logger.info(f"Moved {event.src_path} to {others_folder}") def monitor_folder(folder_path): print(f"App has Started watching - {folder_path}") log_folder = "./logs" if not os.path.exists(log_folder): os.makedirs(log_folder) log_filename = log_folder + "/log_" + time.strftime("%Y%m%d-%H%M%S") + ".log" logging.basicConfig(filename=log_filename, level=logging.INFO) console_handler = logging.StreamHandler() console_handler.setLevel(logging.WARNING) logging.getLogger().addHandler(console_handler) logger = logging.getLogger() event_handler = NewFileHandler(folder_path, logger) observer = Observer() observer.schedule(event_handler, folder_path, recursive=False) observer.start() try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() logger.info("App Stopped.") observer.join() folder_path = config_data["watch_folder_path"] if __name__ == '__main__': monitor_folder(folder_path)
Здесь мы запускаем пользовательское событие сторожевого таймера, которое наследуем от класса FileSystemEventHandler, и переопределяем функцию on_created для выполнения нашего действия. Каждый раз, когда сторожевой таймер обнаруживает новый файл в папке, он выполняет функцию on_created.
Запустить как фоновый процесс
Давайте создадим сценарий оболочки, чтобы запустить этот процесс.
#!/bin/bash cd PATH_TO_THE_APP_FOLDER pwd source app_env/bin/activate python3 app.py
Сохраните его как file_org.sh. Чтобы сделать его исполняемым —
chmod +x file_org.sh
Я использую tmux для запуска этой службы в фоновом режиме на устройстве Mac. Однако есть много других способов сделать это.
Установить tmux —
brew install tmux
Создать новый сеанс (это перенесет вас внутрь нового сеанса) —
tmux new -s file_org
Внутри сеанса запустите сценарий оболочки —
./file_org.sh
Нажмите Control + B, D, чтобы отключиться от активного сеанса.
Чтобы просмотреть активные сеансы tmux —
tmux ls
Демо
Заключение
Так что это было легко право!!!
Надеюсь, вам понравился урок. Если да, то хлопните в ладоши или следите за другими подобными статьями.
Попробуйте этот код и убедитесь сами, насколько он удивителен и облегчит вашу жизнь.
Примечание для моих читателей. Следите за моей страницей, чтобы получать больше обновлений, и если вам очень нравятся мои статьи, то есть новая функция подсказок, чтобы выразить любовь к вашему любимому писателю. Ваша поддержка будет высоко оценена. Спасибо!!!