Local homelab changes
This commit is contained in:
360
rocinante/watcher/watch.py
Normal file
360
rocinante/watcher/watch.py
Normal file
@@ -0,0 +1,360 @@
|
||||
#!/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. Splits long recordings into 10-minute chunks
|
||||
4. Transcribes each chunk via local faster-whisper-server (large-v3, CUDA)
|
||||
5. Stitches results back together with corrected timestamps
|
||||
6. POSTs transcript + metadata to n8n /webhook/class/ingest
|
||||
|
||||
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")
|
||||
CHUNK_MINUTES = int(os.environ.get("CHUNK_MINUTES", "10"))
|
||||
|
||||
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)")
|
||||
|
||||
# Extra check: verify the file handle is released and readable
|
||||
for attempt in range(60):
|
||||
try:
|
||||
with open(path, 'rb') as f:
|
||||
f.read(1024)
|
||||
break
|
||||
except (PermissionError, OSError) as e:
|
||||
log.info(f"File not yet readable ({e}), retrying... ({attempt + 1}/60)")
|
||||
time.sleep(5)
|
||||
else:
|
||||
raise RuntimeError(f"File {path.name} is not readable after 60 attempts")
|
||||
|
||||
|
||||
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 get_duration(mp3_path: Path) -> float:
|
||||
"""Return audio duration in seconds via ffprobe."""
|
||||
result = subprocess.run(
|
||||
[
|
||||
"ffprobe", "-v", "quiet",
|
||||
"-show_entries", "format=duration",
|
||||
"-of", "csv=p=0",
|
||||
str(mp3_path),
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
return float(result.stdout.strip() or "0")
|
||||
|
||||
|
||||
def split_into_chunks(mp3_path: Path, chunk_minutes: int = CHUNK_MINUTES) -> list:
|
||||
"""Split MP3 into chunks. Returns [mp3_path] if short enough."""
|
||||
duration = get_duration(mp3_path)
|
||||
chunk_secs = chunk_minutes * 60
|
||||
|
||||
if duration <= chunk_secs:
|
||||
log.info(f"Recording is {duration/60:.1f}min — no splitting needed")
|
||||
return [mp3_path]
|
||||
|
||||
total_chunks = int(duration / chunk_secs) + 1
|
||||
log.info(f"Recording is {duration/60:.1f}min — splitting into {total_chunks} x {chunk_minutes}min chunks")
|
||||
|
||||
chunks = []
|
||||
start = 0.0
|
||||
i = 0
|
||||
while start < duration:
|
||||
chunk_path = mp3_path.parent / f"{mp3_path.stem}_chunk{i:03d}.mp3"
|
||||
subprocess.run(
|
||||
[
|
||||
"ffmpeg", "-y",
|
||||
"-i", str(mp3_path),
|
||||
"-ss", str(start),
|
||||
"-t", str(chunk_secs),
|
||||
"-codec:a", "copy",
|
||||
str(chunk_path),
|
||||
],
|
||||
capture_output=True,
|
||||
)
|
||||
chunks.append((chunk_path, start))
|
||||
start += chunk_secs
|
||||
i += 1
|
||||
|
||||
return chunks
|
||||
|
||||
|
||||
def transcribe_chunk(mp3_path: Path) -> dict:
|
||||
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=3600,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return resp.json()
|
||||
|
||||
|
||||
def transcribe(mp3_path: Path) -> dict:
|
||||
"""Transcribe an MP3, splitting into chunks if needed."""
|
||||
chunks = split_into_chunks(mp3_path)
|
||||
|
||||
# Short recording — single chunk, no splitting
|
||||
if len(chunks) == 1 and isinstance(chunks[0], Path):
|
||||
log.info(f"Transcribing {mp3_path.name} with {WHISPER_MODEL}...")
|
||||
result = transcribe_chunk(mp3_path)
|
||||
result["segments"] = deduplicate_hallucinations(result.get("segments", []))
|
||||
log.info("Transcription complete")
|
||||
return result
|
||||
|
||||
# Long recording — transcribe each chunk and stitch
|
||||
all_segments = []
|
||||
all_text = []
|
||||
|
||||
for i, (chunk_path, time_offset) in enumerate(chunks):
|
||||
log.info(f"Transcribing chunk {i+1}/{len(chunks)}: {chunk_path.name} (offset {time_offset/60:.1f}min)")
|
||||
try:
|
||||
result = transcribe_chunk(chunk_path)
|
||||
segments = result.get("segments", [])
|
||||
for s in segments:
|
||||
s["start"] = s.get("start", 0) + time_offset
|
||||
s["end"] = s.get("end", 0) + time_offset
|
||||
all_segments.extend(segments)
|
||||
all_text.append(result.get("text", "").strip())
|
||||
except Exception as e:
|
||||
log.error(f"Chunk {i+1} failed: {e}")
|
||||
finally:
|
||||
chunk_path.unlink(missing_ok=True)
|
||||
|
||||
all_segments = deduplicate_hallucinations(all_segments)
|
||||
total_duration = get_duration(mp3_path)
|
||||
log.info(f"Transcription complete — {len(chunks)} chunks stitched")
|
||||
return {
|
||||
"text": " ".join(all_text),
|
||||
"segments": all_segments,
|
||||
"duration": total_duration,
|
||||
}
|
||||
|
||||
|
||||
def is_silence(text: str) -> bool:
|
||||
"""Return True if the segment is just silence/noise markers."""
|
||||
stripped = text.strip()
|
||||
return not stripped or all(c in '.…♪ \t' for c in stripped)
|
||||
|
||||
|
||||
def deduplicate_hallucinations(segments: list, max_repeats: int = 3) -> list:
|
||||
"""
|
||||
Remove Whisper phrase-loop hallucinations.
|
||||
|
||||
When Whisper gets stuck it emits the same segment text hundreds of times
|
||||
in a row (e.g. "I'm also a software engineer." × 200). This function
|
||||
collapses any run longer than `max_repeats` down to at most `max_repeats`
|
||||
occurrences and logs a warning so the truncation is visible in the logs.
|
||||
"""
|
||||
if not segments:
|
||||
return segments
|
||||
|
||||
result = []
|
||||
i = 0
|
||||
while i < len(segments):
|
||||
text = segments[i].get("text", "").strip().lower()
|
||||
j = i + 1
|
||||
while j < len(segments) and segments[j].get("text", "").strip().lower() == text:
|
||||
j += 1
|
||||
count = j - i
|
||||
if count > max_repeats:
|
||||
log.warning(
|
||||
f"Hallucination detected: '{segments[i].get('text', '').strip()[:60]}'"
|
||||
f" × {count} — keeping {max_repeats}, dropping {count - max_repeats}"
|
||||
)
|
||||
result.extend(segments[i : i + max_repeats])
|
||||
else:
|
||||
result.extend(segments[i:j])
|
||||
i = j
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def build_timestamped(segments: list) -> str:
|
||||
lines = []
|
||||
for s in segments:
|
||||
text = s.get('text', '').strip()
|
||||
if is_silence(text):
|
||||
continue
|
||||
mins = int(s.get("start", 0) // 60)
|
||||
secs = int(s.get("start", 0) % 60)
|
||||
lines.append(f"[{mins}:{secs:02d}] {text}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def send_to_n8n(class_name: str, lecture_title: str, result: dict) -> None:
|
||||
segments = result.get("segments", [])
|
||||
# Filter silence from full text
|
||||
meaningful = [s.get("text", "") for s in segments if not is_silence(s.get("text", ""))]
|
||||
full_text = " ".join(meaningful).strip() or 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")
|
||||
Reference in New Issue
Block a user