Введение
Часто нам кажется, что компьютер, на котором мы работаем, загроможден файлами. У вас едва хватает времени, чтобы упорядочить эти файлы, и вам даже сложно найти конкретный файл именно тогда, когда он вам нужен. Даже файловый проводник имеет функцию поиска файлов, но кто запоминает имена файлов дольше. РЖУ НЕ МОГУ!!!
В этой статье я поделюсь пошаговым руководством по коду того, как вы можете всего за несколько минут создать собственную службу автоматизированного органайзера файлов. Давайте углубимся в это.
Дизайн рабочего процесса
Дизайн довольно простой. Служба мониторинга, которая следит за изменением папки. Если он обнаружит изменение, он проверит расширение файла и отправит файл в соответствующую папку. Вот схема, чтобы было легче понять.
Компоненты
Давайте попробуем разобрать его и попытаться понять, что именно нам нужно для построения этого сервиса.
- 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
Демо
Заключение
Так что это было легко право!!!
Надеюсь, вам понравился урок. Если да, то хлопните в ладоши или следите за другими подобными статьями.
Попробуйте этот код и убедитесь сами, насколько он удивителен и облегчит вашу жизнь.
Примечание для моих читателей. Следите за моей страницей, чтобы получать больше обновлений, и если вам очень нравятся мои статьи, то есть новая функция подсказок, чтобы выразить любовь к вашему любимому писателю. Ваша поддержка будет высоко оценена. Спасибо!!!
