#!/usr/bin/env python3
"""
Transcription Review Dashboard
================================
Serves a web UI for reviewing audio segments alongside their transcriptions
and validation scores. Reads from consistency_test/ or transcriptions/ JSON.
Usage:
cd /home/ubuntu/maya3_transcribe
source venv/bin/activate
python bin/dashboard.py [--port 8765] [--data consistency_test/v2_comparison.json]
Then open http://localhost:8765 in your browser.
"""
import os
import sys
import json
import argparse
import mimetypes
from pathlib import Path
from http.server import HTTPServer, SimpleHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
PROJECT_ROOT = Path(__file__).parent.parent
DEFAULT_DATA = PROJECT_ROOT / "consistency_test" / "v2_comparison.json"
DEFAULT_SEG_DIR = "/tmp/maya3_transcribe/pF_BQpHaIdU/extracted/pF_BQpHaIdU/segments"
def load_dashboard_data(data_path: str, seg_dir: str) -> dict:
"""Load transcription data from JSON and resolve audio paths."""
with open(data_path, "r", encoding="utf-8") as f:
raw = json.load(f)
# Handle both consistency-test format and pipeline output format
segments = []
if "runs" in raw:
# Consistency test format: {config, runs: [{seg: {...}}, ...]}
runs = raw["runs"]
all_segs = set()
for run in runs:
all_segs.update(run.keys())
for seg_name in sorted(all_segs):
run_data = []
for ri, run in enumerate(runs):
if seg_name in run:
run_data.append(run[seg_name])
if not run_data:
continue
r0 = run_data[0]
audio_path = os.path.join(seg_dir, seg_name)
consistent = all(
rd.get("transcription") == r0.get("transcription")
for rd in run_data
)
segments.append({
"id": seg_name,
"audio_exists": os.path.exists(audio_path),
"audio_url": f"/audio/{seg_name}",
"transcription": r0.get("transcription", ""),
"romanized": r0.get("romanized", ""),
"detected_language": r0.get("detected_language", ""),
"native_ctc": r0.get("native_ctc", 0),
"roman_mms": r0.get("roman_mms", 0),
"combined": r0.get("combined", 0),
"status": r0.get("status", "unknown"),
"consistent": consistent,
"num_runs": len(run_data),
"runs": run_data,
})
elif "results" in raw:
# Pipeline output format: {results: [{segment_id, transcription: {...}, ...}]}
for r in raw["results"]:
seg_name = r.get("segment_id", "")
audio_path = os.path.join(seg_dir, seg_name)
tx = r.get("transcription", {})
segments.append({
"id": seg_name,
"audio_exists": os.path.exists(audio_path),
"audio_url": f"/audio/{seg_name}",
"transcription": tx.get("transcription", "") if isinstance(tx, dict) else str(tx),
"romanized": tx.get("romanized", "") if isinstance(tx, dict) else "",
"tagged": tx.get("tagged", "") if isinstance(tx, dict) else "",
"detected_language": tx.get("detected_language", "") if isinstance(tx, dict) else "",
"native_ctc": 0,
"roman_mms": 0,
"combined": r.get("validation_score", 0) or 0,
"status": r.get("validation_status", "unknown") or "unknown",
"consistent": True,
"num_runs": 1,
"runs": [],
})
return {"segments": segments, "source": str(data_path)}
class DashboardHandler(SimpleHTTPRequestHandler):
"""HTTP handler serving the dashboard, audio files, and API."""
data = {}
seg_dir = ""
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path
if path == "/" or path == "/index.html":
self._serve_html()
elif path == "/api/data":
self._serve_json(self.data)
elif path.startswith("/audio/"):
self._serve_audio(path[7:])
else:
self.send_error(404)
def _serve_html(self):
html = DASHBOARD_HTML
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(html.encode("utf-8"))
def _serve_json(self, data):
self.send_response(200)
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, filename):
audio_path = os.path.join(self.seg_dir, filename)
if not os.path.exists(audio_path):
self.send_error(404, f"Audio not found: {filename}")
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):
if "/api/" not in str(args[0]) and "/audio/" not in str(args[0]):
super().log_message(format, *args)
DASHBOARD_HTML = r"""
Transcription Review Dashboard
"""
def main():
parser = argparse.ArgumentParser(description="Transcription Review Dashboard")
parser.add_argument("--port", "-p", type=int, default=8765)
parser.add_argument("--data", "-d", default=str(DEFAULT_DATA),
help="Path to transcription JSON")
parser.add_argument("--segments", "-s", default=DEFAULT_SEG_DIR,
help="Path to audio segments directory")
args = parser.parse_args()
if not os.path.exists(args.data):
print(f"Data file not found: {args.data}")
print("Run a consistency test or pipeline first to generate data.")
sys.exit(1)
print(f"Loading data from {args.data}...")
data = load_dashboard_data(args.data, args.segments)
print(f"Loaded {len(data['segments'])} segments")
DashboardHandler.data = data
DashboardHandler.seg_dir = args.segments
server = HTTPServer(("0.0.0.0", args.port), DashboardHandler)
print(f"\nDashboard running at 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()