はじめに
どうもMizokeiです。
普段業務をしていると、ファイルやフォルダを監視したりして、その後のアクションに繋げたい事があるかと思います。
今日は、watchdogというライブラリを使用して、ファイルやフォルダを監視するスクリプトを作ってみたいと思います。ちなみに、今回の実行環境は以下です。
実行環境
- Windows10 64bit
- Python 3.7
いろんな処理に応用可能
今回作成する監視スクリプトによって、ファイルやフォルダの監視ができると、いろいろな処理に応用することが可能です。普段のプログラミングの参考にしてみてください。
- ファイルが格納されたらメールで通知する
- ファイルが格納されたら対象ファイルを移動する
- フォルダが変更されていないかを確認する…etc
watchdogをインストール
まずは今回使用するwatchdogをpipを使ってインストールしましょう。
|
0 1 2 |
pip install watchdog |
以下のように表示されるとインストールは完了です。
まずはライブラリをインポート
まずはいくつかのライブラリをインポートしましょう。フォルダ階層としてフォルダパスを取得するために、osや、一定時間の待機のためにtimeモジュールもインポートします。
インポート対象一覧
- os
- time
- FileSystemEventHandler
- Observer
以下、対象コードです。
|
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 |
# # 標準ライブラリ # #ファイルパスの確認 import os #待機 import time # # 追加ライブラリ # #ファイル・フォルダ監視用 from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer |
次にイベントハンドラを定義する
また、ファイルが作成されたり変更されたりしたときのために、イベントハンドラを定義します。これによって、変更があった場合は、それぞれのイベントに対応した関数が呼ばれる仕組みです。
|
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
#イベントハンドラ class ChangeHandler(FileSystemEventHandler): #ファイルやフォルダが作成された場合 def on_created(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを作成しました。' % filename) #ファイルやフォルダが更新された場合 def on_modified(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを変更しました。' % filename) #ファイルやフォルダが移動された場合 def on_moved(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを移動しました。' % filename) #ファイルやフォルダが削除された場合 def on_deleted(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを削除しました。' % filename) |
最後にフォルダ監視処理を記載する
最後に、対象フォルダの監視を行う処理を記載しましょう。今回は、スクリプトが実行しているフォルダパスを取得し、同一フォルダパス内でのイベント監視してみます。以下が対象コードです。
|
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 |
#メイン処理 if __name__ == '__main__': #起動ログ print('フォルダ・ファイル監視スクリプトを起動します。') #現在のフォルダパスを取得する(プログラムが実行されているフォルダパス) current_directory = os.path.dirname(os.path.abspath(__file__)) #インスタンス作成 event_handler = ChangeHandler() observer = Observer() #フォルダの監視 observer.schedule(event_handler, current_directory, recursive=True) #監視の開始 observer.start() try: #無限ループ while True: #待機 time.sleep(0.05) except KeyboardInterrupt: #監視の終了 observer.stop() #スレッド停止を待つ observer.join() #終了ログ print('フォルダ・ファイル監視スクリプトを終了します。') |
起動のログ
簡単に説明すると、最初に実行時のログを表示させています。
|
0 1 2 3 |
#起動ログ print('フォルダ・ファイル監視スクリプトを起動します。') |
pythonスクリプトのフォルダパスを取得
実行しているpythonスクリプトの階層を監視するために、フォルダパスを絶対パスで取得しています。
|
0 1 2 3 |
#現在のフォルダパスを取得する(プログラムが実行されているフォルダパス) current_directory = os.path.dirname(os.path.abspath(__file__)) |
フォルダ監視の開始
以下のコードでフォルダ監視の準備と監視を開始しています。
|
0 1 2 3 4 5 6 |
#フォルダの監視 observer.schedule(event_handler, current_directory, recursive=True) #監視の開始 observer.start() |
キー操作が無い限り監視
一回の監視で終了させないために、while文が使用されています。while文内で待機処理を行うことで、定期的にフォルダ内のイベントを監視する仕組みです。
また、ctrl+Cなどで、キー操作が行われるとエラーを発生させ、監視は終了します。同時に終了ログも表示させています。
|
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
try: #無限ループ while True: #待機 time.sleep(0.05) except KeyboardInterrupt: #監視の終了 observer.stop() #スレッド停止を待つ observer.join() #終了ログ print('フォルダ・ファイル監視スクリプトを終了します。') |
バッググラウンドのスレッドを待機
以下のコードですが、実際には、バッググラウンドでスレッドが動作しているため、停止するまで待機し、プログラムを終了する流れです。
|
0 1 2 3 |
#スレッド停止を待つ observer.join() |
実行してみると…
作成した監視スクリプトを実行し、いくつかのファイル操作をしてみます。
- 新しいテキストドキュメントを作成する
- 新しいテキストドキュメントをhuga.txtに名前変更する
- 新しいビットマップイメージ.bmpを作成する
すると、以下のような表示となりました。どうもファイルを作成したタイミングで、変更するイベントも呼ばれているみたいですね。
また、ファイル名が変更されたイベントが、対象の名前を持つファイルがなくなった(移動した)ものと、ファイル名が変更されたものの２つのイベントも走っているようです。
このあたりは少し使い方を工夫する必要がありそうですね。
今回作成した監視スクリプトの全文
では、今回作成したファイルやフォルダを監視するpythonスクリプトは以下です。他の処理と組み合わせて、使ってみてください。
|
0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
# -*- coding: utf-8 -*- #!/usr/bin/env python ''' ファイルやフォルダを監視するスクリプト。 ''' # # 標準ライブラリ # #ファイルパスの確認 import os #待機 import time # # 追加ライブラリ # #ファイル・フォルダ監視用 from watchdog.events import FileSystemEventHandler from watchdog.observers import Observer #イベントハンドラ class ChangeHandler(FileSystemEventHandler): #ファイルやフォルダが作成された場合 def on_created(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを作成しました。' % filename) #ファイルやフォルダが更新された場合 def on_modified(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを変更しました。' % filename) #ファイルやフォルダが移動された場合 def on_moved(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを移動しました。' % filename) #ファイルやフォルダが削除された場合 def on_deleted(self, event): filepath = event.src_path filename = os.path.basename(filepath) print('%sを削除しました。' % filename) #メイン処理 if __name__ == '__main__': #起動ログ print('フォルダ・ファイル監視スクリプトを起動します。') #現在のフォルダパスを取得する(プログラムが実行されているフォルダパス) current_directory = os.path.dirname(os.path.abspath(__file__)) #インスタンス作成 event_handler = ChangeHandler() observer = Observer() #フォルダの監視 observer.schedule(event_handler, current_directory, recursive=True) #監視の開始 observer.start() try: #無限ループ while True: #待機 time.sleep(0.05) except KeyboardInterrupt: #監視の終了 observer.stop() #スレッド停止を待つ observer.join() #終了ログ print('フォルダ・ファイル監視スクリプトを終了します。') |
この記事が誰かのお役に立てれば幸いです。それでは。
関連記事
【Python】ピクセル値を操作して画像を作成してみよう【画像処理】
【Python×業務自動化】Excelを起動する【RPA】
【Python超入門】クラスを作成してみよう