まず，フォルダ監視プログラムを再び示します．これは以前のコードから変更ありません．
watch.py
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import shutil # ファイルのコピー，移動などを行う標準ライブラリ
import time
import datetime
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)
def wait_until_complete(path):
"""
watch ディレクトリへのコピー，ファイルサイズの変化がない -> コピー（ダウンロード）が終わったと判断
待っている間にファイル名が変更されたら強制終了
"""
while True:
try:
size1 = path.stat().st_size
time.sleep(5)
size2 = path.stat().st_size
time.sleep(5)
size3 = path.stat().st_size
if size1 == size2 == size3:
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] ファイルの受取りが完了しました")
return True
except FileNotFoundError:
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] ファイルが見つかりません: {path}") # 途中でファイル名を変更したり，削除した場合
return False
class MyHandler(FileSystemEventHandler):
"""
watchdog が用意している FileSystemEventHandler クラスの継承
「ファイル監視イベントが発生したときに何をするか」を定義する
"""
def on_created(self, event):
if event.is_directory: # ディレクトリは無視
return
src = Path(event.src_path)
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 新しいファイルを検出: {src}")
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 新しいファイルの名前: {src.name}")
if not wait_until_complete(src):
return
dst = QUEUE_DIR / src.name # 移動先の指定
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] ファイルの移動開始: {dst}")
shutil.move(src, dst) # ファイルの移動
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] ファイルの移動完了: {dst}")
observer = Observer() # 「監視員を作成する」処理
# 監視条件設定
observer.schedule(
MyHandler(),
path=WATCH_DIR,
recursive=False # とりあえず直下だけを監視
)
observer.start() # 監視を始める → 別スレッド（observerスレッド）が生成される
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 停止します")
observer.stop()
observer.join() # observerスレッドが終わるまで待つ
ワーカープログラムについては，Windows 環境では WhisperX が UnicodeEncodeError を起こすことがあるため UTF-8 を強制するよう次のとおり修正します．
worker.py
from pathlib import Path
import shutil
import time
import datetime
import subprocess
import os
BASE_DIR = Path(__file__).parent
QUEUE_DIR = BASE_DIR / "queue"
WORKING_DIR = BASE_DIR / "working"
DONE_DIR = BASE_DIR / "done"
ERROR_DIR = BASE_DIR / "error"
# ディレクトリを作っておく
QUEUE_DIR.mkdir(parents=True, exist_ok=True)
WORKING_DIR.mkdir(parents=True, exist_ok=True)
DONE_DIR.mkdir(parents=True, exist_ok=True)
ERROR_DIR.mkdir(parents=True, exist_ok=True)
def ask_ollama(prompt):
result = subprocess.run(
["ollama", "run", "qwen3:8b"],
input=prompt,
capture_output=True,
text=True,
encoding="utf-8"
)
return result.stdout
try:
while True:
for src in QUEUE_DIR.glob("*.m*"): # 監視するファイルの拡張子
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] ----------------")
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 処理開始: {src.name}")
working_file = WORKING_DIR / src.name
transcription_dir = WORKING_DIR / src.stem
transcription_dir.mkdir(parents=True, exist_ok=True) # ディレクトリを作っておく
log_file = transcription_dir / "whisperx.log"
shutil.move(src, working_file)
# 文字起こし -----------------------------------
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 作業開始: {working_file}")
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 出力先 : {transcription_dir}")
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 文字起こし処理中...")
# Windows環境で WhisperX が UnicodeEncodeError を起こすことがあるため UTF-8 を強制する
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
try:
with open(log_file, "w", encoding="utf-8") as f:
subprocess.run([
"whisperx",
"--model", "turbo", # tubo / large ...
"--language", "ja",
# "--device", "cuda", # NVIDIA GPU を利用できる場合
"-o", str(transcription_dir),
"--diarize", # 話者分離
str(working_file)
],
env=env, #
stdout=f,
stderr=subprocess.STDOUT,
check=True)
except subprocess.CalledProcessError:
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] WhisperXの処理に失敗しました")
dt = datetime.datetime.now()
err_dir_name = f"{dt.strftime('%Y%m%d-%H%M%S')}_{src.stem}"
err_dst = ERROR_DIR / err_dir_name / src.name
err_transcription_dir = ERROR_DIR / err_dir_name / src.stem
err_transcription_dir.mkdir(parents=True, exist_ok=True) # ディレクトリを作っておく
shutil.move(working_file, err_dst)
shutil.move(transcription_dir, err_transcription_dir)
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 失敗: {err_dst}")
continue
time.sleep(3)
# 議事録生成 -----------------------------------
transcription_file_name = src.stem + ".txt"
transcription_file_path = transcription_dir / transcription_file_name
minutes_file_name = src.stem + ".md" # 議事録ファイル名
minutes_file = transcription_dir / minutes_file_name
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 議事録生成中...")
time.sleep(3)
with open(transcription_file_path, encoding="utf-8") as f:
transcript = f.read()
prompt = f"""
次の文字起こし結果から議事録を作成してください。
{transcript}
"""
minutes = ask_ollama(prompt)
with open(minutes_file, "w", encoding="utf-8") as f:
f.write(minutes)
# print(minutes)
# print("-----------------")
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 議事録生成完了")
# 作業完了 → ファイルの移動 -----------------------------------
dt = datetime.datetime.now()
done_dir_name = f"{dt.strftime('%Y%m%d-%H%M%S')}_{src.stem}"
done_file = DONE_DIR / done_dir_name / src.name
done_transcription_dir = DONE_DIR / done_dir_name
done_transcription_dir.mkdir(parents=True, exist_ok=True) # ディレクトリを作っておく
shutil.move(working_file, done_file)
shutil.move(transcription_dir, done_transcription_dir)
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 完了: {done_file}")
time.sleep(1)
except KeyboardInterrupt:
print(f"[{datetime.datetime.now().strftime('%H:%M:%S.%f')}] 停止します")
ターミナルを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 ⏎
前のページでエラーになった音声ファイルを再度監視フォルダに格納して処理を開始します．
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python watch.py
[18:44:26.316423] 新しいファイルを検出: C:\Users\rinsaka\Documents\python\watchdog\watch\voice-sample-error.m4a
[18:44:26.316868] 新しいファイルの名前: voice-sample-error.m4a
[18:44:36.321152] ファイルの受取りが完了しました
[18:44:36.321367] ファイルの移動開始: C:\Users\rinsaka\Documents\python\watchdog\queue\voice-sample-error.m4a
[18:44:36.322199] ファイルの移動完了: C:\Users\rinsaka\Documents\python\watchdog\queue\voice-sample-error.m4a
Ctrl + C
[18:52:36.762463] 停止します
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
文字コードを強制したことで問題なく処理することができました．
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog> python worker.py
[18:44:36.357295] ----------------
[18:44:36.357405] 処理開始: voice-sample-error.m4a
[18:44:36.358212] 作業開始: C:\Users\rinsaka\Documents\python\watchdog\working\voice-sample-error.m4a
[18:44:36.358277] 出力先 : C:\Users\rinsaka\Documents\python\watchdog\working\voice-sample-error
[18:44:36.358346] 文字起こし処理中...
[18:49:11.615879] 議事録生成中...
[18:49:49.479124] 議事録生成完了
[18:49:49.480362] 完了: C:\Users\rinsaka\Documents\python\watchdog\done\20261102-184949_voice-sample-error\voice-sample-error.m4a
Ctrl + C
[18:52:35.834636] 停止します
(py313watchdog) PS C:\Users\rinsaka\Documents\python\watchdog>
あとは，プロンプトの書き方などを色々工夫すると良いでしょう．