jarianc/media-transcriber
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init: media-transcriber pipeline

Browse Source 
- 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
This commit is contained in:
Jarian 2026-07-07 04:47:18 +00:00
commit 0da74d39a0
7 changed files with 560 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
.gitignore vendored Normal file
View File

@ -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/

39
mt.py Executable file
View File

@ -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 <scan|transcribe|continuous|status>")
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()

61
setup.sh Executable file
View File

@ -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"

95
src/indexer.py Normal file
View File

@ -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

302
src/worker.py Normal file
View File

@ -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)

19
systemd/media-transcriber.service Normal file
View File

@ -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

18
systemd/whisper-server.service Normal file
View File

@ -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