"""
maven dataset QA server
serves audio + text for review, writes keep/delete decisions back to txt file.
usage:
pip install fastapi uvicorn
python qa_server.py --wavs dataset/test --txt data/neutral-voice-dataset-list-clean.txt
open http://localhost:8765
"""
import csv
import json
import argparse
import shutil
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import uvicorn
# ── args ──────────────────────────────────────────────────────────────────────
parser = argparse.ArgumentParser()
parser.add_argument("--wavs", default="dataset/test", help="folder with wav files")
parser.add_argument("--txt", required=True, help="txt file with sentences (one per line)")
parser.add_argument("--port", type=int, default=8765)
args = parser.parse_args()
WAV_DIR = Path(args.wavs)
TXT_FILE = Path(args.txt)
STATE_FILE = TXT_FILE.parent / (TXT_FILE.stem + "-qa-state.json")
META_FILE = Path("tts/dataset/metadata.csv")
# ── state ─────────────────────────────────────────────────────────────────────
def load_sentences() -> list[str]:
return [l.strip() for l in TXT_FILE.read_text(encoding="utf-8").splitlines() if l.strip()]
def load_state() -> dict:
if STATE_FILE.exists():
return json.loads(STATE_FILE.read_text())
return {} # index → "keep" | "delete"
def save_state(state: dict):
STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))
def _build_text_to_wav() -> dict[str, str]:
"""Read metadata.csv (read-only) and build text → absolute wav path mapping."""
mapping = {}
if not META_FILE.exists():
return mapping
with open(META_FILE, "r", encoding="utf-8") as f:
for row in csv.DictReader(f):
text = row["text"].strip()
file_path = row["file_path"].strip()
# resolve to absolute path
p = Path(file_path)
if not p.is_absolute():
# file_path is like "tts/dataset/neutral/0000.wav"
p = Path(__file__).resolve().parent.parent / file_path
if p.exists():
mapping[text] = str(p)
return mapping
# cache at startup
_TEXT_TO_WAV = _build_text_to_wav()
def wav_for(idx: int) -> Path | None:
"""Find wav file for sentence at txt line index via metadata.csv lookup."""
sentences = load_sentences()
if idx >= len(sentences):
return None
text = sentences[idx]
wav_path = _TEXT_TO_WAV.get(text)
if wav_path and Path(wav_path).exists():
return Path(wav_path)
return None
def get_deleted_indices() -> list[int]:
"""Get indices of deleted lines — from state first, then fallback to deleted/ folder."""
state = load_state()
indices = [int(k) for k, v in state.items() if v == "delete"]
if indices:
return sorted(indices)
# fallback: scan deleted/ folder for numeric wav filenames
deleted_dir = WAV_DIR / "deleted"
if not deleted_dir.exists():
return []
for p in deleted_dir.glob("*.wav"):
stem = p.stem.lstrip("0") or "0"
if stem.isdigit():
indices.append(int(stem))
return sorted(indices)
# ── app ───────────────────────────────────────────────────────────────────────
app = FastAPI()
@app.get("/api/info")
def info():
sentences = load_sentences()
state = load_state()
wavs = sorted(int(p.stem) for p in WAV_DIR.glob("*.wav") if p.stem.isdigit())
reviewed = len(state)
kept = sum(1 for v in state.values() if v == "keep")
deleted = sum(1 for v in state.values() if v == "delete")
missing = [i for i in range(len(sentences)) if wav_for(i) is None]
return {
"total": len(sentences),
"reviewed": reviewed,
"kept": kept,
"deleted": deleted,
"missing": missing,
"wavs": wavs,
}
@app.get("/api/next")
def next_unreviewed(after: int = -1):
sentences = load_sentences()
state = load_state()
for i in range(after + 1, len(sentences)):
if str(i) not in state and wav_for(i) is not None:
return {
"index": i,
"text": sentences[i],
"total": len(sentences),
"decision": state.get(str(i)),
}
return {"index": -1, "text": None, "total": len(sentences)}
@app.get("/api/sentence/{idx}")
def get_sentence(idx: int):
sentences = load_sentences()
state = load_state()
if idx >= len(sentences):
raise HTTPException(404)
return {
"index": idx,
"text": sentences[idx],
"total": len(sentences),
"decision": state.get(str(idx)),
"has_wav": wav_for(idx) is not None,
}
@app.get("/api/audio/{idx}")
def get_audio(idx: int):
wav = wav_for(idx)
if wav is None:
raise HTTPException(404, "wav not found")
return FileResponse(wav, media_type="audio/wav")
class Decision(BaseModel):
decision: str # "keep" | "delete"
@app.post("/api/decide/{idx}")
def decide(idx: int, body: Decision):
if body.decision not in ("keep", "delete"):
raise HTTPException(400, "decision must be keep or delete")
sentences = load_sentences()
if idx >= len(sentences):
raise HTTPException(404)
state = load_state()
state[str(idx)] = body.decision
save_state(state)
return {"ok": True}
@app.post("/api/export")
def export():
"""write final txt with only kept sentences + list of missing wavs."""
sentences = load_sentences()
state = load_state()
kept = []
missing = []
for i, s in enumerate(sentences):
decision = state.get(str(i))
if decision == "keep":
kept.append(s)
elif decision is None and wav_for(i) is None:
missing.append(i)
# deleted → skip
out_path = TXT_FILE.parent / (TXT_FILE.stem + "-final.txt")
out_path.write_text("\n".join(kept), encoding="utf-8")
missing_path = TXT_FILE.parent / (TXT_FILE.stem + "-missing.json")
missing_path.write_text(json.dumps(missing), encoding="utf-8")
return {
"kept": len(kept),
"missing": missing,
"out": str(out_path),
}
# ── ui ────────────────────────────────────────────────────────────────────────
HTML = r"""
Maven QA
Maven // Dataset QA
0 reviewed0 kept0 deleted0 total
— / —
⚠ no audio file
loading...
"""
@app.post("/api/reset_deleted")
def reset_deleted():
"""clear all delete decisions from the qa-state.json and remove corresponding lines from the txt file."""
deleted_indices = get_deleted_indices()
if deleted_indices:
# Remove lines from txt file by index (descending to preserve indices)
sentences = load_sentences()
for idx in sorted(deleted_indices, reverse=True):
if idx < len(sentences):
del sentences[idx]
TXT_FILE.write_text("\n".join(sentences) + "\n" if sentences else "", encoding="utf-8")
# Clear all delete entries from state
state = load_state()
cleared = []
for idx_str, decision in list(state.items()):
if decision == "delete":
del state[idx_str]
cleared.append(int(idx_str))
save_state(state)
return {"cleared": len(cleared), "lines_removed": len(deleted_indices), "indices": deleted_indices}
@app.post("/api/cleanup")
def cleanup():
"""move deleted wavs to deleted/ subfolder."""
state = load_state()
deleted_dir = WAV_DIR / "deleted"
deleted_dir.mkdir(exist_ok=True)
moved = []
errors = []
for idx_str, decision in state.items():
if decision != "delete":
continue
wav = wav_for(int(idx_str))
if wav is None:
continue
dest = deleted_dir / wav.name
try:
shutil.move(str(wav), str(dest))
moved.append(int(idx_str))
except Exception as e:
errors.append({"idx": int(idx_str), "error": str(e)})
return {"moved": len(moved), "errors": errors, "dest": str(deleted_dir)}
@app.get("/", response_class=HTMLResponse)
def ui():
return HTML
# ── run ───────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print(f"[*] wavs: {WAV_DIR}")
print(f"[*] txt: {TXT_FILE}")
print(f"[*] state: {STATE_FILE}")
print(f"[*] open http://localhost:{args.port}")
uvicorn.run(app, host="0.0.0.0", port=args.port, log_level="debug")