Watchdogを用いたファイル更新の監視について

はじめに

私が行った研究において，PythonのWatchdogを用いたフォルダ監視及び，更新イベントについて知ることが出来たので，今回得た知識をまとめました．

Watchdogとは

コンピュータや内部の特定の装置やソフトウェアが正常に稼働しているかを定期的に監視する仕組みのことで，主にファイルの作成・更新・削除を監視することが出来る．
以下のコードでファイルが変更された場合に，def on_modified(self, event):以降のコードが読み込まれる．

monitoring.pyによるファイルのファイル変更を検出するまでのコード

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

#wachdogを使用して更新を記録
class DirectoryHandler(FileSystemEventHandler):
    def __init__(self, directory_path):
        super().__init__()
        try:
            self.directory_path = directory_path
            self.file_name = ""
            self.file_changes = 0
        except FileNotFoundError:
            return

    #ファイルが変更されたら
    def on_modified(self, event):

windows環境でのファイル更新イベントについて

今回監視を行ったのが，windows環境であったため，pptxやdocファイルなどのバイナリファイルに対応させた．
具体的にはまず，is_binary()関数を用いて，更新のあったファイルがバイナリファイルかを確認する．これはwatchdogでバイナリファイルの更新イベントを感知すると，一時的なデータファイルである「.tmp」ファイルが作成されたり，それを除いた更新イベントが決まって２回発生するためである．
その後，バイナリファイルであれば，上記で述べた「.tmp」ファイルをカウントしない処理に加え，check_update_flag変数を使用して，２回目の更新イベントをカウントしないようにしている．

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

#フォルダ名と更新回数を保存する辞書
folder_dict = {}
#更新されたときにアクセス回数が増えないようにするflag
check_update_flag = False

#wachdogを使用して更新を記録
class DirectoryHandler(FileSystemEventHandler):
    def __init__(self, directory_path):
        super().__init__()
        try:
            self.directory_path = directory_path
            self.file_name = ""
            self.file_changes = 0
        except FileNotFoundError:
            return

    #ファイルが変更されたら
    def on_modified(self, event):
        global check_update_flag

        if event.is_directory:
            return
        self.file_name = os.path.basename(event.src_path)

        #アクセス回数が増加しないようにflagをTrueにする
        #check_update_flag = True

        if ("tmp" not in self.file_name) and ("~$" not in self.file_name):
            check_file_name = self.file_name #前のイベントのファイル名かを確認する変数

            #バイナリで変更された記録があれば
            if is_binary(event.src_path):
                #イベントファイルが変わったら回数を増やすようにする
                #初めての更新イベント
                if (check_update_flag == False) and (check_file_name != self.file_name):
                    self.file_changes += 1
                    directory_name = os.path.basename(self.directory_path)
                    folder_dict[directory_name][0] += 1
                    check_update_flag = True

                #二回目のファイル更新イベントをスルー    
                elif check_update_flag and (check_file_name == self.file_name):
                    check_update_flag = False

            else:
                self.file_changes += 1
                directory_name = os.path.basename(self.directory_path)
                folder_dict[directory_name][0] += 1
                check_update_flag = True


#引数のファイルがバイナリ形式かどうかを判定する
#b\oはバイナリデータによく含まれるNULLバイトだが，厳密な判定ではない
def is_binary(file_path):
    try:
        with open(file_path, 'rb') as f:
            #バイナリデータを含むかどうかを確認
            return b'\0' in f.read(1024)  #例として最初の1024バイトを読み込む
    except Exception as e:
        print(f"エラー: {e}")
        return False

終わりに

本記事では，私が研究を通じて得たPythonのWatchdogを用いたファイル更新についてまとめました．