前のページまででパイプライン処理の基本形が出来上がりました．ただ，一瞬で全ての処理が終わってしまっていました．将来的にはワーカープログラムでは文字起こしや議事録生成など，時間を要する処理を行うことを考えます．現時点では待ち行列にある複数のファイルのうちどのファイルが処理中であるかがわかりにくい状態です．どのファイルが処理中であるかがわかりやすくなるように，作業用 working フォルダに移動して，そのフォルダで処理を進めるようにしてみましょう．
具体的にはワーカープログラムを次のように修正します．なお，処理中であることがさらにわかりやすいように10秒間のスリープも設定しています．
worker.py
from pathlib import Path
import shutil
import time
BASE_DIR = Path(__file__).parent
QUEUE_DIR = BASE_DIR / "queue"
WORKING_DIR = BASE_DIR / "working"
DONE_DIR = BASE_DIR / "done"
# ディレクトリを作っておく
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
WORKING_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}")
working_file = WORKING_DIR / src.name
shutil.move(src, working_file)
print(f"作業開始: {working_file}")
with open(working_file, "r", encoding="utf-8") as f:
print("処理中...")
time.sleep(10)
print(f.read())
dst = DONE_DIR / src.name
shutil.move(working_file, 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 フォルダにマークダウンファイルを4個格納します．その結果，マークダウンファイルが待ち行列用 queue フォルダに移動し，その中から1個のファイルが作業用 working フォルダに移動し，そこで内容を表示，10秒後に完了用 done フォルダに移動します．実行中に待ち行列用 queue フォルダや作業用 working フォルダの状態を詳細に確認してください．
(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
新しいファイルを検出: C:\Users\rinsaka\Documents\python\watchdog\watch\sample3.md
新しいファイルの名前: sample3.md
新しいファイルを移動: C:\Users\rinsaka\Documents\python\watchdog\queue\sample3.md
新しいファイルを検出: C:\Users\rinsaka\Documents\python\watchdog\watch\sample4.md
新しいファイルの名前: sample4.md
新しいファイルを移動: C:\Users\rinsaka\Documents\python\watchdog\queue\sample4.md
Ctrl + C
停止します
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python worker.py
処理開始: sample1.md
作業開始: C:\Users\rinsaka\Documents\python\watchdog\working\sample1.md
処理中...
# Watchdog
Watchdogでフォルダを監視します．
完了: C:\Users\rinsaka\Documents\python\watchdog\done\sample1.md
処理開始: sample2.md
作業開始: C:\Users\rinsaka\Documents\python\watchdog\working\sample2.md
処理中...
# 自動処理
新たなファイルが保存されたときに，自動的に何らかの処理を行いたい．
完了: C:\Users\rinsaka\Documents\python\watchdog\done\sample2.md
処理開始: sample3.md
作業開始: C:\Users\rinsaka\Documents\python\watchdog\working\sample3.md
処理中...
# サンプル
3番目のファイルです．
完了: C:\Users\rinsaka\Documents\python\watchdog\done\sample3.md
処理開始: sample4.md
作業開始: C:\Users\rinsaka\Documents\python\watchdog\working\sample4.md
処理中...
# サンプル
4番目のサンプルファイルです．
完了: C:\Users\rinsaka\Documents\python\watchdog\done\sample4.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
-a--- 2026/10/25 18:19 47 sample3.md
-a--- 2026/10/25 18:19 59 sample4.md
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
なお，ここまではファイルサイズの小さなマークダウンファイルで処理をしてきました．将来的にはサイズの大きな音声ファイルなどの処理を行うことを考えます．ここで監視用フォルダに音声ファイルをダウンロードして保存しようとしたとき，音声ファイルの全てのダウンロードが完了する前に待ち行列用フォルダに移動する処理が動いてしまう可能性があります．次のページではこうした問題に対応する方法を考えます．