Pythonでwatchdogモジュールを使ってファイルの変更を監視する方法

Pythonでファイル変更をリアルタイムに監視したいと考えたことはありませんか?たとえば、ログファイルの内容を随時確認したり、重要なディレクトリの更新を追跡したりする場合、手動での監視は効率的ではありません。このようなタスクに役立つのが、Pythonの「Watchdog」モジュールです。

Watchdogは、ファイルやディレクトリの変更を検知し、イベントに応じた処理を自動で実行できる強力なツールです。本記事では、Watchdogモジュールを使った基本的な監視方法から応用的な使い方までを詳しく解説します。初心者でも簡単に始められるよう、コード例を交えながら進めていきますので、ぜひご活用ください。

目次

Watchdogモジュールとは

概要

Watchdogモジュールは、Pythonでファイルやディレクトリの変更をリアルタイムに監視するためのライブラリです。このモジュールは、システムネイティブなファイル監視機能を利用して動作し、ファイルの追加・削除・更新といったイベントを検知することができます。

主な特徴

  1. マルチプラットフォーム対応
    Watchdogは、Windows、macOS、Linuxといった主要なプラットフォームで動作します。OS固有の監視APIを利用して、高いパフォーマンスを実現します。
  2. 簡潔なAPI
    WatchdogのAPIは直感的で使いやすく、初心者でも手軽に導入できます。
  3. 拡張性
    Watchdogでは、独自のハンドラーを作成することで、イベントに応じたカスタム処理を実装することが可能です。
  4. リアルタイム監視
    ファイル変更をリアルタイムで検知するため、動的なファイル操作が必要なシステムに最適です。

用途例

  • ログファイルの監視
    システムログやアプリケーションログの変更を検知して、リアルタイムでアラートを送信します。
  • 自動タスク実行
    ファイルの変更をトリガーにして、データ処理や分析を開始するスクリプトを実行します。
  • ディレクトリの同期
    変更が加えられたファイルを別のディレクトリやサーバーに自動でコピーします。

Watchdogモジュールは、シンプルなコードで高機能な監視システムを実現できる便利なライブラリです。この後は、インストール方法と基本的なセットアップについて解説していきます。

Watchdogのインストールとセットアップ

インストール手順

WatchdogモジュールはPythonの標準ライブラリには含まれていないため、まずはインストールを行う必要があります。以下のコマンドを使用して、pip経由でインストールします。

pip install watchdog

インストールが成功すると、Watchdogモジュールをプロジェクトで利用できるようになります。

セットアップの基本

Watchdogを使うには、以下の3つの要素を組み合わせます。

  1. Observer(監視者)
    Observerはファイルシステムの変更を監視する主要なコンポーネントです。監視対象のディレクトリを設定し、変更があると通知を受け取ります。
  2. Event Handler(イベントハンドラー)
    Event Handlerは、Observerからの通知を受けて特定の処理を実行します。標準のハンドラーを使用することも、独自にカスタマイズすることも可能です。
  3. 監視の開始と終了
    Observerは手動で開始(start)および停止(stop)する必要があります。これにより、監視のタイミングを柔軟に管理できます。

基本的なセットアップ例

以下は、Watchdogを使った簡単なファイル変更監視のコード例です。

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

# カスタムイベントハンドラー
class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f"変更を検出: {event.src_path}")

# Observerの設定
if __name__ == "__main__":
    path = "."  # 監視対象のディレクトリ(カレントディレクトリ)
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)

    # 監視の開始
    observer.start()
    print("監視を開始しました")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 監視の停止
        observer.stop()
        print("監視を終了しました")

    observer.join()

コードのポイント

  • FileSystemEventHandlerの継承
    カスタムハンドラーを作成するために使用します。
  • observer.scheduleメソッド
    監視対象のディレクトリとハンドラーを設定します。recursive=Trueにするとサブディレクトリも監視対象になります。
  • 例外処理
    KeyboardInterruptで安全に監視を終了します。

これで、Watchdogを使った監視が始められます。次は、実際のファイル変更検知の仕組みについて詳しく説明します。

基本的な使い方:ファイル変更の検知

ファイル変更を検知する仕組み

Watchdogでは、ファイルシステムのイベントを検知し、イベントごとに対応する処理を実行できます。以下の主要なイベントを監視することが可能です:

  • on_created: ファイルやディレクトリが作成されたとき。
  • on_deleted: ファイルやディレクトリが削除されたとき。
  • on_modified: ファイルが変更されたとき。
  • on_moved: ファイルやディレクトリが移動またはリネームされたとき。

これらのイベントを活用することで、特定のアクションに応じた処理をカスタマイズできます。

コード例:変更検知

以下のコードは、on_modifiedイベントを使用して、ファイルが変更されたときに通知するプログラムです。

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

# イベントハンドラーの作成
class FileChangeHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:  # ディレクトリの変更を除外
            print(f"ファイルが変更されました: {event.src_path}")

# メイン処理
if __name__ == "__main__":
    # 監視するディレクトリを指定(カレントディレクトリ)
    watch_directory = "."
    handler = FileChangeHandler()
    observer = Observer()
    observer.schedule(handler, watch_directory, recursive=False)

    # 監視の開始
    observer.start()
    print("ファイル変更監視を開始しました...")

    try:
        while True:
            time.sleep(1)  # 定期的なポーリング
    except KeyboardInterrupt:
        # 監視の停止
        observer.stop()
        print("ファイル変更監視を停止しました")

    observer.join()

コードの解説

  • on_modifiedメソッド
    ファイル変更イベントをキャッチし、変更されたファイルのパスを出力します。
  • event.is_directoryの確認
    ディレクトリの変更を除外するために使用します。
  • recursive=False
    サブディレクトリを含めず、指定したディレクトリ内のみを監視します。

実行結果

このコードを実行し、監視対象ディレクトリ内のファイルを編集すると、以下のような出力が得られます:

ファイルが変更されました: ./example.txt

イベントごとの動作を追加する

FileSystemEventHandlerの他のメソッドをオーバーライドすることで、複数のイベントに対応可能です。

class MultiEventHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f"作成されました: {event.src_path}")
    def on_deleted(self, event):
        print(f"削除されました: {event.src_path}")
    def on_modified(self, event):
        print(f"変更されました: {event.src_path}")
    def on_moved(self, event):
        print(f"移動されました: {event.src_path} -> {event.dest_path}")

このようにして、さまざまなファイル操作を簡単に追跡できます。次は、イベントハンドラーのカスタマイズ方法について解説します。

ハンドラークラスのカスタマイズ

カスタムハンドラーの必要性

デフォルトのFileSystemEventHandlerをそのまま使用するだけでは、シンプルな処理しか実現できません。しかし、プロジェクトに応じた特定の要件に対応するためには、ハンドラーをカスタマイズし、特定のイベントに応じて独自の処理を追加する必要があります。

カスタムハンドラーの作成

独自のハンドラーを作成するには、FileSystemEventHandlerを継承したクラスを定義し、必要なイベントメソッドをオーバーライドします。

以下は、特定のファイル拡張子に基づいてイベント処理をカスタマイズする例です。

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
import time

# カスタムハンドラーの定義
class CustomHandler(FileSystemEventHandler):
    def __init__(self, target_extension):
        super().__init__()
        self.target_extension = target_extension

    def on_created(self, event):
        if not event.is_directory and event.src_path.endswith(self.target_extension):
            print(f"新しい{self.target_extension}ファイルが作成されました: {event.src_path}")

    def on_modified(self, event):
        if not event.is_directory and event.src_path.endswith(self.target_extension):
            print(f"{self.target_extension}ファイルが変更されました: {event.src_path}")

    def on_deleted(self, event):
        if not event.is_directory and event.src_path.endswith(self.target_extension):
            print(f"{self.target_extension}ファイルが削除されました: {event.src_path}")

# メイン処理
if __name__ == "__main__":
    watch_directory = "."  # 監視対象ディレクトリ
    target_extension = ".txt"  # 対象とするファイル拡張子
    handler = CustomHandler(target_extension)

    observer = Observer()
    observer.schedule(handler, watch_directory, recursive=False)

    # 監視の開始
    observer.start()
    print(f"{target_extension}ファイルの変更監視を開始しました...")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        # 監視の停止
        observer.stop()
        print(f"{target_extension}ファイルの監視を停止しました")

    observer.join()

コードの解説

  • __init__コンストラクタ
    監視対象のファイル拡張子を指定できるようにしています。
  • メソッドの条件分岐
    event.src_path.endswith(self.target_extension)を使い、特定の拡張子のみを監視します。
  • イベントメソッドのカスタマイズ
    ファイルの作成、変更、削除に応じて異なる処理を実装しています。

実行結果例

このプログラムを実行し、監視対象ディレクトリで.txtファイルを作成、変更、または削除すると、以下のような出力が表示されます。

新しい.txtファイルが作成されました: ./example.txt
.txtファイルが変更されました: ./example.txt
.txtファイルが削除されました: ./example.txt

応用例:ロギングやアクションの追加

カスタムハンドラーで、ログを記録したり、変更が検知されたファイルを自動でバックアップするような処理を追加できます。

import logging

class LoggingHandler(FileSystemEventHandler):
    def __init__(self, log_file):
        super().__init__()
        logging.basicConfig(filename=log_file, level=logging.INFO)

    def on_modified(self, event):
        if not event.is_directory:
            logging.info(f"変更されたファイル: {event.src_path}")
            print(f"ログに記録しました: {event.src_path}")

このように、カスタムハンドラーを使用することで、柔軟で実用的なファイル監視が実現できます。次は、ディレクトリ全体を監視する方法を紹介します。

ディレクトリ全体を監視する方法

サブディレクトリの監視が必要な理由

多くのプロジェクトでは、ディレクトリ構造全体を監視する必要があります。例えば、プロジェクトのフォルダ内で新しいサブディレクトリやファイルが追加された場合、それらの変更も検知したい場面がよくあります。Watchdogでは、recursive=Trueオプションを設定することで、このような要求に応えることができます。

サブディレクトリも含めた監視の設定

以下は、ディレクトリ全体を監視し、すべてのファイル変更イベントを検知する例です。

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

# イベントハンドラーの作成
class DirectoryHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f"作成されました: {event.src_path}")

    def on_deleted(self, event):
        print(f"削除されました: {event.src_path}")

    def on_modified(self, event):
        print(f"変更されました: {event.src_path}")

    def on_moved(self, event):
        print(f"移動されました: {event.src_path} -> {event.dest_path}")

# メイン処理
if __name__ == "__main__":
    watch_directory = "."  # カレントディレクトリ
    handler = DirectoryHandler()
    observer = Observer()

    # recursive=Trueでサブディレクトリも監視対象に
    observer.schedule(handler, watch_directory, recursive=True)

    observer.start()
    print("ディレクトリ全体の監視を開始しました...")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("ディレクトリ全体の監視を停止しました")

    observer.join()

コードのポイント

  • recursive=True
    サブディレクトリ内の変更もすべて検知する設定です。
  • 汎用ハンドラーの活用
    on_createdon_deletedon_modifiedon_movedをすべて実装することで、どのような変更もキャッチ可能です。

実行結果例

このプログラムを実行すると、監視対象ディレクトリ内およびそのサブディレクトリで発生するすべての変更が記録されます。

例えば、以下のような結果が得られます:

作成されました: ./subdir/new_file.txt
変更されました: ./subdir/new_file.txt
移動されました: ./subdir/new_file.txt -> ./subdir/renamed_file.txt
削除されました: ./subdir/renamed_file.txt

応用:特定のディレクトリだけを監視する

必要に応じて、特定のサブディレクトリやファイルパターンのみを監視するようにフィルタを追加できます。以下は、特定のサブディレクトリを監視対象外にする例です:

class FilteredHandler(FileSystemEventHandler):
    def __init__(self, exclude_dir):
        super().__init__()
        self.exclude_dir = exclude_dir

    def on_any_event(self, event):
        if self.exclude_dir not in event.src_path:
            print(f"検出: {event.event_type} - {event.src_path}")

このように、柔軟な条件で監視をカスタマイズ可能です。次は、Watchdogの高度な使い方としてイベントフィルタリングを解説します。

Watchdogの高度な使い方:イベントフィルタリング

イベントフィルタリングの必要性

ディレクトリ全体を監視する場合、すべてのイベントを処理するとパフォーマンスが低下したり、不要なイベントを無視するロジックが複雑になる可能性があります。Watchdogでは、特定の条件に基づいてイベントをフィルタリングする仕組みを導入することで、効率的な監視を実現できます。

基本的なフィルタリングの実装

以下の例では、特定の拡張子(例: .txt)のみを監視するフィルタリングを行います。

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
import time

# カスタムハンドラーでイベントをフィルタリング
class FilteredHandler(FileSystemEventHandler):
    def __init__(self, allowed_extensions):
        super().__init__()
        self.allowed_extensions = allowed_extensions

    def on_modified(self, event):
        if not event.is_directory and self._is_allowed_file(event.src_path):
            print(f"変更検知: {event.src_path}")

    def _is_allowed_file(self, file_path):
        return any(file_path.endswith(ext) for ext in self.allowed_extensions)

# メイン処理
if __name__ == "__main__":
    watch_directory = "."  # 監視対象ディレクトリ
    handler = FilteredHandler(allowed_extensions=[".txt", ".log"])

    observer = Observer()
    observer.schedule(handler, watch_directory, recursive=True)

    observer.start()
    print("指定拡張子の変更を監視中...")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("監視を停止しました")

    observer.join()

コードのポイント

  • allowed_extensions
    監視対象とするファイルの拡張子をリストで指定しています。
  • _is_allowed_fileメソッド
    ファイルパスが許可された拡張子で終わるかを確認するヘルパーメソッドです。

特定のディレクトリを除外するフィルタリング

特定のサブディレクトリやファイルパターンを除外する方法もあります。以下の例では、特定のディレクトリ名を含むイベントを無視します。

class ExcludeDirectoryHandler(FileSystemEventHandler):
    def __init__(self, excluded_dirs):
        super().__init__()
        self.excluded_dirs = excluded_dirs

    def on_any_event(self, event):
        if not any(excluded in event.src_path for excluded in self.excluded_dirs):
            print(f"イベント検知: {event.event_type} - {event.src_path}")

実行例

特定のディレクトリを除外したり、特定のファイルだけを対象にすることで、より精密なイベント処理が可能です。

例えば、.txtファイルのみを監視している場合、以下のような結果が得られます:

変更検知: ./important.txt

一方、除外フィルタを適用した場合、指定したディレクトリのイベントは無視されます。

高度な条件でのフィルタリング

フィルタリング条件をさらに細かく設定することもできます。たとえば、以下のような複合条件が可能です:

  • ファイルサイズが一定以上の場合のみ処理
  • ファイル名が特定のパターンに一致する場合のみ処理
  • 作成時間や更新時間に基づく条件

以下は、ファイルサイズに基づくフィルタリングの例です:

import os

class SizeFilteredHandler(FileSystemEventHandler):
    def __init__(self, min_size_bytes):
        super().__init__()
        self.min_size_bytes = min_size_bytes

    def on_modified(self, event):
        if not event.is_directory and os.path.getsize(event.src_path) > self.min_size_bytes:
            print(f"大きなファイルが変更されました: {event.src_path}")

まとめ

イベントフィルタリングを活用することで、Watchdogをより効率的かつ実用的に使用できます。次は、応用例としてログファイルの監視とリアルタイム解析について解説します。

応用例:ログファイルの監視とリアルタイム解析

ログファイル監視の必要性

サーバーやアプリケーションの運用では、ログファイルをリアルタイムで監視し、異常をいち早く検知することが重要です。たとえば、エラーメッセージや特定のキーワードを検知した際に、通知を送ったり処理を実行する仕組みが求められます。Watchdogを活用すれば、これを簡単に実現できます。

基本的なログファイル監視の例

以下の例では、特定のキーワード(例: “ERROR”)を含む行を検知する簡単なログ監視ツールを実装しています。

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

# ログファイル監視用のハンドラー
class LogFileHandler(FileSystemEventHandler):
    def __init__(self, keyword):
        super().__init__()
        self.keyword = keyword

    def on_modified(self, event):
        if not event.is_directory and event.src_path.endswith(".log"):
            print(f"変更を検知: {event.src_path}")
            self._check_log_file(event.src_path)

    def _check_log_file(self, file_path):
        with open(file_path, "r") as file:
            for line in file:
                if self.keyword in line:
                    print(f"キーワード '{self.keyword}' を検出: {line.strip()}")

# メイン処理
if __name__ == "__main__":
    watch_directory = "."  # 監視対象のディレクトリ
    keyword_to_watch = "ERROR"  # 検知対象のキーワード

    handler = LogFileHandler(keyword=keyword_to_watch)
    observer = Observer()
    observer.schedule(handler, watch_directory, recursive=True)

    observer.start()
    print(f"キーワード '{keyword_to_watch}' のログ監視を開始しました...")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
        print("ログ監視を停止しました")

    observer.join()

コードの解説

  • _check_log_fileメソッド
    ログファイルの変更時に呼び出され、変更内容を解析します。
  • キーワード検知
    特定のキーワードを含む行を検索し、検知した場合に出力します。
  • .logファイルに限定
    ログファイル拡張子に限定することで、不要なファイルの監視を避けます。

通知機能の追加

キーワード検知時に通知を送る仕組みを追加することで、即座にアクションを取れるようにできます。以下は、キーワード検知時にSlack通知を送る例です。

import requests

class LogFileHandlerWithNotification(LogFileHandler):
    def __init__(self, keyword, webhook_url):
        super().__init__(keyword)
        self.webhook_url = webhook_url

    def _check_log_file(self, file_path):
        with open(file_path, "r") as file:
            for line in file:
                if self.keyword in line:
                    print(f"キーワード '{self.keyword}' を検出: {line.strip()}")
                    self._send_notification(line.strip())

    def _send_notification(self, message):
        payload = {"text": f"ログでキーワード検出: {message}"}
        requests.post(self.webhook_url, json=payload)

ログ解析の高度な応用

  • リアルタイムデータ解析
    ログの内容を解析して統計情報を取得し、グラフやダッシュボードで可視化する。
  • ログの自動分類
    AIモデルを組み合わせて、ログメッセージを自動で分類する。
  • アラートトリガー
    特定の条件(エラーの頻度が一定を超えた場合など)でアラートを発する。

実行例

ログファイルに以下のような行が追加された場合:

[2023-12-02 10:00:00] INFO: Process started successfully
[2023-12-02 10:01:00] ERROR: Failed to connect to database

以下の出力が得られます:

変更を検知: ./application.log
キーワード 'ERROR' を検出: [2023-12-02 10:01:00] ERROR: Failed to connect to database

まとめ

Watchdogを用いたログファイル監視により、リアルタイムのエラー検出や通知が簡単に実現できます。この手法を基に、独自のログ解析システムを構築することも可能です。次は、Watchdogの限界と代替案について解説します。

Watchdogモジュールの限界と代替案

Watchdogの限界

Watchdogは便利なファイル監視ツールですが、使用環境やユースケースによってはいくつかの制約があります。以下はその主な限界です。

1. パフォーマンスの問題

大量のファイルやディレクトリを監視する場合、特に頻繁な変更が発生すると、パフォーマンスが低下する可能性があります。また、CPU使用率が増加し、システム全体の負荷が高まることがあります。

2. ネイティブAPIへの依存

WatchdogはOSのネイティブAPI(例: WindowsのReadDirectoryChangesW、Linuxのinotify)に依存しています。これにより、OSごとの仕様により以下の制限が発生します:

  • Linux: inotifyでは、監視できるファイル数に上限(inotify watchesの数)があり、制限を超える場合はエラーが発生します。
  • Windows: ネイティブAPIのパフォーマンスは優れていますが、深いディレクトリ階層を持つ構造では遅延が発生する可能性があります。

3. 特定のイベントの検知に制限がある

ファイルの内容変更を検知することはできますが、内容の具体的な違い(差分)を直接把握することはできません。そのため、差分解析を実現するには追加のロジックが必要です。

4. ネットワークストレージの監視が困難

NFSやSMBのようなネットワークストレージでは、ファイル変更イベントが正しく検知されない場合があります。これらの環境では、他の方法を組み合わせる必要があります。

代替案

Watchdogが適さない場合、以下の代替案を検討することができます。

1. **`pyinotify`(Linux限定)**

pyinotifyは、Linuxのinotify機能を直接使用するPythonライブラリです。Watchdogよりも軽量で、Linux環境に特化した監視を行う場合に適しています。

pip install pyinotify

2. **ファイルのポーリング**

ファイルの最終変更日時やハッシュ値を定期的にチェックすることで、変更を検知する方法です。この方法はOSや環境に依存せず、ネットワークストレージにも対応できますが、リアルタイム性は低下します。

例:

import os
import time

file_path = "example.txt"
last_modified = os.path.getmtime(file_path)

while True:
    time.sleep(1)
    current_modified = os.path.getmtime(file_path)
    if current_modified != last_modified:
        print(f"{file_path} が変更されました")
        last_modified = current_modified

3. **クラウドサービスのイベント通知**

クラウドストレージ(例: AWS S3、Google Drive)を使用している場合、サービス提供元が提供するイベント通知機能を利用する方法です。例えば、AWS S3では、バケット内の変更イベントをSNSやLambdaで処理できます。

4. **専用ツールの使用**

以下のような専用ツールやライブラリも検討できます:

  • Filebeat: Elasticsearch社のツールで、ログやファイル変更をリアルタイムで監視し、データを収集します。
  • OS固有ツール: LinuxのauditdやWindowsのFile System Auditなど、OSの監査機能を活用します。

どの代替案を選ぶべきか

ユースケースや環境に応じて適切なツールを選択する必要があります。

  • リアルタイム性が最優先: Watchdogやpyinotify
  • ネットワークストレージ対応: ファイルのポーリングまたはクラウドサービスの通知
  • 大規模なログ処理: Filebeatやクラウドツール

まとめ

Watchdogはシンプルで使いやすいモジュールですが、用途や規模によっては制約を感じる場合があります。その場合、代替案を検討し、最適なツールを選択することで、効率的で安定したファイル監視を実現できます。次は、本記事のまとめと今後の活用方法について説明します。

まとめ

本記事では、PythonのWatchdogモジュールを活用したファイル監視の方法について解説しました。基本的な使い方から、カスタムハンドラーの作成、ディレクトリ全体の監視、高度なイベントフィルタリング、そしてログファイル監視の応用例までを詳しく紹介しました。

また、Watchdogの限界と代替案についても触れ、特定の用途や環境に応じて最適なツールを選択する重要性を示しました。Watchdogは、小規模な監視タスクやプロトタイプ開発に最適なツールですが、要件に応じて代替案を組み合わせることで、さらに高性能な監視システムを構築できます。

ぜひこの記事を参考に、効率的なファイル監視やログ解析システムを構築してみてください。次のステップとして、クラウド連携やリアルタイムデータ処理への応用も検討してみてはいかがでしょうか。

コメント

コメントする

目次