ここまでの作業で監視用の watch フォルダを監視するし待ち行列用 queue フォルダへ移動する基本的なプログラム (watch.py) と，待ち行列用 quque フォルダに並んだファイル順に処理し完了用 done フォルダに移動する基本的なプログラム (worker.py) ができました．
現時点の watch.py は次のとおりです．
watch.py
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import shutil # ファイルのコピー，移動などを行う標準ライブラリ
import time
BASE_DIR = Path(__file__).parent
WATCH_DIR = BASE_DIR / "watch"
QUEUE_DIR = BASE_DIR / "queue"
# ディレクトリを作っておく
WATCH_DIR.mkdir(parents=True, exist_ok=True)
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
class MyHandler(FileSystemEventHandler):
"""
watchdog が用意している FileSystemEventHandler クラスの継承
「ファイル監視イベントが発生したときに何をするか」を定義する
"""
def on_created(self, event):
if event.is_directory: # ディレクトリは無視
return
src = Path(event.src_path)
print(f"新しいファイルを検出: {src}")
print(f"新しいファイルの名前: {src.name}")
dst = QUEUE_DIR / src.name # 移動先の指定
shutil.move(src, dst) # ファイルの移動
print(f"新しいファイルを移動: {dst}")
observer = Observer() # 「監視員を作成する」処理
# 監視条件設定
observer.schedule(
MyHandler(),
path=WATCH_DIR,
recursive=False # とりあえず直下だけを監視
)
observer.start() # 監視を始める → 別スレッド（observerスレッド）が生成される
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("停止します")
observer.stop()
observer.join() # observerスレッドが終わるまで待つ
また worker.py は次のとおりの内容です．
worker.py
from pathlib import Path
import shutil
import time
BASE_DIR = Path(__file__).parent
QUEUE_DIR = BASE_DIR / "queue"
DONE_DIR = BASE_DIR / "done"
# ディレクトリを作っておく
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
DONE_DIR.mkdir(parents=True, exist_ok=True)
try:
while True:
for src in QUEUE_DIR.glob("*.md"): # 拡張子が .md のファイルだけ
print(f"処理開始: {src.name}")
with open(src, "r", encoding="utf-8") as f:
print(f.read())
dst = DONE_DIR / src.name
shutil.move(src, dst)
print(f"完了: {dst}")
time.sleep(1)
except KeyboardInterrupt:
print("停止します")
ターミナルを2個起動し，それぞれで watch.py と worker.py を実行します．
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python watch.py ⏎
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python worker.py ⏎
2つのプログラムが開始されたら，監視用 watch フォルダにマークダウンファイルを2個格納します．その結果，マークダウンファイルが待ち行列用 queue フォルダに移動し，それらの内容が表示された後，完了用 done フォルダに移動しました．
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python watch.py
新しいファイルを検出: C:\Users\rinsaka\Documents\python\watchdog\watch\sample1.md
新しいファイルの名前: sample1.md
新しいファイルを移動: C:\Users\rinsaka\Documents\python\watchdog\queue\sample1.md
新しいファイルを検出: C:\Users\rinsaka\Documents\python\watchdog\watch\sample2.md
新しいファイルの名前: sample2.md
新しいファイルを移動: C:\Users\rinsaka\Documents\python\watchdog\queue\sample2.md
Ctrl + C
停止します
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python worker.py
処理開始: sample1.md
# Watchdog
Watchdogでフォルダを監視します．
完了: C:\Users\rinsaka\Documents\python\watchdog\done\sample1.md
処理開始: sample2.md
# 自動処理
新たなファイルが保存されたときに，自動的に何らかの処理を行いたい．
完了: C:\Users\rinsaka\Documents\python\watchdog\done\sample2.md
Ctrl + C
停止します
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
完了用 done フォルダにファイルが移動したことを確認しておきます．
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> gci done ⏎
ディレクトリ:C:\Users\rinsaka\Documents\python\watchdog\done
Mode LastWriteTime Length Name
---- ------------- ------ ----
-a--- 2026/10/25 18:18 56 sample1.md
-a--- 2026/10/25 18:18 115 sample2.md
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
このように，監視用 watch フォルダにファイルを設置するだけで，待ち行列を形成し，順番に何らかの処理を行うパイプライン処理の基本形が出来上がりました．今の状態ではファイルを設置して瞬時にすべての処理が終了してしまいます．将来的にはワーカープログラムでは文字起こしや議事録生成など，時間を要する処理を行うことを考えます．現時点では待ち行列にある複数のファイルのうちどのファイルが処理中であるかがわかりにくい状態です．どのファイルが処理中であるかがわかりやすくなるように次のページでは作業用 working フォルダに移動して，そのフォルダで処理を進めるようにしてみましょう．