commit 0da74d39a060cdef6ee1510a1e89835c27a18f8c Author: Jarian Date: Tue Jul 7 04:47:18 2026 +0000 init: media-transcriber pipeline - whisper.cpp base.en transcription on RTX 5060 Ti (GPU 1) - 25,439 videos indexed from /mnt/mediaserver/ - pipeline: ffmpeg extract -> whisper-server infer -> save SRT - thermal cycle: 45min work / 15min rest - fixes: curl quotes for spaces, --convert + WorkingDirectory=/tmp diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2272d30 --- /dev/null +++ b/.gitignore @@ -0,0 +1,26 @@ +# Virtual env +venv/ +.venv/ + +# Index (25k videos, large) +index.json +index.json.tmp + +# Logs +logs/ + +# Transcripts +transcripts/ + +# Temp files +*.wav +tmp/ + +# Python +__pycache__/ +*.pyc +*.pyo + +# IDE +.vscode/ +.idea/ diff --git a/mt.py b/mt.py new file mode 100755 index 0000000..0398f29 --- /dev/null +++ b/mt.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +import sys +import os + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "src")) + +def main(): + if len(sys.argv) < 2: + print("Usage: mt.py ") + sys.exit(1) + + cmd = sys.argv[1] + if cmd == "scan": + import json + from indexer import scan + skip_dur = "--duration" not in sys.argv + idx = scan(skip_duration=skip_dur) + print(json.dumps({ + "total": idx["total_videos"], + "transcribed": idx["transcribed"], + "pending": idx["total_videos"] - idx["transcribed"], + }, indent=2)) + elif cmd in ("transcribe", "continuous"): + from worker import run_once, run_continuous + run_once() if cmd == "transcribe" else run_continuous() + elif cmd == "status": + import json + with open("/home/jarian/projects/media-transcriber/index.json") as f: + idx = json.load(f) + print(f"Total: {idx['total_videos']}") + print(f"Done: {idx['transcribed']}") + print(f"Pending: {idx['total_videos'] - idx['transcribed']}") + print(f"Updated: {idx['last_updated']}") + else: + print(f"Unknown command: {cmd}") + sys.exit(1) + +if __name__ == "__main__": + main() diff --git a/setup.sh b/setup.sh new file mode 100755 index 0000000..a45467d --- /dev/null +++ b/setup.sh @@ -0,0 +1,61 @@ +#!/usr/bin/env bash +set -euo pipefail + +PROJECT_DIR="/home/jarian/projects/media-transcriber" +WHISPER_CPP="/home/jarian/whisper.cpp" + +echo "=== Media Transcriber Setup ===" + +# Create venv +cd "$PROJECT_DIR" +python3 -m venv venv +source venv/bin/activate + +# Install deps +pip install --upgrade pip + +# Ensure ffmpeg available +if ! command -v ffmpeg &>/dev/null; then + echo "ERROR: ffmpeg not found. Install with: sudo apt install ffmpeg" + exit 1 +fi +if ! command -v ffprobe &>/dev/null; then + echo "ERROR: ffprobe not found. Install with: sudo apt install ffmpeg" + exit 1 +fi + +# Check whisper-server binary +if [ ! -f "$WHISPER_CPP/build/bin/whisper-server" ]; then + echo "ERROR: whisper-server not built. Run:" + echo " cd $WHISPER_CPP && cmake -B build -DWHISPER_CUDA=ON && cmake --build build" + exit 1 +fi + +# Download small model if missing +MODEL_PATH="$WHISPER_CPP/models/ggml-small.bin" +if [ ! -f "$MODEL_PATH" ]; then + echo "Downloading whisper small model..." + bash "$WHISPER_CPP/models/download-ggml-model.sh" small +fi + +# Initial scan +echo "Running initial video scan..." +cd "$PROJECT_DIR" +source venv/bin/activate +python src/worker.py scan + +echo "" +echo "=== Setup complete ===" +echo "" +echo "To start whisper-server on GPU 1:" +echo " CUDA_VISIBLE_DEVICES=1 $WHISPER_CPP/build/bin/whisper-server \\" +echo " -m $MODEL_PATH -ngl 99 -ps 16384" +echo "" +echo "Or enable systemd service:" +echo " sudo cp systemd/whisper-server.service /etc/systemd/system/" +echo " sudo systemctl enable --start whisper-server" +echo "" +echo "To enable nightly transcription:" +echo " sudo cp systemd/media-transcriber.timer /etc/systemd/system/" +echo " sudo cp systemd/media-transcriber.service /etc/systemd/system/" +echo " sudo systemctl enable --now media-transcriber.timer" diff --git a/src/indexer.py b/src/indexer.py new file mode 100644 index 0000000..e826c91 --- /dev/null +++ b/src/indexer.py @@ -0,0 +1,95 @@ +import os +import json +import hashlib +import subprocess +from datetime import datetime, timezone + +VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".flv", ".wmv", ".m4v"} +BACKUP_KEYWORDS = {"backup", "backups", "#recycle"} + +MEDIA_ROOT = os.environ.get("MEDIA_ROOT", "/mnt/mediaserver") +SCAN_DIRS = os.environ.get("SCAN_DIRS", "Youtube,Movies,TV").split(",") +INDEX_PATH = os.environ.get("INDEX_PATH", "/home/jarian/projects/media-transcriber/index.json") +TRANSCRIPTS_DIR = os.environ.get("TRANSCRIPTS_DIR", "/home/jarian/projects/media-transcriber/transcripts") + + +def is_video(path): + return os.path.splitext(path)[1].lower() in VIDEO_EXTENSIONS + + +def is_backup(path): + parts = path.replace(MEDIA_ROOT, "").split("/") + return any(k.lower() in {p.lower() for p in parts} for k in BACKUP_KEYWORDS) + + +def video_hash(filepath): + stat = os.stat(filepath) + raw = f"{stat.st_size}:{stat.st_mtime_ns}:{filepath}" + return hashlib.sha256(raw.encode()).hexdigest()[:16] + + +def get_duration(filepath): + try: + result = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", filepath], + capture_output=True, text=True, timeout=10 + ) + return round(float(result.stdout.strip()), 2) + except Exception: + return None + + +def transcript_exists(rel_path): + full_path = os.path.join(MEDIA_ROOT, rel_path) + parent = os.path.dirname(full_path) + trans_dir = os.path.join(parent, "_transcriptions") + base = os.path.splitext(os.path.basename(full_path))[0] + for ext in (".srt", ".vtt", ".txt", ".json"): + candidate = os.path.join(trans_dir, f"{base}_transcription{ext}") + if os.path.exists(candidate): + return True, candidate + return False, None + + +def scan(skip_duration=False): + videos = {} + now = datetime.now(timezone.utc).isoformat() + for scan_dir in SCAN_DIRS: + root = os.path.join(MEDIA_ROOT, scan_dir) + if not os.path.isdir(root): + continue + for dirpath, _, filenames in os.walk(root): + for fname in filenames: + full = os.path.join(dirpath, fname) + if not is_video(full) or is_backup(full): + continue + rel = os.path.relpath(full, MEDIA_ROOT) + has_transcript, transcript_file = transcript_exists(rel) + vid_hash = video_hash(full) + stat = os.stat(full) + videos[rel] = { + "path": full, + "hash": vid_hash, + "duration_seconds": None if skip_duration else get_duration(full), + "size_bytes": stat.st_size, + "transcribed": has_transcript, + "transcript_file": transcript_file, + "last_scanned": now, + } + + transcribed_count = sum(1 for v in videos.values() if v["transcribed"]) + index = { + "version": "1", + "last_updated": now, + "total_videos": len(videos), + "transcribed": transcribed_count, + "videos": videos, + } + + tmp = INDEX_PATH + ".tmp" + with open(tmp, "w") as f: + json.dump(index, f, indent=2) + os.replace(tmp, INDEX_PATH) + + return index diff --git a/src/worker.py b/src/worker.py new file mode 100644 index 0000000..eaf0937 --- /dev/null +++ b/src/worker.py @@ -0,0 +1,302 @@ +import os +import sys +import json +import time +import threading +import subprocess +from queue import Queue, Empty +from datetime import datetime, timezone + +sys.path.insert(0, os.path.dirname(__file__)) +from indexer import INDEX_PATH, scan, video_hash + +WHISPER_SERVER = os.environ.get("WHISPER_SERVER_URL", "http://127.0.0.1:8888") +WHISPER_MODEL = os.environ.get("WHISPER_MODEL", "base.en") +WORK_SECONDS = int(os.environ.get("WORK_SECONDS", "2700")) # 45 min active +REST_SECONDS = int(os.environ.get("REST_SECONDS", "900")) # 15 min thermal break +BATCH_LOG = os.environ.get("BATCH_LOG", "/home/jarian/projects/media-transcriber/logs/batch.log") +SPEED_FACTOR = 2.5 # base.en ~2-3x realtime, safety margin + +stop_event = threading.Event() + + +def load_index(): + with open(INDEX_PATH) as f: + return json.load(f) + + +def save_index(index): + tmp = INDEX_PATH + ".tmp" + with open(tmp, "w") as f: + json.dump(index, f, indent=2) + os.replace(tmp, INDEX_PATH) + + +def log(msg): + ts = datetime.now(timezone.utc).isoformat() + line = f"[{ts}] {msg}" + print(line, flush=True) + os.makedirs(os.path.dirname(BATCH_LOG), exist_ok=True) + with open(BATCH_LOG, "a") as f: + f.write(line + "\n") + + +# --- Pipeline stages --- + +class ExtractWorker: + def __init__(self, video_queue, audio_queue): + self.video_queue = video_queue + self.audio_queue = audio_queue + + def run(self): + while not stop_event.is_set(): + try: + rel, v, index = self.video_queue.get(timeout=1) + except Empty: + continue + + audio_path = f"/tmp/mt_{rel.replace('/', '_')}.wav" + try: + if video_hash(v["path"]) != v["hash"]: + log(f" SKIP {rel} - file changed") + continue + log(f" [CPU] Extract: {rel}") + subprocess.run( + ["ffmpeg", "-y", "-i", v["path"], "-vn", "-acodec", "pcm_s16le", + "-ar", "16000", "-ac", "1", audio_path], + check=True, timeout=300, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL + ) + self.audio_queue.put((rel, v, audio_path, index)) + except subprocess.TimeoutExpired: + log(f" [CPU] Extract timeout: {rel}") + except Exception as e: + log(f" [CPU] Extract FAIL {rel}: {e}") + + +class InferWorker: + def __init__(self, audio_queue, result_queue): + self.audio_queue = audio_queue + self.result_queue = result_queue + + def run(self): + while not stop_event.is_set(): + try: + item = self.audio_queue.get(timeout=1) + except Empty: + continue + if item is None: + continue + + rel, v, audio_path, index = item + dur = v.get("duration_seconds") or 120 + vid_timeout = max(int(dur * SPEED_FACTOR) + 60, 120) + + try: + log(f" [GPU] Transcribe: {rel} (est {dur * SPEED_FACTOR:.0f}s)") + resp = subprocess.run( + ["curl", "-s", "--max-time", str(vid_timeout), + f"{WHISPER_SERVER}/inference", + "-F", f'file=@"{audio_path}"', + "-F", "response_format=srt"], + capture_output=True, text=True, timeout=vid_timeout + 30 + ) + if resp.returncode != 0: + raise RuntimeError(f"whisper-server error: {resp.stderr}") + if not resp.stdout.strip(): + raise RuntimeError("empty result") + self.result_queue.put((rel, resp.stdout, index)) + except subprocess.TimeoutExpired: + log(f" [GPU] Timeout: {rel}") + except Exception as e: + log(f" [GPU] FAIL {rel}: {e}") + finally: + if os.path.exists(audio_path): + os.remove(audio_path) + + +class SaveWorker: + def __init__(self, result_queue, stats): + self.result_queue = result_queue + self.stats = stats + + def run(self): + while not stop_event.is_set(): + try: + item = self.result_queue.get(timeout=1) + except Empty: + continue + if item is None: + continue + + rel, srt, index = item + try: + video_path = index["videos"][rel]["path"] + parent = os.path.dirname(video_path) + trans_dir = os.path.join(parent, "_transcriptions") + base = os.path.splitext(os.path.basename(video_path))[0] + out = os.path.join(trans_dir, f"{base}_transcription.srt") + os.makedirs(trans_dir, exist_ok=True) + with open(out, "w") as f: + f.write(srt) + + index["videos"][rel]["transcribed"] = True + index["videos"][rel]["transcript_file"] = out + index["videos"][rel]["transcribed_at"] = datetime.now(timezone.utc).isoformat() + index["transcribed"] += 1 + save_index(index) + self.stats["done"] += 1 + log(f" [SAVE] OK {rel}") + except Exception as e: + log(f" [SAVE] FAIL {rel}: {e}") + self.stats["failed"] += 1 + + +def thermal_sleep(seconds, label): + """Sleep with progress logging. Respects stop_event.""" + for i in range(seconds, 0, -30): + if stop_event.is_set(): + break + log(f" [{label}] Sleeping... {i}s remaining") + stop_event.wait(30) + + +def run_continuous(): + """Run 24/7 with thermal breaks: WORK_SECONDS on, REST_SECONDS off.""" + log(f"Continuous mode: model={WHISPER_MODEL}, server={WHISPER_SERVER}") + log(f"Thermal cycle: {WORK_SECONDS}s work / {REST_SECONDS}s rest") + log(f"Speed factor: {SPEED_FACTOR}x") + + cycle = 0 + while not stop_event.is_set(): + cycle += 1 + log(f"=== Cycle {cycle}: WORK phase ({WORK_SECONDS}s) ===") + + # Re-scan index each cycle to pick up new files + index = load_index() + pending = [(rel, v) for rel, v in index["videos"].items() if not v["transcribed"]] + + if not pending: + log("All videos transcribed! Exiting.") + return + + log(f"Pending: {len(pending)} videos") + + # Queues + video_queue = Queue(maxsize=2) + audio_queue = Queue(maxsize=2) + result_queue = Queue(maxsize=4) + stats = {"done": 0, "failed": 0} + + # Start pipeline FIRST (consumers need to run before we enqueue) + threads = [] + threads.append(threading.Thread(target=ExtractWorker(video_queue, audio_queue).run, daemon=True)) + threads.append(threading.Thread(target=InferWorker(audio_queue, result_queue).run, daemon=True)) + threads.append(threading.Thread(target=SaveWorker(result_queue, stats).run, daemon=True)) + for t in threads: + t.start() + + # Enqueue eligible videos + enqueued = 0 + for rel, v in pending: + dur = v.get("duration_seconds") or 120 + est = dur * SPEED_FACTOR + if est <= WORK_SECONDS: + video_queue.put((rel, v, index)) + enqueued += 1 + log(f"Enqueued for this cycle: {enqueued}") + + # Monitor work phase + phase_start = time.time() + last_done = 0 + while True: + time.sleep(5) + if stop_event.is_set(): + break + + elapsed = time.time() - phase_start + if elapsed >= WORK_SECONDS: + log(f"Work phase done ({elapsed:.0f}s). {stats['done']} transcribed.") + break + + if stats["done"] > last_done: + last_done = stats["done"] + + stop_event.set() + for t in threads: + t.join(timeout=3) + + total_before = index["transcribed"] - stats["done"] - stats["failed"] + log(f"Cycle {cycle} result: {stats['done']} done, {stats['failed']} failed") + + # Thermal break + remaining = load_index()["total_videos"] - load_index()["transcribed"] + if remaining <= 0: + log("All videos transcribed! Done.") + return + + log(f"=== Cycle {cycle}: REST phase ({REST_SECONDS}s) - {remaining} remaining ===") + stop_event.clear() + thermal_sleep(REST_SECONDS, "REST") + stop_event.set() + + +def run_once(): + """Single-shot batch (legacy mode).""" + log(f"Single batch: model={WHISPER_MODEL}, server={WHISPER_SERVER}") + + index = load_index() + pending = [(rel, v) for rel, v in index["videos"].items() if not v["transcribed"]] + if not pending: + log("No pending videos.") + return + + log(f"Found {len(pending)} untranscribed videos") + + video_queue = Queue(maxsize=2) + audio_queue = Queue(maxsize=2) + result_queue = Queue(maxsize=4) + stats = {"done": 0, "failed": 0} + + for rel, v in pending: + video_queue.put((rel, v, index)) + + threads = [] + threads.append(threading.Thread(target=ExtractWorker(video_queue, audio_queue).run, daemon=True)) + threads.append(threading.Thread(target=InferWorker(audio_queue, result_queue).run, daemon=True)) + threads.append(threading.Thread(target=SaveWorker(result_queue, stats).run, daemon=True)) + for t in threads: + t.start() + + start = time.time() + while not video_queue.empty() or not audio_queue.empty() or not result_queue.empty(): + time.sleep(5) + elapsed = time.time() - start + if elapsed >= WORK_SECONDS + REST_SECONDS: + log(f"Timeout ({elapsed:.0f}s). {stats['done']} done.") + stop_event.set() + break + stop_event.set() + for t in threads: + t.join(timeout=3) + + elapsed = time.time() - start + log(f"Done: {stats['done']} transcribed, {stats['failed']} failed, {elapsed:.0f}s") + + +if __name__ == "__main__": + cmd = sys.argv[1] if len(sys.argv) > 1 else "continuous" + if cmd == "scan": + skip_dur = "--duration" not in sys.argv + idx = scan(skip_duration=skip_dur) + print(json.dumps({ + "total": idx["total_videos"], + "transcribed": idx["transcribed"], + "pending": idx["total_videos"] - idx["transcribed"], + }, indent=2)) + elif cmd == "continuous": + run_continuous() + elif cmd == "transcribe": + run_once() + else: + print(f"Usage: {sys.argv[0]} [scan|transcribe|continuous]") + sys.exit(1) diff --git a/systemd/media-transcriber.service b/systemd/media-transcriber.service new file mode 100644 index 0000000..92d4f9f --- /dev/null +++ b/systemd/media-transcriber.service @@ -0,0 +1,19 @@ +[Unit] +Description=Media Transcriber Continuous Worker +Requires=whisper-server.service +After=whisper-server.service + +[Service] +Type=simple +User=jarian +WorkingDirectory=/home/jarian/projects/media-transcriber +ExecStart=/home/jarian/projects/media-transcriber/venv/bin/python \ + /home/jarian/projects/media-transcriber/src/worker.py continuous +Restart=on-failure +RestartSec=60 +Environment=WORK_SECONDS=2700 +Environment=REST_SECONDS=900 +Environment=WHISPER_MODEL=base.en + +[Install] +WantedBy=multi-user.target diff --git a/systemd/whisper-server.service b/systemd/whisper-server.service new file mode 100644 index 0000000..738074e --- /dev/null +++ b/systemd/whisper-server.service @@ -0,0 +1,18 @@ +[Unit] +Description=Whisper.cpp Transcription Server +After=network.target + +[Service] +Type=simple +User=jarian +ExecStart=/home/jarian/whisper.cpp/build/bin/whisper-server \ + -m /home/jarian/whisper.cpp/models/ggml-base.en.bin \ + --host 127.0.0.1 \ + --port 8888 \ + --convert +Environment=CUDA_VISIBLE_DEVICES=1 +Restart=on-failure +RestartSec=10 + +[Install] +WantedBy=multi-user.target