Python Tech

Python フォルダ内の変更を監視する Watchdog の使用例

フォルダ・ファイルの変更を監視するのは、簡単なようですが、効率良くするのには難易度が高めです。

ですが、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.

API Reference — watchdog 0.8.2 documentation

Python で,ファイル更新を監視

JPRS用語辞典|ハッシュ値(ダイジェスト値)

  • この記事を書いた人

次世代ペンギン

長いのでペンギンとお呼びください。システム開発・プログラミングのお仕事をしています。甘味とコーヒーは生命線。多くの人に役立つ情報のシェアが目標です。

人気の記事

1

会社員でプログラマーとして働いている人、インフラやネットワークのエンジニアとして働いている人の中には、フリーランスのプログラマーとして独立、もしくは転向したい人もいるので ...

2

キャリアアップのため、または高収入を目指して、しっかりプログラミングを学びたいという人が増えてきましたね。 この記事では現役のエンジニアである私が、実際に仕事で稼げるよう ...

3

フリーランスのプログラマーにとって収入の向上に最も直結するのはスキルです。 必要なスキル、スキルの獲得方法が気になる人も多いでしょう。 また、これからフリーランスを目指す ...

4

Vuetifyの v-progress-circular コンポーネントは、数値データや処理状況を環状(円状)のデザインで教えてくれるUIデザインです。 ローディングのス ...

5

Vuexのstore(ストア)を使うと、各コンポーネント間で個別にデータのやり取りすることなく、データを一元的に管理できます。Vueでは欠かせない機能といえるでしょう。 ...

-Python, Tech
-, ,

© 2021 ペンギンのーと