MRが楽しい

MRやVRについて学習したことを書き残す

Pythonでファイル変更イベントを検出する

本日は Python の技術調査枠です。
Python でファイル変更イベントを検出する方法についてまとめます。

「Watchdog」のインストール

今回は Python でファイル監視を行うAPIを提供する「Watchdog」パッケージを利用します。
Pythonの標準パッケージではなく、外部パッケージになので別途インストールが必要です。
以下の環境構築を実施していれば、Python のパッケージ管理システムである pip がインストールされており、利用できます。
bluebirdofoz.hatenablog.com

ただし pip へのパスが通されていない場合は以下のパスを環境変数の"PATH"に登録するか、ディレクトリ直下で直接 pip を実行する必要があります。
・C:\Users\(ユーザ名)\AppData\Local\Programs\Python\Python37\Scripts

pip へのパスを通したら以下のコマンドを実行して「Watchdog」をインストールします。
・pip install watchdog
f:id:bluebirdofoz:20190321132402j:plain

なお、proxy 下で実行する場合は pip install コマンドの前に以下のコマンドで proxy
環境変数をしておく必要があります。
・set HTTP_PROXY=http://192.168.X.X:XXXX
・set HTTPS_PROXY=http://192.168.X.X:XXXX

Pythonサンプルコード

試しにファイル変更イベントを検出したら標準出力に通知を行うサンプルコードを作成します。
ファイル変更イベントを検出するには FileSystemEventHandler の継承クラスを作成します。
・FileEvent.py

# ファイル変更イベント検出のため、watchdogをインポート
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer

# ファイルアクセスとスリープのため、osとtimeをインポート
import os
import time

# 監視対象ディレクトリを指定する
target_dir = 'D:\\Python\\FileEventTest\\TestDir'

# FileSystemEventHandler の継承クラスを作成
class FileChangeHandler(FileSystemEventHandler):
     # ファイル作成時のイベント
     def on_created(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s created' % filename)

     # ファイル変更時のイベント
     def on_modified(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s changed' % filename)

     # ファイル削除時のイベント
     def on_deleted(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s deleted' % filename)

     # ファイル移動時のイベント
     def on_moved(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s moved' % filename)

# コマンド実行の確認
if __name__ == "__main__":
     # ファイル監視の開始
     event_handler = FileChangeHandler()
     observer = Observer()
     observer.schedule(event_handler, target_dir, recursive=True)
     observer.start()
     # 処理が終了しないようスリープを挟んで無限ループ
     try:
         while True:
             time.sleep(0.1)
     except KeyboardInterrupt:
         observer.stop()
     observer.join()

保存したスクリプトを以下のコマンドで実行します。
・py .\fileevent.py

この状態で指定ディレクトリにテキストファイルを作成してみます。
f:id:bluebirdofoz:20190321132418j:plain

画像の作成と変更が検出されました。成功です。
f:id:bluebirdofoz:20190321132427j:plain

なお、イベントハンドラークラスは以下の4つが提供されています。
・FileSystemEventHandler
・PatternMatchingEventHandler
・RegexMatchingEventHandler
・LoggingEventHandler

ファイルを指定して監視を行いたい場合、PatternMatchingEventHandler などを利用します。
以下は txt ファイルを指定した例になります。
・FileEvent_PatternMatch.py

# ファイル変更イベント検出のため、watchdogをインポート
from watchdog.events import PatternMatchingEventHandler
from watchdog.observers import Observer

# ファイルアクセスとスリープのため、osとtimeをインポート
import os
import time

# 監視対象ディレクトリを指定する
target_dir = 'D:\\Python\\FileEventTest\\TestDir'
# 監視対象ファイルのパターンマッチを指定する
target_file = '*.txt'

# PatternMatchingEventHandler の継承クラスを作成
class FileChangeHandler(PatternMatchingEventHandler):
     # クラス初期化
     def __init__(self, patterns):
         super(FileChangeHandler, self).__init__(patterns=patterns)

     # ファイル作成時のイベント
     def on_created(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s created' % filename)

     # ファイル変更時のイベント
     def on_modified(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s changed' % filename)

     # ファイル削除時のイベント
     def on_deleted(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s deleted' % filename)

     # ファイル移動時のイベント
     def on_moved(self, event):
         filepath = event.src_path
         filename = os.path.basename(filepath)
         print('%s moved' % filename)

# コマンド実行の確認
if __name__ == "__main__":
     # ファイル監視の開始
     event_handler = FileChangeHandler([target_file])
     observer = Observer()
     observer.schedule(event_handler, target_dir, recursive=True)
     observer.start()
     # 処理が終了しないようスリープを挟んで無限ループ
     try:
         while True:
             time.sleep(0.1)
     except KeyboardInterrupt:
         observer.stop()
     observer.join()