bdc81cbc2e
- Chunk Text: word-level fallback for sentences > 600 chars - watch.py: keep MP3 after transcription - Respond Accepted: remove cross-node reference - rocinante/: Whisper CUDA + watcher stack for RTX 4090 transcription
214 lines
7.1 KiB
Python
214 lines
7.1 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Class recording watcher for Rocinante.
|
|
|
|
Watches C:/Recordings (mounted as /recordings) for new .wav files.
|
|
For each file detected:
|
|
1. Waits for the file to finish being written
|
|
2. Converts WAV -> MP3 via ffmpeg (deletes original WAV)
|
|
3. Transcribes via local faster-whisper-server (large-v3, CUDA)
|
|
4. POSTs transcript + metadata to n8n /webhook/class/ingest
|
|
5. Deletes the MP3
|
|
|
|
Folder convention: subfolder name = class name
|
|
/recordings/HIST-2020/lecture-03-3.wav -> class=HIST-2020, title=lecture 03 3
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import subprocess
|
|
import logging
|
|
import requests
|
|
from pathlib import Path
|
|
from watchdog.observers.polling import PollingObserver as Observer
|
|
from watchdog.events import FileSystemEventHandler
|
|
|
|
# Config from environment
|
|
WHISPER_URL = os.environ.get("WHISPER_URL", "http://whisper:8000")
|
|
N8N_WEBHOOK = os.environ.get("N8N_WEBHOOK_URL", "https://n8n.paccoco.com/webhook/class/ingest")
|
|
WATCH_DIR = os.environ.get("WATCH_DIR", "/recordings")
|
|
WHISPER_MODEL = os.environ.get("WHISPER_MODEL", "large-v3")
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)-8s %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
log = logging.getLogger("watcher")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def wait_for_file_stable(path: Path, interval: float = 2.0, required_stable: int = 3) -> None:
|
|
"""Block until the file size stops changing (i.e. the copy is complete)."""
|
|
log.info(f"Waiting for {path.name} to finish writing...")
|
|
prev_size = -1
|
|
stable_count = 0
|
|
while stable_count < required_stable:
|
|
try:
|
|
size = path.stat().st_size
|
|
except FileNotFoundError:
|
|
time.sleep(interval)
|
|
continue
|
|
if size == prev_size and size > 0:
|
|
stable_count += 1
|
|
else:
|
|
stable_count = 0
|
|
prev_size = size
|
|
time.sleep(interval)
|
|
log.info(f"{path.name} is stable ({prev_size:,} bytes)")
|
|
|
|
|
|
def convert_to_mp3(wav_path: Path) -> Path:
|
|
mp3_path = wav_path.with_suffix(".mp3")
|
|
log.info(f"Converting to MP3: {wav_path.name}")
|
|
result = subprocess.run(
|
|
[
|
|
"ffmpeg", "-y",
|
|
"-i", str(wav_path),
|
|
"-codec:a", "libmp3lame",
|
|
"-qscale:a", "4",
|
|
str(mp3_path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"ffmpeg failed:\n{result.stderr}")
|
|
wav_path.unlink()
|
|
log.info(f"Deleted original WAV: {wav_path.name}")
|
|
return mp3_path
|
|
|
|
|
|
def transcribe(mp3_path: Path) -> dict:
|
|
log.info(f"Transcribing {mp3_path.name} with {WHISPER_MODEL}...")
|
|
with open(mp3_path, "rb") as f:
|
|
resp = requests.post(
|
|
f"{WHISPER_URL}/v1/audio/transcriptions",
|
|
files={"file": (mp3_path.name, f, "audio/mpeg")},
|
|
data={
|
|
"model": WHISPER_MODEL,
|
|
"language": "en",
|
|
"response_format": "verbose_json",
|
|
},
|
|
timeout=7200, # 2 hours — large recordings on first load can be slow
|
|
)
|
|
resp.raise_for_status()
|
|
log.info("Transcription complete")
|
|
return resp.json()
|
|
|
|
|
|
def build_timestamped(segments: list) -> str:
|
|
lines = []
|
|
for s in segments:
|
|
mins = int(s.get("start", 0) // 60)
|
|
secs = int(s.get("start", 0) % 60)
|
|
lines.append(f"[{mins}:{secs:02d}] {s.get('text', '').strip()}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def send_to_n8n(class_name: str, lecture_title: str, result: dict) -> None:
|
|
segments = result.get("segments", [])
|
|
full_text = result.get("text", "").strip()
|
|
duration = result.get("duration") or (segments[-1].get("end", 0) if segments else 0)
|
|
|
|
payload = {
|
|
"class_name": class_name,
|
|
"lecture_title": lecture_title,
|
|
"transcript": full_text,
|
|
"timestamped_transcript": build_timestamped(segments),
|
|
"duration_seconds": duration,
|
|
}
|
|
|
|
log.info(f"Sending to n8n: {class_name} / {lecture_title} ({len(full_text)} chars)")
|
|
resp = requests.post(N8N_WEBHOOK, json=payload, timeout=300)
|
|
resp.raise_for_status()
|
|
log.info(f"n8n accepted — HTTP {resp.status_code}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Processing pipeline
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def process_wav(wav_path: Path) -> None:
|
|
# Must be inside a class subfolder, not the root recordings dir
|
|
if wav_path.parent == Path(WATCH_DIR):
|
|
log.warning(f"Skipping {wav_path.name} — drop files into a class subfolder, e.g. /recordings/HIST-2020/")
|
|
return
|
|
|
|
class_name = wav_path.parent.name
|
|
lecture_title = wav_path.stem.replace("-", " ").replace("_", " ")
|
|
|
|
log.info(f"--- Processing: {class_name} / {lecture_title} ---")
|
|
|
|
wait_for_file_stable(wav_path)
|
|
mp3_path = convert_to_mp3(wav_path)
|
|
|
|
result = transcribe(mp3_path)
|
|
send_to_n8n(class_name, lecture_title, result)
|
|
|
|
log.info(f"--- Done: {class_name} / {lecture_title} ---")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Watchdog handler
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class WavHandler(FileSystemEventHandler):
|
|
def on_created(self, event):
|
|
if event.is_directory:
|
|
return
|
|
path = Path(event.src_path)
|
|
if path.suffix.lower() == ".wav":
|
|
log.info(f"Detected new WAV: {path}")
|
|
try:
|
|
process_wav(path)
|
|
except Exception as exc:
|
|
log.error(f"Failed to process {path}: {exc}", exc_info=True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def scan_existing(watch_dir: Path) -> None:
|
|
"""Process any .wav files already present at startup."""
|
|
existing = list(watch_dir.rglob("*.wav")) + list(watch_dir.rglob("*.WAV"))
|
|
if not existing:
|
|
return
|
|
log.info(f"Found {len(existing)} existing WAV(s) at startup — processing...")
|
|
for wav in existing:
|
|
try:
|
|
process_wav(wav)
|
|
except Exception as exc:
|
|
log.error(f"Failed to process {wav}: {exc}", exc_info=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
watch_dir = Path(WATCH_DIR)
|
|
watch_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
log.info(f"Whisper URL : {WHISPER_URL}")
|
|
log.info(f"n8n webhook : {N8N_WEBHOOK}")
|
|
log.info(f"Watching : {watch_dir} (recursive)")
|
|
|
|
# Process any WAVs already sitting in the folder
|
|
scan_existing(watch_dir)
|
|
|
|
observer = Observer()
|
|
observer.schedule(WavHandler(), str(watch_dir), recursive=True)
|
|
observer.start()
|
|
log.info("Watcher started — drop a .wav into a class subfolder to begin")
|
|
|
|
try:
|
|
while True:
|
|
time.sleep(1)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
observer.stop()
|
|
observer.join()
|
|
log.info("Watcher stopped")
|