#!/usr/bin/env python3
"""
Transcription Review Dashboard v2
===================================
Features:
- Audio player + waveform visualizer per segment (wavesurfer.js)
- Segment numbering (SEG_0000, SEG_0001, ...)
- Filter by status (accept/review/retry/reject)
- Detailed view: large waveform + start/end trim sliders + timeline
- Re-transcribe trimmed audio with history (old vs new diff)
- Keyboard navigation (J/K/Space)
Usage:
cd /home/ubuntu/maya3_transcribe && source venv/bin/activate
python bin/dashboard.py [--port 8765] [--data consistency_test/full_run.json]
"""
import io
import os
import sys
import json
import argparse
import mimetypes
import tempfile
from pathlib import Path
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
PROJECT_ROOT = Path(__file__).parent.parent
DEFAULT_SEG_DIR = "/tmp/maya3_transcribe/pF_BQpHaIdU/extracted/pF_BQpHaIdU/segments"
# Lazy-load heavy imports only when re-transcription is requested
_transcriber = None
_config = None
def _get_transcriber():
global _transcriber, _config
if _transcriber is None:
sys.path.insert(0, str(PROJECT_ROOT))
from src.backend.gemini_transcriber import GeminiTranscriber, TranscriptionConfig
from src.validators import validate_transcription
_transcriber = GeminiTranscriber()
_config = TranscriptionConfig(
model="gemini-3-flash-preview", thinking_level="low",
temperature=0.0, language="Telugu"
)
return _transcriber, _config
def load_data(data_path: str, seg_dir: str) -> dict:
"""Load transcription data from JSON."""
with open(data_path, "r", encoding="utf-8") as f:
raw = json.load(f)
segments = []
if "segments" in raw and isinstance(raw["segments"], list):
# full_run.py format: {meta, segments: [{seg_num, ...}]}
for i, s in enumerate(raw["segments"]):
seg_name = s.get("original_file", s.get("id", f"seg_{i}"))
audio_file = s.get("audio_file", seg_name)
# Resolve audio path: try polished dir first, then original
audio_path = s.get("audio_path", "")
if not audio_path or not os.path.exists(audio_path):
audio_path = os.path.join(seg_dir, audio_file)
if not os.path.exists(audio_path):
audio_path = os.path.join(seg_dir, seg_name)
segments.append({
"idx": i,
"seg_num": s.get("seg_num", f"SEG_{i:04d}"),
"id": seg_name,
"audio_file": audio_file,
"audio_path": audio_path,
"audio_exists": os.path.exists(audio_path),
"audio_url": f"/audio?path={audio_path}",
"duration_sec": s.get("duration_sec", 0),
"chunk_index": s.get("chunk_index", 0),
"total_chunks": s.get("total_chunks", 1),
"transcription": s.get("transcription", ""),
"tagged": s.get("tagged", ""),
"romanized": s.get("romanized", ""),
"uroman": s.get("uroman", ""),
"detected_language": s.get("detected_language", ""),
"speaker": s.get("speaker", {}),
"native_ctc": s.get("native_ctc", 0),
"roman_mms": s.get("roman_mms", 0),
"combined": s.get("combined", 0),
"status": s.get("status", "unknown"),
"error": s.get("error", ""),
"history": [],
})
elif "runs" in raw:
# consistency test format
runs = raw["runs"]
all_segs = set()
for run in runs:
all_segs.update(run.keys())
for i, seg_name in enumerate(sorted(all_segs)):
r0 = runs[0].get(seg_name, {})
audio_path = os.path.join(seg_dir, seg_name)
segments.append({
"idx": i,
"seg_num": f"SEG_{i:04d}",
"id": seg_name,
"audio_file": seg_name,
"audio_path": audio_path,
"audio_exists": os.path.exists(audio_path),
"audio_url": f"/audio?path={audio_path}",
"duration_sec": 0,
"chunk_index": 0, "total_chunks": 1,
"transcription": r0.get("transcription", ""),
"tagged": "", "romanized": r0.get("romanized", ""),
"uroman": "", "detected_language": r0.get("detected_language", ""),
"speaker": {},
"native_ctc": r0.get("native_ctc", 0),
"roman_mms": r0.get("roman_mms", 0),
"combined": r0.get("combined", 0),
"status": r0.get("status", "unknown"),
"error": "",
"history": [],
"runs": [runs[ri].get(seg_name, {}) for ri in range(len(runs))],
})
return {
"segments": segments,
"meta": raw.get("meta", {}),
"source": str(data_path),
}
def retranscribe_trimmed(audio_path: str, start_sec: float, end_sec: float,
language: str = "Telugu") -> dict:
"""Trim audio to [start_sec, end_sec] and re-transcribe."""
import soundfile as sf
import numpy as np
audio, sr = sf.read(audio_path)
if len(audio.shape) > 1:
audio = audio.mean(axis=1)
ss = int(start_sec * sr)
es = int(end_sec * sr)
trimmed = audio[max(0, ss):min(len(audio), es)]
# Write to temp file
tmp = tempfile.NamedTemporaryFile(suffix=".flac", delete=False)
sf.write(tmp.name, trimmed, sr)
tmp.close()
try:
transcriber, config = _get_transcriber()
raw = transcriber.transcribe_audio(tmp.name, config)
if raw.get("error"):
return {"error": raw["error"]}
native = raw.get("transcription", "")
lang_code_map = {"Telugu": "te", "Hindi": "hi", "Tamil": "ta", "English": "en"}
lang_code = lang_code_map.get(language, "te")
from src.validators import validate_transcription
val = validate_transcription(tmp.name, native, language=lang_code,
duration_sec=end_sec - start_sec)
return {
"transcription": native,
"tagged": raw.get("tagged", ""),
"romanized": raw.get("romanized", ""),
"detected_language": raw.get("detected_language", ""),
"native_ctc": val.native_ctc_score,
"roman_mms": val.roman_mms_score,
"combined": val.combined_score,
"status": val.status,
"trim_start": start_sec,
"trim_end": end_sec,
"trim_duration": round(end_sec - start_sec, 2),
}
finally:
os.unlink(tmp.name)
class DashboardHandler(BaseHTTPRequestHandler):
data = {}
seg_dir = ""
data_path = ""
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path
params = parse_qs(parsed.query)
if path == "/" or path == "/index.html":
self._serve_html()
elif path == "/api/data":
self._json_response(self.data)
elif path == "/audio":
audio_path = params.get("path", [""])[0]
self._serve_audio(audio_path)
elif path == "/api/reload":
self.__class__.data = load_data(self.data_path, self.seg_dir)
self._json_response({"ok": True, "segments": len(self.data["segments"])})
else:
self.send_error(404)
def do_POST(self):
parsed = urlparse(self.path)
if parsed.path == "/api/retranscribe":
length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(length))
audio_path = body.get("audio_path", "")
start = body.get("start_sec", 0)
end = body.get("end_sec", 0)
lang = body.get("language", "Telugu")
if not audio_path or not os.path.exists(audio_path):
self._json_response({"error": "Audio not found"}, status=404)
return
result = retranscribe_trimmed(audio_path, start, end, lang)
# Store in history
for seg in self.data.get("segments", []):
if seg.get("audio_path") == audio_path:
seg.setdefault("history", []).append(result)
break
self._json_response(result)
else:
self.send_error(404)
def _serve_html(self):
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(DASHBOARD_HTML.encode("utf-8"))
def _json_response(self, data, status=200):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))
def _serve_audio(self, audio_path):
if not audio_path or not os.path.exists(audio_path):
self.send_error(404, "Audio not found")
return
mime = mimetypes.guess_type(audio_path)[0] or "audio/flac"
size = os.path.getsize(audio_path)
self.send_response(200)
self.send_header("Content-Type", mime)
self.send_header("Content-Length", str(size))
self.send_header("Accept-Ranges", "bytes")
self.end_headers()
with open(audio_path, "rb") as f:
self.wfile.write(f.read())
def log_message(self, format, *args):
msg = str(args[0]) if args else ""
if "/api/" not in msg and "/audio" not in msg:
super().log_message(format, *args)
DASHBOARD_HTML = r"""
Transcription Review Dashboard
"""
def main():
parser = argparse.ArgumentParser(description="Transcription Review Dashboard v2")
parser.add_argument("--port", "-p", type=int, default=8765)
parser.add_argument("--data", "-d", default=None,
help="Path to JSON data (auto-detects full_run.json or v2_comparison.json)")
parser.add_argument("--segments", "-s", default=DEFAULT_SEG_DIR)
args = parser.parse_args()
# Auto-detect data file
if args.data is None:
candidates = [
PROJECT_ROOT / "consistency_test" / "full_run.json",
PROJECT_ROOT / "consistency_test" / "v2_comparison.json",
]
for c in candidates:
if c.exists():
args.data = str(c)
break
if args.data is None:
print("No data file found. Run full_run.py or consistency test first.")
sys.exit(1)
print(f"Loading data from {args.data}...")
data = load_data(args.data, args.segments)
print(f"Loaded {len(data['segments'])} segments")
DashboardHandler.data = data
DashboardHandler.seg_dir = args.segments
DashboardHandler.data_path = args.data
server = HTTPServer(("0.0.0.0", args.port), DashboardHandler)
print(f"\nDashboard: http://localhost:{args.port}")
print("Press Ctrl+C to stop\n")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nShutting down...")
server.server_close()
if __name__ == "__main__":
main()