PyWatch — Первая программа
20.09.2025       Backend       Лайков: 0
PyWatch — Первая программа
В этой статье мы познакомились с PyWatch — простым инструментом для мониторинга изменений в файловой системе с помощью Python. Мы создали виртуальное окружение на macOS, установили библиотеку watchdog и написали первую программу-наблюдатель.

Введение

 

PyWatch — это мой учебный проект, цель которого понять, как работает мониторинг файловой системы. Мы будем отслеживать создание, изменение, перемещение и удаление файлов и папок в реальном времени.

Обычные Python-скрипты работают по запросу: запустил → получил результат → программа завершилась. PyWatch устроен иначе — он «сидит на проводе» и реагирует на события, которые происходят в ОС.

 

Подготовка окружения

 

Работаем на macOS в VS Code.

Создаём и активируем виртуальное окружение:

 

python3 -m venv venv
source venv/bin/activate

 

Устанавливаем библиотеку watchdog:

 

pip install watchdog

 

Если после установки IDE (VS Code) не видит библиотеку, просто перезапустите редактор.

 

Начало кода

 

import time
from datetime import datetime as dt
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

 

Здесь мы подключаем модули:

  • time и datetime — для работы со временем;
  • Path из модуля pathlib — для удобной работы с путями;
  • Observer и FileSystemEventHandler — ключевые классы из watchdog, которые позволяют следить за событиями файловой системы.

 

Настройки

 

WATCH_PATH = "."

 

Задаём путь, за которым будем следить. Точка (".") означает текущую папку, где запущена программа.

 

Вспомогательные функции

 

def stamp():
    return dt.now().strftime("%Y-%m-%d %H:%M:%S")

def norm(p: str) -> str:
    return str(Path(p).resolve())

 

  • stamp() возвращает текущее время в удобном формате для логов.
  • norm(p) преобразует путь в абсолютный, чтобы всегда работать с единообразными строками.

 

Зачем нужен .resolve()?

Метод .resolve() делает путь «нормальным» для ОС:

  • приводит относительные пути к абсолютным ("./file.txt" → "/Users/user/project/file.txt"),
  • убирает «.», «..» и другие избыточные элементы,
  • на некоторых системах ещё и разворачивает символические ссылки.

Таким образом, если в программу попадут разные варианты одного и того же пути, .resolve() приведёт их к единой строке. Это важно, чтобы не было дублей и ошибок в логике.

 

Обработчик событий

 

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [CREATE] {kind} {norm(event.src_path)}")

    def on_modified(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [MODIFY] {kind} {norm(event.src_path)}")

    def on_deleted(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [DELETE] {kind} {norm(event.src_path)}")

    def on_moved(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [MOVE]   {kind} {norm(event.src_path)} -> {norm(event.dest_path)}")

 

Здесь мы создаём класс-наблюдатель.

 

Что делает этот класс?

 

FileSystemEventHandler — это базовый класс из библиотеки watchdog. Мы создаём свой класс MyHandler, который от него наследуется, и переопределяем методы:

  • on_created — вызывается, когда ОС сигнализирует о создании файла/папки,
  • on_modified — при изменении,
  • on_deleted — при удалении,
  • on_moved — при перемещении или переименовании.

 

Каждый метод получает объект event, в котором лежит:

  • event.src_path — путь к объекту,
  • event.dest_path — новый путь (только для перемещений),
  • event.is_directory — флаг, указывает, файл это или папка.

 

Как это работает на уровне ОС?

 

Когда вы создаёте или удаляете файл, ядро ОС фиксирует событие в файловой системе. У каждой платформы свой механизм:

  • macOS использует FSEvents,
  • Linux — inotify,
  • Windows — ReadDirectoryChangesW.

 

Эти системные службы отслеживают изменения и сообщают о них «наблюдателю» (Observer). Watchdog под капотом подключается именно к этим API.

Observer получает событие от ОС и передаёт его обработчику. Обработчик вызывает соответствующий метод (on_created, on_deleted и т.д.), и в нём мы уже решаем, что делать — например, вывести лог в консоль.

Это значит, что программа не «ходит в цикл и проверяет», есть ли изменения, а именно получает готовые сигналы от ОС. Это и делает её быстрой и эффективной.

 

Чуть-чуть подробнее: как ОС отслеживает изменения в файловой системе

 

Когда мы пишем код на Python и используем библиотеку watchdog, мы работаем на высоком уровне — просто вызываем Observer и Handler. Но важно понимать, что под капотом всё завязано на механизмах самой операционной системы.

У каждой ОС есть свой собственный интерфейс для отслеживания изменений в файловой системе:

 

  • Linux — inotify
    Это часть ядра Linux. При создании файлового дескриптора inotify ядро начинает отправлять события при изменениях в выбранных директориях.
    События приходят в очередь: «файл создан», «файл удалён», «файл изменён». Каждое событие содержит тип операции и путь.
    Watchdog просто слушает эту очередь и конвертирует её в привычные Python-события (on_created, on_deleted и т.д.).
  • macOS — FSEvents
    Здесь используется сервис ядра под названием File System Events. macOS ведёт журнал изменений для каталогов. Программа подписывается на этот журнал и получает список событий с отметками времени и путями.
    В отличие от inotify, FSEvents может объединять сразу несколько событий (например, пакет изменений в папке) и передавать их одной пачкой.
  • Windows — ReadDirectoryChangesW
    Это системный вызов в WinAPI. Программа указывает директорию, и ОС начинает слать уведомления в буфер, когда внутри неё происходят изменения.
    Здесь можно подписаться сразу на несколько типов событий: создание, удаление, перемещение, изменение атрибутов.

 

Общий принцип:

  1. ОС ведёт внутренний журнал изменений файловой системы.
  2. Наблюдатель (Observer) подписывается на этот журнал через системный API.
  3. Когда пользователь или программа создаёт/удаляет/меняет файл, ядро записывает событие в журнал.
  4. Watchdog получает это событие и вызывает соответствующий метод (on_created, on_deleted, on_modified, on_moved).

 

Таким образом, наш код реагирует на реальные сигналы от ядра, а не просто «ходит по кругу и проверяет состояние папки». Это экономит ресурсы и позволяет реагировать мгновенно.

 

┌─────────────┐      ┌─────────────┐      ┌───────────────┐      ┌────────────────┐
│   ОС ядро   │ ---> │  Observer   │ ---> │   Handler     │ ---> │   Ваш код      │
│ (inotify /  │      │ (watchdog)  │      │ (on_created,  │      │ (print, логика │
│  FSEvents / │      │             │      │  on_deleted…) │      │  бэкапы и т.д.)│
│ ReadDir...) │      │             │      │               │      │                │
└─────────────┘      └─────────────┘      └───────────────┘      └────────────────┘

 

Запуск наблюдателя

 

if __name__ == "__main__":
    path = WATCH_PATH
    handler = MyHandler()
    observer = Observer()
    observer.schedule(handler, path, recursive=True)
    observer.start()

    print(f"Watching {Path(path).resolve()} (Ctrl+C to stop)")
    try:
        while True:
            time.sleep(0.25)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

 

Что здесь происходит:

  1. Создаём экземпляр MyHandler.
  2. Создаём Observer — он отвечает за связь с ОС.
  3. «Назначаем» обработчик на нужную папку через schedule.
  4. Запускаем наблюдателя.
  5. Программа работает в бесконечном цикле, пока мы её не остановим Ctrl+C.
  6. При остановке корректно завершаем работу (observer.stop() и observer.join()).

 

Запуск программы

 

Сохраняем файл, например, как pywatch.py, и запускаем:

 

python3 pywatch.py

 

Теперь попробуйте создать или удалить файл в папке, за которой идёт слежение. События сразу появятся в консоли.

 

Итоговый код

 

import time
from datetime import datetime
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

WATCH_PATH = "."

def stamp():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def norm(p: str) -> str:
    return str(Path(p).resolve())

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [CREATE] {kind} {norm(event.src_path)}")

    def on_modified(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [MODIFY] {kind} {norm(event.src_path)}")

    def on_deleted(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [DELETE] {kind} {norm(event.src_path)}")

    def on_moved(self, event):
        kind = "DIR " if event.is_directory else "FILE"
        print(f"{stamp()} [MOVE]   {kind} {norm(event.src_path)} -> {norm(event.dest_path)}")

if __name__ == "__main__":
    path = WATCH_PATH
    handler = MyHandler()
    observer = Observer()
    observer.schedule(handler, path, recursive=True)
    observer.start()

    print(f"Watching {Path(path).resolve()} (Ctrl+C to stop)")
    try:
        while True:
            time.sleep(0.25)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()