フォルダ・ファイルの変更を監視するのは、簡単なようですが、効率良くするのには難易度が高めです。
ですが、Pythonでは手軽に効率的なフォルダ・ファイル監視を実装できる外部ライブラリが使えます。
今回解説する Watchdog （英語で「番犬」の意味）はOSのファイル管理機能を利用しており、非常に効率的に状態変化を監視することが可能。
このページでは、基本的なWatchdogの使い方やコードサンプルを示しながら、現役のエンジニアが分かりやすく解説しています。
- 特定のフォルダ/ファイルの作成・変更・削除を監視
- あるパターンにマッチする名前のファイルを監視する方法
- ファイルの中身の変更を監視
実行環境
今回検証している環境は次のとおりです。
Debian GNU/Linux 10 (buster)
$ python --version
Python 3.9.2
$ pip freeze
watchdog==2.1.6
※通常新しいバージョンなら動きます。無理にバージョンを一致させなくてもOK。
Watchdogを使用する準備
Watchdog はWindows、Mac（またはLinux）のいずれでも動作します。
以下の解説はPythonがインストールされていることを前提にしています。最新バージョンか、サポートされているPython 3.6以上のインストールが必要です。
まずはWatchdogをインストールします。pipなら次のようにコマンドを実行するだけ。
$ pip install watchdog
基本的な使い方
まずは、公式のドキュメントにあるサンプルコードを実行してみます。
import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
path = sys.argv[1] if len(sys.argv) > 1 else '.'
event_handler = LoggingEventHandler()
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
例えば "file_watchdog.py" などの名前でPythonファイルを作成し、上記をそのままコピー＆ペーストして保存します。その後、次のようにスクリプトを実行します。
$ python file_watchdog.py
今実行しているPythonファイルがある同じフォルダで、ファイルを作成・更新・削除の操作をしてみます。そうすると次のようにコンソールに出力されます。
2021-10-07 16:29:43 - Created file: ./test.txt
2021-10-07 16:29:43 - Modified directory: .
2021-10-07 16:29:43 - Modified directory: .
2021-10-07 16:29:54 - Modified file: ./test.txt
2021-10-07 16:29:54 - Modified file: ./test.txt
2021-10-07 16:29:54 - Modified directory: .
2021-10-07 16:30:04 - Deleted file: ./test.txt
2021-10-07 16:30:04 - Modified directory: .
フォルダ内のファイルの更新状況を一覧でチェックできました。かんたんですね。
終了するときは Ctrl+C を押下します。
サンプルコードで重要な部分は次の箇所になります。
event_handler = LoggingEventHandler()
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
observer.start()
Watchdogの実装では、まずEventHandlerのインスタンスを作り、それをObserver.scheduleに登録することで、ファイルの状態変化を検知したときの動作を指定します。
後述するように、EventHandlerクラスを継承したクラスを作成して自由に動作を記述できます。
これでWatchdogの基本的な動作が確認できました。
特定のファイルを監視
指定フォルダ内の特定のファイルだけを監視したい場合はどうすればいいでしょうか？
例えば testフォルダの拡張子が .txt のファイルを監視対象にする場合を考えてみます。
この要件では、LoggingEventHandler の代わりに PatternMatchingEventHandler を用います。対象ファイル名のパターン（下記のPATTERN）ではワイルドカード *を使って「拡張子.txtのすべてのファイル」を指定します。
また、PatternMatchingEventHandlerはファイル変更検知時の動作を定めていないので、自前で on_modified() 関数を用意し置き換えます。
file_watchdog_pattern.py
from watchdog.events import PatternMatchingEventHandler
from watchdog.observers import Observer
import os
import time
if __name__ == "__main__":
# 対象ディレクトリ
DIR_WATCH = './test'
# 対象ファイル名のパターン
PATTERNS = ['*.txt']
def on_modified(event):
filepath = event.src_path
filename = os.path.basename(filepath)
print('%s changed' % filename)
# event_handler = LoggingEventHandler()
event_handler = PatternMatchingEventHandler(PATTERNS)
event_handler.on_modified = on_modified
observer = Observer()
observer.schedule(event_handler, DIR_WATCH, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
実行方法は同じです。
$ python file_watchdog_pattern.py
終了するときは Ctrl+C を押下します。
ファイルを新規作成する touch コマンド、文字列を挿入する echo ... >> [file] で動作を確認してみます。
$ python file_watchdog_pattern.py & ←バックグラウンドで実行させる
$ touch ./test/test.txt
test.txt changed
$ echo "hoge" >> ./test/test.txt
test.txt changed
$ touch ./test/test.csv ←反応なし
$ echo "hoge" >> ./test/test.csv ←反応なし
.txtのファイルにだけ反応することが確認できました。また、表示内容はon_modified()で指定したとおりに出力されています。
パターンにマッチする名前のファイルを監視
ファイル名にもっと複雑なパターンを指定したい場合は、イベントハンドラとしてPatternMatchingEventHandler の代わりに RegexMatchingEventHandlerを利用します。
複雑なパターンとは、例えばこのようなもの。
- 先頭が数値のみでアンダーバー"_"が続きその後は英数字のみで、拡張子が.csv
- 先頭が hoge か fuga で、ハイフン"-"が続きその後数字のみで、拡張子が.xlsx
- ・・・
RegexMatchingEventHandlerではパターンの指定に正規表現を使用します。
正規表現は理解するのが大変ですが、とても強力なパターンマッチができる仕組みです。ここでは深く踏み込みませんが、基本的な部分だけでも理解しておくと、後々かなり役立つはず。
今回のサンプルでは「testで始まり、終わりが.txtとなっているファイル」を対象としたいので、抽出パターンは ^\./test.*\.txt$ としています。
コードの構造はPatternMatchingEventHandlerと同じです。
from watchdog.events import RegexMatchingEventHandler
from watchdog.observers import Observer
import os
import time
if __name__ == "__main__":
# 対象ディレクトリ
DIR_WATCH = './test'
# 対象ファイルパスのパターン
PATTERNS = [r'^\./test.*\.txt$']
def on_modified(event):
filepath = event.src_path
filename = os.path.basename(filepath)
print('%s changed' % filename)
event_handler = RegexMatchingEventHandler(PATTERNS)
observer = Observer()
observer.schedule(event_handler, DIR_WATCH, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
各イベント時の動作を変更
監視対象としているフォルダ・ファイルが作成・変更・削除されるなどのイベントを検知したときの動作（イベントハンドラ）を変更できます。
上のサンプルでは、LoggingEventHandlerクラスを使用すると、単純なログメッセージが表示されるだけでした。
独自のイベントハンドラを作成するには、PatternMatchingEventHandlerやRegexMatchingEventHandlerなど、FileSystemEventHandlerを基底とするクラスを継承したクラスを作成します。
以下のサンプルでは、説明のためにMyFileWatchHandlerクラスを作成し、単純なメッセージが表示されるようにしています。
file_watchdog_reg.py
from watchdog.events import RegexMatchingEventHandler
from watchdog.observers import Observer
import os
import time
import datetime
class MyFileWatchHandler(RegexMatchingEventHandler):
def __init__(self, regexes):
super().__init__(regexes=regexes)
# ファイル作成時の動作
def on_created(self, event):
filepath = event.src_path
filename = os.path.basename(filepath)
print(f"{datetime.datetime.now()} {filename} created")
# ファイル変更時の動作
def on_modified(self, event):
filepath = event.src_path
filename = os.path.basename(filepath)
print(f"{datetime.datetime.now()} {filename} changed")
# ファイル削除時の動作
def on_deleted(self, event):
filepath = event.src_path
filename = os.path.basename(filepath)
print(f"{datetime.datetime.now()} {filename} deleted")
# ファイル移動時の動作
def on_moved(self, event):
filepath = event.src_path
filename = os.path.basename(filepath)
print(f"{datetime.datetime.now()} {filename} moved")
if __name__ == "__main__":
# 対象ディレクトリ
DIR_WATCH = './test'
# 対象ファイルパスのパターン
PATTERNS = [r'^\./test.*\.txt$']
def on_modified(event):
filepath = event.src_path
filename = os.path.basename(filepath)
print('%s changed' % filename)
event_handler = MyFileWatchHandler(PATTERNS)
observer = Observer()
observer.schedule(event_handler, DIR_WATCH, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
例えば、ファイルの作成が検知されたときの動作を変更するには on_created()、更新が検知されたときの動作を変更するには on_modified() を修正するといった具合です。
ファイルが更新されたときに、そのファイルサイズも一緒に表示させたいとします。ファイルサイズは os パッケージの関数 os.path.getsize() で取得できるので、次のようにon_modified()を修正します。
def on_modified(self, event):
filepath = event.src_path
filename = os.path.basename(filepath)
dt_now = datetime.datetime.now()
fsize = os.path.getsize(filepath)
print(f"{dt_now} {filename}")
print(f"-- size: {fsize}")$ python file_watchdog_reg.py
2021-10-07 19:03:01.603036 test_bbb.txt
-- size: 0
2021-10-07 19:03:01.603647 test_bbb.txt
-- size: 8
ファイル内容の変更を検知
FileSystemEventHandler.on_modified()はファイルのタイムスタンプが更新されるタイミングで呼ばれるため、ファイルの内容が一切変わっていないときでも実行される場合があります。
ファイル内容が実質的に変更されている場合に限って検知するには少々工夫が必要です。
今回は、ハッシュ（Hash）を使い、ファイル内容の変更を検知する方法を解説します。
ハッシュ（ハッシュ値）とは、任意のデータを使って算出する短い値で、要約値やダイジェスト値とも表現されます。ハッシュ技術は主にデータベースの検索のキーとして使われたり、改ざんの防止、暗号化技術などに役立っています。
Pythonでは、標準ライブラリ hashlib を利用して、ハッシュ関数 MD5 などを用いたハッシュの算出ができます。次の例のように、少し文字列が変わるだけで、ほとんど全然違うハッシュが算出されます。
import hashlib
text = b'foo bar baz'
hash = hashlib.md5(text).hexdigest()
print(hash)
# ab07acbb1e496801937adfa772424bf7
text = b'foo bar bazx'
hash = hashlib.md5(text).hexdigest()
print(hash)
# 6916d8bb3a33af3119656b257496ad8c
ハッシュ関数を使って、ファイル変更時に内容も変化しているか判定するイベントハンドラ MyFileWatchHandler は次のように書けるでしょう。
class MyFileWatchHandler(RegexMatchingEventHandler):
def __init__(self, regexes, target_dir='./'):
self._target_dir = target_dir
super().__init__(
regexes=regexes,
ignore_directories=True,
)
# 各ファイルのハッシュ（初期値）を取得
self._hash_cur = {}
self._init_hash()
def on_modified(self, event):
"""ファイル変更時の動作
"""
filepath = event.src_path
filename = os.path.basename(filepath)
# 新旧ハッシュ値の取得
hash_new = self._get_hash(filepath)
try:
hash_old = self._hash_cur[filepath]
except KeyError:
hash_old = None
# ハッシュ値比較によるファイルの変更判定
if hash_old != hash_new:
print(f"{datetime.datetime.now()} {filename} Something changed!")
# 次回更新時に変更チェックするためのハッシュ値格納
self._hash_cur[filepath] = self._get_hash(filepath)
def _get_hash(self, filepath: str) -> str:
"""ファイル変更検知のためのハッシュ値を取得
"""
with open(filepath, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
def _init_hash(self) -> None:
"""対象ディレクトリの全ファイルについてハッシュを取得
"""
paths = glob.glob(f'{self._target_dir}/*')
for p in paths:
if os.path.isfile(p):
self._hash_cur[p] = self._get_hash(p)
Watchdogによってファイルの更新が検知される（on_modified()が呼ばれる）たびに、そのファイルのハッシュを計算します。得られたハッシュと前回計算したハッシュ値を比べて、異なる場合はファイル内容が変化していると判断します。
このイベントハンドラを利用する側のコードはほとんど同じです。
if __name__ == "__main__":
# 対象ディレクトリ
DIR_WATCH = './test'
# 対象ファイルパスのパターン
PATTERNS = ['^\./test.*\.txt$']
event_handler = MyFileWatchHandler(PATTERNS, target_dir=DIR_WATCH)
observer = Observer()
observer.schedule(event_handler, DIR_WATCH, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
実際に使用するときには環境（OS）に合わせて修正が必要なほか、例外処理を適宜補うべきです。注意してください。
今回参考にしたページ・資料
gorakhargosh/watchdog: Python library and shell utilities to monitor filesystem events.
