概要（フォルダ監視は「ファイルが増えた瞬間に自動で動く仕組み」）
フォルダ監視は、「あるフォルダにファイルが置かれたら、自動で処理を走らせる」ための仕組みです。
例としては、ダウンロードフォルダにPDFが来たら自動で整理、Excelが来たら自動集計、画像が来たらリサイズ…など。
Pythonでは主に次の2パターンがあります。
- ライブラリ watchdog を使って、OSのイベントを拾う（本格的・リアルタイム）
- 単純な「定期ポーリング」（一定間隔で中身を見て変化を検出）
初心者がまず押さえるべき重要ポイントは、
- どのフォルダを、どんなイベント（作成・変更・削除）で監視するか
- 「何が起きたら、どの関数を呼ぶか」をはっきり決める
- 二重実行・処理中ファイル・エラー時のログをどう扱うか
この3つです。
基本：watchdog を使ったフォルダ監視の流れ
watchdog をインストールする
まずはライブラリのインストールから始めます。
pip install watchdog
watchdog は、OSのファイルシステムイベント（ファイル作成・変更など）を拾ってくれるライブラリです。
最小サンプル：新しいファイルが来たら通知する
「指定フォルダに新しいファイルが作られたら、そのファイル名を表示する」だけの最小コードです。
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from pathlib import Path
import time
class MyHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return # フォルダは無視
print(f"新しいファイルが作成されました: {event.src_path}")
def main():
target_dir = Path("watch") # 監視したいフォルダ
target_dir.mkdir(exist_ok=True) # なければ作る
event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, str(target_dir), recursive=False)
observer.start()
try:
print(f"{target_dir} を監視中... Ctrl+C で停止")
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
if __name__ == "__main__":
main()
この状態で watch フォルダにファイルをコピーすると、コンソールに「新しいファイルが作成されました: …」と出ます。
ここまでで押さえておきたいポイントは、
- FileSystemEventHandler を継承したクラスを作り、イベント用メソッド（on_created など）を定義する
- Observer に「どのフォルダを」「再帰的に見るか（recursive）」を登録する
- メインループでプログラムを生かしておく（sleepでOK）
の3つです。
応用１：特定の拡張子だけを拾って処理する
例：新しい Excel が来たら自動集計を呼ぶ
「.xlsx ファイルだけを見たい」「ファイル名に特定の文字を含むものだけ処理したい」といったパターンです。
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from pathlib import Path
import time
class ExcelHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
path = Path(event.src_path)
if path.suffix.lower() != ".xlsx":
return # Excel 以外は無視
print(f"新しいExcelを検知: {path.name}")
self.process_excel(path)
def process_excel(self, path: Path):
print(f"{path} を集計します...")
# ここに集計ロジック（pandasなど）を書く
# 例: 集計結果を別フォルダに保存する 等
def main():
target_dir = Path("watch_excel")
target_dir.mkdir(exist_ok=True)
handler = ExcelHandler()
observer = Observer()
observer.schedule(handler, str(target_dir), recursive=False)
observer.start()
try:
print(f"{target_dir} の .xlsx を監視中... Ctrl+C で停止")
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
if __name__ == "__main__":
main()
ここで重要なのは、
- event.src_path を Path に変換して拡張子（suffix）でフィルタする
- 実際の処理は別メソッド（process_excel）に切り出して、テストしやすくする
という構造です。イベント処理の中には、あまり重い処理を書きすぎない方が安全です（次で詳しく）。
重要ポイントの深掘り１：処理タイミング・ファイルの「書き込み中」問題
ファイルが「まだ書き込み中」の可能性
現実世界では、「大きなファイルをコピー中」に on_created が走ってしまうことがあります。
その場合、すぐに開いて処理を始めると「壊れた状態」のファイルを読みに行ってしまう可能性があります。
典型的な対策は、
- 少し待ってから処理する（数秒 sleep）
- サイズが変化しなくなるまで待つ（ポーリング）
です。
簡単な「サイズが安定するまで待つ」例を挙げます。
import time
from pathlib import Path
def wait_until_stable(path: Path, timeout=30, interval=1):
start = time.time()
last_size = -1
while True:
if not path.exists():
raise FileNotFoundError(path)
size = path.stat().st_size
if size == last_size:
return # サイズが変わらなくなった → 安定したとみなす
last_size = size
if time.time() - start > timeout:
raise TimeoutError(f"{path} の安定待ちでタイムアウトしました")
time.sleep(interval)
これを process_excel の中の最初に呼ぶと、書き込み中のファイルを避けやすくなります。
def process_excel(self, path: Path):
wait_until_stable(path)
# ここから安全に処理を開始
「どこまでやるか」はプロジェクトの重要度次第ですが、
“書き込み中のファイルを触らない”という発想は必ず意識しておくべきポイントです。
重要ポイントの深掘り２：エラー処理・ログ・二重実行の防止
例外は握りつぶさず、ログを出す
イベントハンドラ内で例外が起きると、そのままスレッドが落ちて気づきにくくなることがあります。
最低限、try/except でエラー内容を表示・記録しておきます。
import traceback
class ExcelHandler(FileSystemEventHandler):
def on_created(self, event):
if event.is_directory:
return
path = Path(event.src_path)
if path.suffix.lower() != ".xlsx":
return
try:
self.process_excel(path)
except Exception:
print(f"エラーが発生しました: {path}")
traceback.print_exc()
業務利用なら logging モジュールでファイルに出すのがおすすめです。
同じファイルを何度も処理しない工夫
OSやアプリケーションによっては、「作成→変更」と複数イベントが飛んだり、
監視範囲やプログラム側のバグで二重実行が起きることがあります。
単純な対策としては、
- 「処理済みファイル一覧」をメモリやファイルに持っておき、二度目はスキップする
- 処理が終わったら「done」フォルダへ移動する（監視対象外の場所へ退避する）
などがあります。
processed = set()
class ExcelHandler(FileSystemEventHandler):
def on_created(self, event):
...
key = path.resolve()
if key in processed:
return # すでに処理済み
self.process_excel(path)
processed.add(key)
実運用では「処理済みファイルのログ」や「移動先フォルダ」を決めておくと安心です。
シンプルな代替案：定期ポーリングでのフォルダ監視
監視ライブラリを使わず「一定間隔で中身をチェック」する
環境や制約によって watchdog が使いにくい場合もあります。
そのときは、単純に「1分ごとにフォルダ内のファイル一覧を見て、新顔を処理する」という方法もあります。
from pathlib import Path
import time
def poll_folder(folder: Path, interval=10):
known = set(p.name for p in folder.glob("*"))
print(f"{folder} をポーリング監視中...")
while True:
current = set(p.name for p in folder.glob("*"))
new_files = current - known
for name in new_files:
path = folder / name
print(f"新しいファイルを検知: {path}")
# ここで処理を呼ぶ
# process(path)
known = current
time.sleep(interval)
if __name__ == "__main__":
folder = Path("watch_poll")
folder.mkdir(exist_ok=True)
poll_folder(folder, interval=10)
リアルタイム性は落ちますが、
- 実装がシンプル
- プラットフォーム依存性が少ない
- タスクスケジューラや cron と組み合わせやすい
という利点があります。
実運用イメージ（ダウンロードフォルダを監視して自動整理）
例：「ダウンロード」に入ったファイルを拡張子ごとに仕分ける
最後に、少し現実的な例をまとめます。
- 監視対象:
~/Downloads - PDF →
~/Downloads/pdf - 画像 →
~/Downloads/images - Excel →
~/Downloads/excel - それ以外 → 無視
という自動整理スクリプトです。
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from pathlib import Path
import shutil
import time
import os
class SortHandler(FileSystemEventHandler):
def __init__(self, base_dir: Path):
self.base_dir = base_dir
self.dir_pdf = base_dir / "pdf"
self.dir_img = base_dir / "images"
self.dir_xlsx = base_dir / "excel"
for d in [self.dir_pdf, self.dir_img, self.dir_xlsx]:
d.mkdir(exist_ok=True)
def on_created(self, event):
if event.is_directory:
return
path = Path(event.src_path)
ext = path.suffix.lower()
# 一瞬待ってから（書き込み中対策として）
time.sleep(1)
if ext in [".pdf"]:
dest = self.dir_pdf / path.name
elif ext in [".png", ".jpg", ".jpeg"]:
dest = self.dir_img / path.name
elif ext in [".xlsx", ".xls"]:
dest = self.dir_xlsx / path.name
else:
return
print(f"{path.name} → {dest.parent.name} に移動")
shutil.move(str(path), str(dest))
def main():
base_dir = Path(os.path.expanduser("~/Downloads"))
handler = SortHandler(base_dir)
observer = Observer()
observer.schedule(handler, str(base_dir), recursive=False)
observer.start()
try:
print(f"{base_dir} を監視中... Ctrl+C で停止")
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
if __name__ == "__main__":
main()
実際に動かすと、「ダウンロードした瞬間に自動でフォルダ分け」されます。
まとめ（「何を監視し、何が起きたら何をするか」をコードに落とす）
フォルダ監視の本質は、次の3行に集約されます。
- どのフォルダを監視するか
- どの種類のイベントを拾うか（作成・変更・削除）
- そのイベントが起きたときに、どの処理を呼ぶか
watchdog を使えばリアルタイムにイベントを拾えますし、
単純なポーリングでも「定期的な自動処理」として十分役に立ちます。
大事なのは、「書き込み中のファイルを触らない」「二重実行を避ける」
「エラーやログをきちんと残す」などの細かい設計です。