Blame for Studio.py

#!/usr/bin/env python3
"""

studio.py

A local web app for managing cloned voices: write a script, pick a voice, generate takes,

compare profiles, and approve audio. Runs on your own computer; nothing is uploaded anywhere.



  .venv-tts\\Scripts\\activate

  python studio.py



Then open http://127.0.0.1:7860 (it opens automatically). The voice model loads once and

stays loaded, so every generation after the first starts right away.



Options:

  --model standard|turbo   which Chatterbox model (voice profiles must match it)

  --port 7860

  --host 127.0.0.1         use 0.0.0.0 to let others on your network open it (see README)

"""

import os
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS_WARNING", "1")
os.environ.setdefault("HF_HUB_DISABLE_SYMLINKS", "1")

import argparse
import csv
import json
import math
import mimetypes
import queue
import random
import re
import shutil
import tempfile
import threading
import time
import traceback
import uuid
import webbrowser
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import parse_qs, unquote, urlparse

import soundfile as sf

import speak

ROOT = Path.cwd()
OUTPUT = ROOT / "output"
VOICES = ROOT / "voices"
LEGACY_REFERENCE_DIR = ROOT / "voice_refs"
NAMED_REFERENCE_SUFFIX = "_voice_refs"
REFERENCE_DIR = (
    ROOT / "brian_voice_refs"
    if (ROOT / "brian_voice_refs").exists() or not LEGACY_REFERENCE_DIR.exists()
    else LEGACY_REFERENCE_DIR
)
PEOPLE_MANIFEST = VOICES / "people.json"
REFERENCE_COUNT = 6
MAX_VIDEO_UPLOAD_BYTES = 8 * 1024 ** 3
MAX_REFERENCE_UPLOAD_BYTES = 100 * 1024 ** 2
VIDEO_UPLOAD_EXTS = {".mp4", ".mov", ".mkv", ".webm", ".avi", ".m4v", ".mts", ".wav", ".mp3", ".m4a"}
REFERENCE_LOCK = threading.Lock()
AUDIO_EXTS = {".wav", ".flac", ".mp3", ".ogg", ".m4a"}
HTML = Path(__file__).with_name("studio.html")
APP_ASSETS = {
    "/manifest.webmanifest": (Path(__file__).with_name("manifest.webmanifest"), "application/manifest+json"),
    "/mobile-manifest.webmanifest": (Path(__file__).with_name("mobile-manifest.webmanifest"), "application/manifest+json"),
    "/icon.svg": (Path(__file__).with_name("icon.svg"), "image/svg+xml"),
    "/sw.js": (Path(__file__).with_name("sw.js"), "text/javascript; charset=utf-8"),
}


def people_registry():
    if not PEOPLE_MANIFEST.is_file():
        return {"people": [], "sources": {}}
    data = json.loads(PEOPLE_MANIFEST.read_text(encoding="utf-8"))
    if not isinstance(data, dict) or not isinstance(data.get("people", []), list):
        raise ValueError("People registry is invalid.")
    data.setdefault("sources", {})
    return data


def save_people_registry(data):
    VOICES.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=VOICES,
                                     suffix=".tmp", delete=False) as handle:
        temp_path = Path(handle.name)
        json.dump(data, handle, indent=2)
    try:
        temp_path.replace(PEOPLE_MANIFEST)
    except Exception:
        temp_path.unlink(missing_ok=True)
        raise


def available_people():
    registry = people_registry()
    records = [{"id": "brian", "name": "Brian"}]
    records.extend(item for item in registry["people"] if item.get("id") != "brian")
    known_ids = {person["id"] for person in records}
    for directory in named_reference_dirs():
        folder_name = directory.name[:-len(NAMED_REFERENCE_SUFFIX)]
        person_id = re.sub(r"[^a-z0-9]+", "-", folder_name.lower()).strip("-")[:40].strip("-")
        if person_id and person_id not in known_ids:
            name = re.sub(r"[_-]+", " ", folder_name).strip().title()
            records.append({"id": person_id, "name": name})
            known_ids.add(person_id)
    return records


def get_person(person_id):
    if not isinstance(person_id, str) or not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,39}", person_id):
        raise ValueError("Choose a valid person.")
    for person in available_people():
        if person["id"] == person_id:
            return person
    raise ValueError("That person does not exist. Refresh the page and try again.")


def create_person(name):
    if not isinstance(name, str) or not name.strip():
        raise ValueError("Enter a name for the person.")
    name = name.strip()
    slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")[:40].strip("-")
    if not slug:
        raise ValueError("Use a name that includes at least one letter or number.")
    with REFERENCE_LOCK:
        registry = people_registry()
        if any(person["id"] == slug for person in available_people()):
            raise ValueError("A person with that name already exists.")
        person = {"id": slug, "name": name}
        registry["people"].append(person)
        registry["sources"].setdefault(slug, [])
        save_people_registry(registry)
    return person


def save_person_sources(person_id, sources):
    person = get_person(person_id)
    if not isinstance(sources, list):
        raise ValueError("Source folders must be a list.")
    available = {item["path"] for item, _ in profile_sources()}
    if any(not isinstance(source, str) or source not in available for source in sources):
        raise ValueError("One or more selected source folders are no longer available.")
    registry = people_registry()
    registry.setdefault("sources", {})[person["id"]] = list(dict.fromkeys(sources))
    save_people_registry(registry)
    return registry["sources"][person["id"]]


def person_profile_paths(person, kind):
    if kind not in {"standard", "turbo"}:
        raise ValueError("Choose a supported voice model.")
    if person["id"] == "brian":
        stem = "brian" if kind == "standard" else "brian_turbo"
        directory = VOICES
    else:
        stem = kind
        directory = VOICES / "people" / person["id"]
    active = directory / f"{stem}.pt"
    candidate_stem = f"{stem}_candidate"
    return {
        "active": active,
        "candidate": directory / f"{candidate_stem}.pt",
    }


def person_reference_dir(person):
    if person["id"] == "brian":
        return REFERENCE_DIR
    folder_name = re.sub(r"[^a-z0-9]+", "_", person["id"].lower()).strip("_")
    named_dir = ROOT / f"{folder_name}{NAMED_REFERENCE_SUFFIX}"
    if named_dir.is_dir():
        return named_dir
    return VOICES / "people" / person["id"] / "references"


def person_reference_manifest(person):
    return person_reference_dir(person) / "reference_pool.json"


def person_video_dir(person):
    return VOICES / "people" / person["id"] / "uploads" / "videos"


def person_dataset_dir(person):
    return VOICES / "people" / person["id"] / "datasets" / "prepared"


def person_reference_audio(person):
    directory = person_reference_dir(person)
    return [
        directory / name
        for name in [*(f"voice_ref{index}.wav" for index in range(1, 6)),
                     *(f"ref_{index:02d}.wav" for index in range(1, 7))]
        if (directory / name).is_file()
    ][:5]


def person_media_status(person_id):
    person = get_person(person_id)
    video_dir = person_video_dir(person)
    videos = sorted(
        path.name for path in video_dir.iterdir()
        if path.is_file() and path.suffix.lower() in
        {".mp4", ".mov", ".mkv", ".webm", ".avi", ".m4v", ".mts", ".wav", ".mp3", ".m4a"}
    ) if video_dir.is_dir() else []
    references = person_reference_audio(person)
    return {
        "person": person,
        "videos": videos,
        "references": [path.name for path in references],
        "dataset": rel(person_dataset_dir(person)),
        "dataset_ready": (person_dataset_dir(person) / "report.csv").is_file(),
    }


def delete_uploaded_voice_sample(person_id, name):
    person = get_person(person_id)
    if not isinstance(name, str) or not re.fullmatch(r"voice_ref[1-5]\.wav", name):
        raise ValueError("Choose one of this person's uploaded voice samples.")
    directory = person_reference_dir(person)
    target = directory / name
    with REFERENCE_LOCK:
        if not target.is_file():
            raise ValueError("That voice sample is no longer available.")
        target.unlink()
        manifest_path = person_reference_manifest(person)
        if manifest_path.is_file():
            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
            slots = manifest.get("slots", {})
            if name in slots:
                del slots[name]
                save_reference_manifest(manifest, person_id)
    return {"deleted": name, "media": person_media_status(person_id)}


def managed_reference_dirs():
    return [person_reference_dir(person) for person in available_people()]


def named_reference_dirs():
    return sorted(
        path for path in ROOT.iterdir()
        if path.is_dir() and path.name.lower().endswith(NAMED_REFERENCE_SUFFIX)
    )


def clip_dirs():
    return list(dict.fromkeys([
        ROOT / "refs",
        LEGACY_REFERENCE_DIR,
        *named_reference_dirs(),
        *[person_reference_dir(person) for person in available_people() if person["id"] != "brian"],
        *[person_video_dir(person) for person in available_people()],
    ]))


# ---------------------------------------------------------------- engine (one model, one job at a time)

class Engine:
    def __init__(self, kind, device):
        self.kind = kind
        self.device = device
        self.model = None
        self.error = None
        self.voice_key = None
        self.jobs = {}
        self.rebuild_jobs = {}
        self.dataset_jobs = {}
        self.compare_jobs = {}
        self.rebuild_submit_lock = threading.Lock()
        self.dataset_submit_lock = threading.Lock()
        self.queue = queue.Queue()
        self.lock = threading.Lock()
        threading.Thread(target=self._load, daemon=True).start()
        threading.Thread(target=self._worker, daemon=True).start()

    # -- model
    def _load(self):
        try:
            self.model = speak.load_model(self.kind, self.device)
        except Exception as e:
            self.error = f"The voice model didn't load: {e}"
            traceback.print_exc()

    def status(self):
        busy = sum(1 for j in self.jobs.values() if j["state"] in ("queued", "running"))
        return {"ready": self.model is not None, "error": self.error, "model": self.kind,
                "device": self.device, "jobs_waiting": busy}

    # -- voices
    def use_voice(self, voice):
        """voice = {"type": "profile"|"clip", "path": ...}; only re-processes when it changes."""
        path = safe_path(voice["path"], [VOICES] if voice["type"] == "profile" else clip_dirs())
        file_version = None
        if voice["type"] == "clip":
            stat = path.stat()
            file_version = (stat.st_mtime_ns, stat.st_size)
        key = (voice["type"], voice["path"], voice.get("exaggeration") if voice["type"] == "clip" else None,
               file_version)
        if key == self.voice_key:
            return
        if voice["type"] == "profile":
            meta = path.with_suffix(".json")
            if meta.exists():
                made_for = json.loads(meta.read_text(encoding="utf-8")).get("model", self.kind)
                if made_for != self.kind:
                    raise ValueError(f"{path.name} was built for the {made_for} model, but the studio is "
                                     f"running {self.kind}. Restart with --model {made_for}.")
            speak.load_voice_profile(self.model, self.kind, path, self.device)
        else:
            if sf.info(str(path)).duration < 5.5:
                raise ValueError(f"{path.name} is too short to copy a voice from. Use a clip of 7-15 seconds.")
            speak.prepare_voice(self.model, self.kind, path, voice.get("exaggeration", 0.5))
        self.voice_key = key

    # -- jobs
    def submit(self, req):
        job_id = uuid.uuid4().hex[:10]
        voice = req.get("voice") or {}
        safe_path(voice.get("path", ""), [VOICES] if voice.get("type") == "profile" else clip_dirs())
        text = (req.get("text") or "").strip()
        if not text:
            raise ValueError("Write a script first.")
        if len(text) > 6000:
            raise ValueError("That script is very long. Split it into parts of about 6,000 characters.")
        takes = max(1, min(int(req.get("takes", 1)), 5))
        seed = req.get("seed")
        seed = int(seed) if str(seed).strip().lstrip("-").isdigit() else random.randint(0, 2 ** 31 - 1)
        chunks = speak.split_text(text, int(req.get("max_chars", 280)))
        if not chunks:
            raise ValueError("Write a script first.")
        job = {
            "id": job_id, "state": "queued", "created": time.time(), "text": text,
            "voice": req["voice"], "takes_wanted": takes, "seed": seed,
            "settings": {
                "exaggeration": float(req.get("exaggeration", 0.5)),
                "cfg": float(req.get("cfg", 0.5)),
                "temperature": float(req.get("temperature", 0.8)),
                "pause": float(req.get("pause", 0.25)),
                "para_pause": float(req.get("para_pause", 0.7)),
                "max_chars": int(req.get("max_chars", 280)),
            },
            "chunks": chunks, "take": 0, "piece": 0, "pieces": len(chunks) * takes,
            "done_pieces": 0, "results": [], "error": None,
        }
        self.jobs[job_id] = job
        self.queue.put(job_id)
        return job

    def _worker(self):
        while True:
            job_id = self.queue.get()
            job = self.jobs.get(job_id)
            if not job:
                continue
            while self.model is None and self.error is None:
                time.sleep(0.5)
            if self.error:
                job.update(state="error", error=self.error)
                continue
            job["state"] = "running"
            job["started"] = time.time()
            try:
                with self.lock:
                    self._run(job)
                job["state"] = "done"
            except ValueError as e:  # expected problems (bad voice file etc.): message only
                job.update(state="error", error=str(e))
            except Exception as e:
                traceback.print_exc()
                job.update(state="error", error=f"Generation failed: {e}")
            job["finished"] = time.time()

    def _run(self, job):
        s = job["settings"]
        voice = dict(job["voice"], exaggeration=s["exaggeration"])
        self.use_voice(voice)
        args = SimpleNamespace(exaggeration=s["exaggeration"], cfg=s["cfg"], temperature=s["temperature"])
        stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        OUTPUT.mkdir(parents=True, exist_ok=True)
        for k in range(job["takes_wanted"]):
            seed = job["seed"] + k
            job["take"] = k + 1
            speak.seed_everything(seed)
            pieces = []
            for i, (chunk, pause_kind) in enumerate(job["chunks"], 1):
                job["piece"] = i
                wav = speak.generate_piece(self.model, self.kind, chunk, args)
                pieces.append((speak.trim(wav, self.model.sr), pause_kind))
                job["done_pieces"] += 1
            audio = speak.assemble(pieces, self.model.sr, s["pause"], s["para_pause"])
            name = f"studio_{stamp}_{job['id']}_take{k + 1}.wav"
            path = OUTPUT / name
            sf.write(str(path), audio, self.model.sr, subtype="PCM_16")
            settings = {
                "text": job["text"],
                "reference": job["voice"]["path"] if job["voice"]["type"] == "clip" else None,
                "voice_profile": job["voice"]["path"] if job["voice"]["type"] == "profile" else None,
                "model": self.kind, "seed": seed, **{k2: s[k2] for k2 in
                    ("exaggeration", "cfg", "temperature", "pause", "para_pause", "max_chars")},
                "created": datetime.now().isoformat(timespec="seconds"),
                "take": k + 1, "of_takes": job["takes_wanted"], "source": "studio",
                "approved": False,
                "note": "AI-generated voice (Chatterbox, Perth watermark).",
            }
            path.with_suffix(".json").write_text(json.dumps(settings, indent=2), encoding="utf-8")
            job["results"].append(entry_for(path, settings))

    def public(self, job):
        keys = ("id", "state", "take", "takes_wanted", "piece", "pieces", "done_pieces",
                "results", "error", "seed")
        out = {k: job.get(k) for k in keys}
        out["chunks"] = len(job["chunks"])
        if job.get("started"):
            out["elapsed"] = round((job.get("finished") or time.time()) - job["started"], 1)
        ahead = [j for j in self.jobs.values()
                 if j["state"] in ("queued", "running") and j["created"] < job["created"]]
        out["ahead"] = len(ahead)
        return out

    # -- voice profile rebuilds
    def submit_dataset(self, person_id):
        person = get_person(person_id)
        media = person_media_status(person["id"])
        references = person_reference_audio(person)
        if not media["videos"]:
            raise ValueError(f"Upload one or more videos for {person['name']} first.")
        if not 1 <= len(references) <= 5:
            raise ValueError(
                f"Upload 1-5 clean 10-30 second WAV voice samples for {person['name']} first."
            )
        with self.dataset_submit_lock:
            if any(job["state"] in ("queued", "running") for job in self.dataset_jobs.values()):
                raise ValueError("A video-to-dataset job is already in progress.")
            job_id = uuid.uuid4().hex[:10]
            job = {
                "id": job_id, "person": person["id"], "state": "queued", "progress": 0,
                "message": "Waiting to prepare clips…", "error": None, "result": None,
                "created": time.time(),
            }
            self.dataset_jobs[job_id] = job
            settings = {
                "person": person["id"],
                "input": person_video_dir(person),
                "references": references,
                "output": person_dataset_dir(person),
            }
            threading.Thread(target=self._dataset_worker, args=(job_id, settings), daemon=True).start()
        return self.public_dataset(job)

    def _dataset_worker(self, job_id, settings):
        job = self.dataset_jobs[job_id]
        job.update(state="running", started=time.time(), message="Starting clip preparation…")
        try:
            import prepare_dataset

            def report(message, progress):
                job.update(message=message, progress=progress)

            while self.model is None and self.error is None:
                job.update(message="Waiting for Studio to finish loading before preparing clips…")
                time.sleep(0.5)

            device = self.device if self.device in {"cpu", "cuda"} else \
                ("cuda" if prepare_dataset.torch.cuda.is_available() else "cpu")
            args = SimpleNamespace(
                input=settings["input"], reference=settings["references"], out=settings["output"],
                threshold=0.60, sentence_threshold=0.55, min_unit=0.6, sample_rate=44100,
                min_len=3.0, max_len=15.0, max_gap=0.6, pad=0.10, min_words=2,
                min_dbfs=-40.0, whisper_model="large-v3", language="en", device=device,
                keep_temp=False,
            )
            with self.lock:
                reload_voice_model = self.model is not None
                if reload_voice_model:
                    job.update(message="Freeing GPU memory for video processing…")
                    self.model = None
                    self.voice_key = None
                    import gc
                    gc.collect()
                    if prepare_dataset.torch.cuda.is_available():
                        prepare_dataset.torch.cuda.empty_cache()
                try:
                    result = prepare_dataset.run_dataset(args, progress=report)
                finally:
                    if reload_voice_model:
                        job.update(message="Reloading Studio’s voice model…", progress=99)
                        gc.collect()
                        if prepare_dataset.torch.cuda.is_available():
                            prepare_dataset.torch.cuda.empty_cache()
                        try:
                            self.model = speak.load_model(self.kind, self.device)
                        except Exception as e:
                            self.error = f"The voice model didn't reload after video preparation: {e}"
                            raise RuntimeError(self.error) from e
                        self.voice_key = None
            if result["videos_processed"] == 0 and result["failures"]:
                failed = "; ".join(
                    f"{item['video']}: {item['error']}" for item in result["failures"]
                )
                raise RuntimeError(f"No videos were processed successfully. Failed: {failed}")
            registry = people_registry()
            sources = registry.setdefault("sources", {}).setdefault(settings["person"], [])
            dataset_path = rel(settings["output"])
            if dataset_path not in sources:
                sources.append(dataset_path)
                save_people_registry(registry)
            job.update(
                state="done", progress=100,
                message=f"Prepared {result['clips_kept']} clips from {result['videos_processed']} videos.",
                result={**result, "dataset": dataset_path},
            )
        except (ValueError, RuntimeError) as e:
            job.update(state="error", error=str(e), message="Video clip preparation failed.")
        except Exception as e:
            traceback.print_exc()
            job.update(state="error", error=f"Video clip preparation failed: {e}",
                       message="Video clip preparation failed.")
        finally:
            job["finished"] = time.time()

    @staticmethod
    def public_dataset(job):
        result = job.get("result")
        if result:
            result = {key: result.get(key) for key in (
                "videos_found", "videos_processed", "videos_skipped", "clips_kept",
                "failures", "dataset",
            )}
        return {
            **{key: job.get(key) for key in (
                "id", "person", "state", "progress", "message", "error",
            )},
            "result": result,
        }

    def submit_rebuild(self, req):
        if not isinstance(req, dict):
            raise ValueError("Rebuild request must be an object.")

        person = get_person(req.get("person", "brian"))
        available_sources = {item["path"]: path for item, path in profile_sources()}
        requested_sources = req.get("sources")
        if not isinstance(requested_sources, list) or not requested_sources:
            raise ValueError("Select at least one audio source.")
        if any(not isinstance(source, str) or source not in available_sources for source in requested_sources):
            raise ValueError("One or more selected audio sources are no longer available. Refresh the page.")
        available_anchors = {item["path"]: path for item, path in profile_anchors(requested_sources)}
        anchor_key = req.get("anchor")
        if not isinstance(anchor_key, str) or anchor_key not in available_anchors:
            raise ValueError("Choose an available anchor clip.")
        try:
            max_clips = int(req.get("max_clips", 400))
            min_sim = float(req.get("min_sim", 0.65))
            blend = float(req.get("blend", 1.0))
        except (TypeError, ValueError) as e:
            raise ValueError("Rebuild settings must be valid numbers.") from e
        if not 3 <= max_clips <= 4000 or not math.isfinite(min_sim) or not math.isfinite(blend) or \
                not 0 <= min_sim <= 1 or not 0 <= blend <= 1:
            raise ValueError("Use 3-4000 clips, similarity 0-1, and blend 0-1.")

        with self.rebuild_submit_lock:
            if any(job["state"] in ("queued", "running") for job in self.rebuild_jobs.values()):
                raise ValueError("A voice rebuild is already in progress.")
            job_id = uuid.uuid4().hex[:10]
            job = {
                "id": job_id, "state": "queued", "progress": 0, "message": "Waiting for the voice model…",
                "error": None, "result": None, "created": time.time(),
            }
            self.rebuild_jobs[job_id] = job
            settings = {
                "person": person["id"],
                "sources": [available_sources[source] for source in dict.fromkeys(requested_sources)],
                "anchor": available_anchors[anchor_key],
                "max_clips": max_clips, "min_sim": min_sim, "blend": blend,
            }
            threading.Thread(target=self._rebuild_worker, args=(job_id, settings), daemon=True).start()
        return self.public_rebuild(job)

    def _rebuild_worker(self, job_id, settings):
        job = self.rebuild_jobs[job_id]
        while self.model is None and self.error is None:
            time.sleep(0.5)
        if self.error:
            job.update(state="error", error=self.error, message="The voice model failed to load.")
            return
        job.update(state="running", started=time.time(), message="Starting voice profile rebuild…")
        try:
            from build_voice import build_voice_profile

            def report(message, progress):
                job.update(message=message, progress=progress)

            profile_path = person_profile_paths(get_person(settings["person"]), self.kind)["candidate"]
            with self.lock:
                result = build_voice_profile(
                    self.model, self.kind, self.device, settings["sources"], settings["anchor"],
                    profile_path, max_clips=settings["max_clips"], min_sim=settings["min_sim"],
                    blend=settings["blend"], progress=report,
                )
                self.voice_key = None
            job.update(state="done", progress=100, message="Voice profile rebuilt.",
                       result={**result, "path": rel(profile_path)})
        except ValueError as e:
            job.update(state="error", error=str(e), message="Voice profile rebuild failed.")
        except Exception as e:
            traceback.print_exc()
            job.update(state="error", error=f"Voice profile rebuild failed: {e}",
                       message="Voice profile rebuild failed.")
        finally:
            job["finished"] = time.time()

    @staticmethod
    def public_rebuild(job):
        result = job.get("result")
        if result:
            result = {key: result.get(key) for key in
                      ("path", "clips_used", "clips_rejected_quality", "source_groups",
                       "anchor_agreement", "blend", "created")}
        return {
            **{key: job.get(key) for key in ("id", "state", "progress", "message", "error")},
            "result": result,
        }

    def submit_compare(self, req):
        if not isinstance(req, dict):
            raise ValueError("Comparison request must be an object.")
        text = req.get("text")
        if not isinstance(text, str) or not text.strip() or len(text) > 500:
            raise ValueError("Enter a comparison script of 1-500 characters.")
        person = get_person(req.get("person", "brian"))
        paths = person_profile_paths(person, self.kind)
        current, candidate = paths["active"], paths["candidate"]
        if not current.is_file() or not candidate.is_file():
            raise ValueError("Build both the current voice profile and a candidate before comparing.")
        settings = {
            "text": text.strip(),
            "exaggeration": self._bounded_setting(req.get("exaggeration", 0.5), 0.25, 1.2, "Energy"),
            "cfg": self._bounded_setting(req.get("cfg", 0.5), 0.1, 0.9, "Pace"),
            "temperature": self._bounded_setting(req.get("temperature", 0.8), 0.4, 1.1, "Variation"),
            "seed": self._bounded_seed(req.get("seed")),
            "current": current,
            "candidate": candidate,
        }
        job_id = uuid.uuid4().hex[:10]
        job = {"id": job_id, "state": "queued", "progress": 0, "message": "Waiting for voice model…",
               "error": None, "result": None}
        self.compare_jobs[job_id] = job
        threading.Thread(target=self._compare_worker, args=(job_id, settings), daemon=True).start()
        return self.public_compare(job)

    @staticmethod
    def _bounded_setting(value, minimum, maximum, label):
        try:
            number = float(value)
        except (TypeError, ValueError) as e:
            raise ValueError(f"{label} must be a number.") from e
        if not math.isfinite(number) or not minimum <= number <= maximum:
            raise ValueError(f"{label} must be between {minimum} and {maximum}.")
        return number

    @staticmethod
    def _bounded_seed(value):
        try:
            seed = int(value)
        except (TypeError, ValueError) as e:
            raise ValueError("Comparison seed must be an integer.") from e
        if not 0 <= seed <= 2**31 - 1:
            raise ValueError("Comparison seed must be between 0 and 2147483647.")
        return seed

    def _compare_worker(self, job_id, settings):
        job = self.compare_jobs[job_id]
        while self.model is None and self.error is None:
            time.sleep(0.5)
        if self.error:
            job.update(state="error", error=self.error, message="The voice model failed to load.")
            return
        job.update(state="running", message="Generating matched A/B samples…")
        compare_dir = OUTPUT / ".compare"
        outputs = []
        args = SimpleNamespace(
            exaggeration=settings["exaggeration"], cfg=settings["cfg"], temperature=settings["temperature"]
        )
        chunks = speak.split_text(settings["text"], 280)
        if not chunks:
            job.update(state="error", error="Enter a comparison script.", message="A/B comparison failed.")
            return
        try:
            with self.lock:
                previous_conds = self.model.conds
                previous_voice_key = self.voice_key
                try:
                    compare_dir.mkdir(parents=True, exist_ok=True)
                    for index, profile in enumerate((settings["current"], settings["candidate"])):
                        job.update(progress=index * 50, message=f"Generating {'A' if index == 0 else 'B'} sample…")
                        self.use_voice({"type": "profile", "path": rel(profile)})
                        speak.seed_everything(settings["seed"])
                        pieces = []
                        for chunk, pause_kind in chunks:
                            wav = speak.generate_piece(self.model, self.kind, chunk, args)
                            pieces.append((speak.trim(wav, self.model.sr), pause_kind))
                        audio = speak.assemble(pieces, self.model.sr, 0.25, 0.7)
                        label = "current" if index == 0 else "candidate"
                        path = compare_dir / f"{job_id}_{label}.wav"
                        sf.write(str(path), audio, self.model.sr, subtype="PCM_16")
                        outputs.append({"label": label, "file": rel(path)})
                finally:
                    self.model.conds = previous_conds
                    self.voice_key = previous_voice_key
            job.update(state="done", progress=100, message="A/B samples are ready.", result=outputs)
        except ValueError as e:
            job.update(state="error", error=str(e), message="A/B comparison failed.")
        except Exception as e:
            traceback.print_exc()
            job.update(state="error", error=f"A/B comparison failed: {e}", message="A/B comparison failed.")
    @staticmethod
    def public_compare(job):
        return {key: job.get(key) for key in
                ("id", "state", "progress", "message", "error", "result")}

    def adopt_candidate(self, person_id="brian"):
        person = get_person(person_id)
        paths = person_profile_paths(person, self.kind)
        current, candidate = paths["active"], paths["candidate"]
        candidate_meta = candidate.with_suffix(".json")
        if not candidate.is_file() or not candidate_meta.is_file():
            raise ValueError("Build a candidate profile before using it.")
        try:
            metadata = json.loads(candidate_meta.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as e:
            raise ValueError("Candidate profile metadata is unreadable.") from e
        if metadata.get("model") != self.kind:
            raise ValueError(f"Candidate profile is not built for the {self.kind} model.")
        with self.lock:
            current.parent.mkdir(parents=True, exist_ok=True)
            temp_profile = current.with_name(current.name + ".tmp")
            temp_meta = current.with_suffix(".json.tmp")
            try:
                import shutil
                shutil.copy2(candidate, temp_profile)
                shutil.copy2(candidate_meta, temp_meta)
                os.replace(temp_profile, current)
                os.replace(temp_meta, current.with_suffix(".json"))
            finally:
                for path in (temp_profile, temp_meta):
                    try:
                        path.unlink()
                    except FileNotFoundError:
                        pass
            self.voice_key = None
        return {"path": rel(current), "created": metadata.get("created"), "clips_used": metadata.get("clips_used")}


# ---------------------------------------------------------------- files

def safe_path(rel, allowed_dirs):
    """Resolve a path the browser sent and make sure it stays inside the allowed folders."""
    p = (ROOT / unquote(rel)).resolve()
    for d in allowed_dirs:
        try:
            p.relative_to(d.resolve())
            if p.is_file():
                return p
        except ValueError:
            continue
    raise ValueError("That file isn't available.")


def rel(p):
    return p.resolve().relative_to(ROOT.resolve()).as_posix()


def entry_for(wav, meta):
    try:
        seconds = round(sf.info(str(wav)).duration, 1)
    except Exception:
        seconds = None
    return {
        "file": rel(wav), "name": wav.name, "seconds": seconds,
        "text": meta.get("text", ""), "seed": meta.get("seed"),
        "voice": Path(meta.get("voice_profile") or meta.get("reference") or "").name or None,
        "created": meta.get("created"), "take": meta.get("take"), "of_takes": meta.get("of_takes"),
        "approved": bool(meta.get("approved")), "approved_at": meta.get("approved_at"),
        "settings": {k: meta.get(k) for k in ("exaggeration", "cfg", "temperature", "model")},
    }


def list_voices():
    profiles = []
    people = {person["id"]: person for person in available_people()}
    for p in sorted(VOICES.rglob("*.pt")) if VOICES.exists() else []:
        meta = {}
        if p.with_suffix(".json").exists():
            try:
                meta = json.loads(p.with_suffix(".json").read_text(encoding="utf-8"))
            except Exception:
                pass
        person_id = p.parent.name if p.parent.parent == VOICES / "people" else "brian"
        person = people.get(person_id, {"name": p.stem})
        model_name = meta.get("model")
        display_name = person["name"] if person_id == "brian" else \
            f"{person['name']} ({model_name or p.stem})"
        profiles.append({"type": "profile", "path": rel(p), "name": display_name,
                         "clips": meta.get("clips_used"), "model": meta.get("model"),
                         "anchor": Path(meta.get("anchor", "")).name or None})
    clips = []
    managed_dirs = {path.resolve() for path in managed_reference_dirs()}
    for d in clip_dirs():
        if not d.exists():
            continue
        for p in sorted(d.glob("*.wav")):
            if p.parent == REFERENCE_DIR and not re.fullmatch(r"ref_0[1-6]\.wav", p.name):
                continue
            if p.parent != REFERENCE_DIR and p.name.startswith("ref_") and not re.fullmatch(r"ref_0[1-6]\.wav", p.name):
                continue
            try:
                dur = sf.info(str(p)).duration
            except Exception:
                continue
            active_reference = (
                p.parent.resolve() in managed_dirs and
                re.fullmatch(r"(?:ref_0[1-6]|voice_ref[1-6])\.wav", p.name)
            )
            if active_reference or 5.5 <= dur <= 40:
                clips.append({"type": "clip", "path": rel(p), "name": f"{d.name}/{p.stem}",
                              "seconds": round(dur, 1)})
    return {"profiles": profiles, "clips": clips}


def profile_sources():
    sources = []
    candidates = [
        ROOT / "refs", LEGACY_REFERENCE_DIR, *named_reference_dirs(),
        *managed_reference_dirs(),
        *[person_dataset_dir(person) for person in available_people()],
    ]
    candidates.extend(
        path for path in ROOT.iterdir()
        if path.is_dir() and (path / "report.csv").is_file() and (path / "wavs").is_dir()
    )
    seen = set()
    for path in candidates:
        if not path.is_dir() or path.resolve() in seen:
            continue
        seen.add(path.resolve())
        if (path / "report.csv").is_file() and (path / "wavs").is_dir():
            with open(path / "report.csv", newline="", encoding="utf-8") as f:
                count = sum(1 for row in csv.DictReader(f) if row.get("status") == "kept")
        else:
            count = sum(1 for item in path.rglob("*") if item.is_file() and item.suffix.lower() in
                        {".wav", ".flac", ".mp3", ".ogg", ".m4a"})
        if count:
            sources.append(({"path": rel(path), "name": path.name, "clips": count}, path))
    return sources


def profile_anchors(requested_sources=None):
    sources = profile_sources()
    if requested_sources is None:
        folders = [path for _, path in sources]
    else:
        available = {item["path"]: path for item, path in sources}
        if any(source not in available for source in requested_sources):
            raise ValueError("One or more selected source folders are no longer available.")
        folders = [available[source] for source in dict.fromkeys(requested_sources)]
    anchors, seen = [], set()
    for folder in folders:
        search_root = folder / "wavs" if (folder / "report.csv").is_file() and (folder / "wavs").is_dir() else folder
        for path in search_root.rglob("*"):
            if not path.is_file() or path.suffix.lower() not in AUDIO_EXTS or path.resolve() in seen:
                continue
            seen.add(path.resolve())
            try:
                duration = sf.info(str(path)).duration
            except (OSError, RuntimeError):
                continue
            if 5.5 <= duration <= 40:
                anchors.append(({"path": rel(path), "name": f"{path.parent.name}/{path.name}",
                                 "seconds": round(duration, 1)}, path))
    return anchors


def reference_candidates(requested_sources):
    available = {item["path"]: path for item, path in profile_sources()}
    if not isinstance(requested_sources, list) or not requested_sources:
        raise ValueError("Select at least one audio source.")
    if any(not isinstance(source, str) or source not in available for source in requested_sources):
        raise ValueError("One or more selected audio sources are no longer available. Refresh the page.")

    from pick_voice_ref import analyze

    candidates, seen = [], set()
    for source_key in dict.fromkeys(requested_sources):
        folder = available[source_key]
        if (folder / "report.csv").is_file() and (folder / "wavs").is_dir():
            with open(folder / "report.csv", newline="", encoding="utf-8") as report:
                rows = ((row, folder / "wavs" / f"{row.get('clip_id', '')}.wav")
                        for row in csv.DictReader(report) if row.get("status") == "kept")
                for row, path in rows:
                    try:
                        similarity = float(row["similarity"])
                    except (KeyError, TypeError, ValueError):
                        continue
                    add_reference_candidate(candidates, seen, path, similarity, row.get("text", ""), analyze)
        else:
            for path in folder.rglob("*"):
                if path.is_file() and path.suffix.lower() in AUDIO_EXTS:
                    add_reference_candidate(candidates, seen, path, None, "", analyze)
    candidates.sort(key=lambda item: (-item["score"], item["path"]))
    return candidates


def add_reference_candidate(candidates, seen, path, similarity, text, analyze):
    resolved = path.resolve()
    if resolved.parent in {directory.resolve() for directory in managed_reference_dirs()} and \
            re.fullmatch(r"(?:ref_0[1-6]|voice_ref[1-5])\.wav", resolved.name):
        return
    try:
        resolved.relative_to(REFERENCE_DIR.resolve())
    except ValueError:
        pass
    else:
        return
    if resolved in seen or not resolved.is_file():
        return
    seen.add(resolved)
    try:
        quality = analyze(resolved)
    except (OSError, RuntimeError, ValueError):
        return
    if quality is None or not 7 <= quality["seconds"] <= 16 or quality["clipped"] > 0.0005:
        return
    match_score = similarity if similarity is not None else 0.5
    length_bonus = 1.0 - min(abs(quality["seconds"] - 12.0) / 6.0, 1.0)
    candidates.append({
        "path": rel(resolved), "name": resolved.name, "seconds": round(quality["seconds"], 1),
        "source": resolved.parent.parent.name if resolved.parent.name == "wavs" else resolved.parent.name,
        "similarity": round(similarity, 3) if similarity is not None else None,
        "score": round(3.0 * match_score + quality["speech_ratio"] + 0.5 * length_bonus, 4),
        "text": text,
    })


def reference_items(person_id="brian"):
    person = get_person(person_id)
    reference_dir = person_reference_dir(person)
    manifest_path = person_reference_manifest(person)
    manifest = json.loads(manifest_path.read_text(encoding="utf-8")) if manifest_path.is_file() else {}
    slots = manifest.get("slots", {})
    items = []
    for index in range(1, REFERENCE_COUNT + 1):
        standard_name = f"ref_{index:02d}.wav"
        legacy_name = f"voice_ref{index}.wav"
        filename = standard_name if (reference_dir / standard_name).is_file() else legacy_name
        path = reference_dir / filename
        if not path.is_file():
            continue
        entry = slots.get(filename, {})
        try:
            seconds = round(sf.info(str(path)).duration, 1)
        except (OSError, RuntimeError):
            seconds = None
        items.append({
            "slot": filename, "file": rel(path), "name": filename, "seconds": seconds,
            "source": entry.get("source"), "similarity": entry.get("similarity"),
            "source_clip": Path(entry.get("path", "")).name or None,
            "text": entry.get("text", ""),
        })
    return items


def stage_reference_copy(source, person_id="brian"):
    reference_dir = person_reference_dir(get_person(person_id))
    reference_dir.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile(dir=reference_dir, suffix=".wav", delete=False) as handle:
        staged_path = Path(handle.name)
    try:
        audio, sample_rate = sf.read(str(source), dtype="float32", always_2d=True)
        sf.write(str(staged_path), audio, sample_rate, subtype="PCM_16")
    except Exception:
        staged_path.unlink(missing_ok=True)
        raise
    return staged_path


def save_reference_manifest(manifest, person_id="brian"):
    reference_dir = person_reference_dir(get_person(person_id))
    manifest_path = person_reference_manifest(get_person(person_id))
    reference_dir.mkdir(parents=True, exist_ok=True)
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=reference_dir,
                                     suffix=".tmp", delete=False) as handle:
        temp_path = Path(handle.name)
        json.dump(manifest, handle, indent=2)
    try:
        temp_path.replace(manifest_path)
    except Exception:
        temp_path.unlink(missing_ok=True)
        raise


def active_reference_score(path, manifest, person_id="brian"):
    known = manifest.get("slots", {})
    entry = known.get(path.name, {})
    if entry.get("managed_path") == rel(path) or "score" in entry:
        return float(entry.get("score", 0))
    from pick_voice_ref import analyze
    try:
        quality = analyze(path)
    except (OSError, RuntimeError, ValueError):
        return 0
    if quality is None:
        return 0
    length_bonus = 1.0 - min(abs(quality["seconds"] - 12.0) / 6.0, 1.0)
    return 1.5 + quality["speech_ratio"] + 0.5 * length_bonus


def start_reference_review(requested_sources, person_id="brian"):
    person = get_person(person_id)
    candidates = reference_candidates(requested_sources)
    selected_sources = list(dict.fromkeys(requested_sources))
    with REFERENCE_LOCK:
        manifest_path = person_reference_manifest(person)
        manifest = json.loads(manifest_path.read_text(encoding="utf-8")) \
            if manifest_path.is_file() else {}
        manifest["review_sources"] = selected_sources
        manifest["review_pool"] = candidates
        manifest.setdefault("reviewed", {})
        save_reference_manifest(manifest, person_id)
        return next_reference_candidate(manifest)


def next_reference_candidate(manifest):
    reviewed = manifest.get("reviewed", {})
    promoted = set(manifest.get("promoted", []))
    active = {entry.get("path") for entry in manifest.get("slots", {}).values()}
    remaining = [
        candidate for candidate in manifest.get("review_pool", [])
        if candidate.get("path") not in reviewed and candidate.get("path") not in promoted and
        candidate.get("path") not in active and Path(ROOT / candidate.get("path", "")).is_file()
    ]
    result = {
        "top_candidates": remaining[:6],
        "remaining": len(remaining),
    }
    if remaining:
        return {**result, "candidate": remaining[0]}
    return {**result, "candidate": None}
def mark_dataset_clip_deleted(source_root, audio_path):
    report_path = source_root / "report.csv"
    if not report_path.is_file():
        return
    with report_path.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle)
        fieldnames = reader.fieldnames
        rows = list(reader)
    if not fieldnames or not {"clip_id", "status", "reason"}.issubset(fieldnames):
        raise ValueError("The dataset report format is not recognized; the clip was not deleted.")
    clip_id = audio_path.stem
    matches = [row for row in rows if row.get("clip_id") == clip_id]
    if len(matches) != 1:
        raise ValueError("The dataset report does not identify this clip uniquely; it was not deleted.")
    matches[0]["status"] = "rejected"
    matches[0]["reason"] = "deleted_in_studio_review"
    with tempfile.NamedTemporaryFile("w", newline="", encoding="utf-8",
                                     dir=source_root, suffix=".tmp", delete=False) as handle:
        temp_path = Path(handle.name)
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)
    try:
        temp_path.replace(report_path)
    except Exception:
        temp_path.unlink(missing_ok=True)
        raise


def review_reference(action, candidate_path, requested_sources, person_id="brian"):
    person = get_person(person_id)
    reference_dir = person_reference_dir(person)
    manifest_path = person_reference_manifest(person)
    if action not in {"keep", "delete", "promote"}:
        raise ValueError("Choose Keep, Delete, or Promote.")
    if not isinstance(candidate_path, str):
        raise ValueError("No review clip was selected.")
    if not isinstance(requested_sources, list) or not requested_sources:
        raise ValueError("Select at least one source folder.")
    if not manifest_path.is_file():
        raise ValueError("Start a reference review before choosing an action.")
    with REFERENCE_LOCK:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
        sources = manifest.get("review_sources", [])
        if list(dict.fromkeys(requested_sources or [])) != sources:
            raise ValueError("The selected sources changed. Restart the reference review.")
        available = {item["path"]: path for item, path in profile_sources()}
        if not sources or any(source not in available for source in sources):
            raise ValueError("A selected source folder is no longer available.")
        candidate = next((item for item in manifest.get("review_pool", [])
                          if item.get("path") == candidate_path), None)
        if candidate is None:
            raise ValueError("That clip is not in the current review queue.")
        allowed_dirs = [available[source] for source in sources]
        audio_path = safe_path(candidate_path, allowed_dirs)
        if candidate_path in manifest.get("reviewed", {}):
            raise ValueError("This clip has already been reviewed.")

        if action == "delete":
            source_root = max(
                (folder for folder in allowed_dirs
                 if audio_path == folder or folder in audio_path.parents),
                key=lambda folder: len(folder.parts), default=None,
            )
            if source_root is None:
                raise ValueError("The clip is outside the selected source folders.")
            report_path = source_root / "report.csv"
            report_backup = report_path.read_bytes() if report_path.is_file() else None
            if report_path.is_file():
                mark_dataset_clip_deleted(source_root, audio_path)
            try:
                audio_path.unlink()
            except OSError:
                if report_backup is not None:
                    report_path.write_bytes(report_backup)
                raise
            manifest.setdefault("rejected", []).append(candidate_path)
            result_message = "Source clip deleted."
        elif action == "keep":
            manifest.setdefault("reviewed", {})[candidate_path] = "kept"
            result_message = "Clip kept for later; active references unchanged."
        else:
            slots = manifest.setdefault("slots", {})
            active_items = reference_items(person_id)
            occupied = {item["slot"] for item in active_items}
            if len(occupied) < REFERENCE_COUNT:
                slot = next(f"ref_{index:02d}.wav" for index in range(1, REFERENCE_COUNT + 1)
                            if f"ref_{index:02d}.wav" not in occupied)
            else:
                weakest = min(active_items, key=lambda item: (
                    active_reference_score(reference_dir / item["name"], manifest, person_id), item["slot"]))
                slot = weakest["slot"]
            target = reference_dir / slot
            backup = target.read_bytes() if target.is_file() else None
            staged_path = stage_reference_copy(audio_path, person_id)
            try:
                staged_path.replace(target)
                slots[slot] = {**candidate, "managed_path": rel(target)}
                manifest.setdefault("promoted", []).append(candidate_path)
                manifest.setdefault("reviewed", {})[candidate_path] = "promoted"
                save_reference_manifest(manifest, person_id)
            except Exception:
                staged_path.unlink(missing_ok=True)
                if backup is not None:
                    target.write_bytes(backup)
                else:
                    target.unlink(missing_ok=True)
                raise
            result_message = f"Promoted into {slot}, replacing the weakest active reference."
            promoted_slot = slot

        if action != "promote":
            manifest.setdefault("reviewed", {})[candidate_path] = action
            save_reference_manifest(manifest, person_id)
        next_item = next_reference_candidate(manifest)
        result = {"message": result_message, **next_item, "references": reference_items(person_id)}
        if action == "promote":
            result["promoted_slot"] = promoted_slot
        return result


def history(limit=60, approved_only=False):
    items = []
    if OUTPUT.exists():
        for meta_path in OUTPUT.glob("*.json"):
            wav = meta_path.with_suffix(".wav")
            if not wav.exists():
                continue
            try:
                meta = json.loads(meta_path.read_text(encoding="utf-8"))
            except Exception:
                continue
            if approved_only and not meta.get("approved"):
                continue
            items.append((wav.stat().st_mtime, entry_for(wav, meta)))
    items.sort(key=lambda x: -x[0])
    return [e for _, e in items[:limit]]


def set_approved(file, approved):
    wav = safe_path(file, [OUTPUT])
    meta_path = wav.with_suffix(".json")
    meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.exists() else {}
    meta["approved"] = bool(approved)
    meta["approved_at"] = datetime.now().isoformat(timespec="seconds") if approved else None
    meta_path.write_text(json.dumps(meta, indent=2), encoding="utf-8")
    return entry_for(wav, meta)


def discard_take(file):
    if not isinstance(file, str) or not file.lower().endswith(".wav"):
        raise ValueError("Choose a Studio audio take to discard.")
    wav = safe_path(file, [OUTPUT])
    meta_path = wav.with_suffix(".json")
    if not meta_path.is_file():
        raise ValueError("This take has no Studio metadata and cannot be discarded here.")
    try:
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as e:
        raise ValueError("This take's metadata can't be read, so it was not deleted.") from e
    if not isinstance(meta, dict) or (
        meta.get("source") != "studio" and
        not str(meta.get("note", "")).startswith("AI-generated voice (Chatterbox")
    ):
        raise ValueError("Only Chatterbox audio takes with matching project metadata can be discarded here.")
    wav.unlink()
    meta_path.unlink()
    return {"deleted": True, "file": rel(wav)}


# ---------------------------------------------------------------- web server

class Handler(BaseHTTPRequestHandler):
    engine: Engine = None
    server_version = "VoiceStudio/1.0"

    def log_message(self, fmt, *args):  # keep the terminal quiet except for errors
        if args and str(args[1]).startswith(("4", "5")) and "favicon" not in str(args[0]):
            super().log_message(fmt, *args)

    def _json(self, obj, code=200):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def _error(self, msg, code=400):
        self._json({"error": msg}, code)

    def _body(self):
        n = int(self.headers.get("Content-Length") or 0)
        return json.loads(self.rfile.read(n) or b"{}") if n else {}

    def do_GET(self):
        url = urlparse(self.path)
        path = url.path
        try:
            if path in ("/", "/index.html", "/mobile", "/mobile/"):
                body = HTML.read_bytes()
                if path in ("/mobile", "/mobile/"):
                    body = body.replace(
                        b'href="/manifest.webmanifest"',
                        b'href="/mobile-manifest.webmanifest"',
                        1,
                    )
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Cache-Control", "no-store")
                self.end_headers()
                self.wfile.write(body)
            elif path in APP_ASSETS:
                asset, content_type = APP_ASSETS[path]
                body = asset.read_bytes()
                self.send_response(200)
                self.send_header("Content-Type", content_type)
                self.send_header("Content-Length", str(len(body)))
                self.send_header("Cache-Control", "no-cache")
                self.end_headers()
                self.wfile.write(body)
            elif path == "/api/status":
                self._json(self.engine.status())
            elif path == "/api/voices":
                self._json(list_voices())
            elif path == "/api/people":
                registry = people_registry()
                self._json({"people": available_people(), "sources": registry.get("sources", {})})
            elif path == "/api/person/media":
                person_id = parse_qs(url.query).get("person", ["brian"])[0]
                self._json(person_media_status(person_id))
            elif path == "/api/person/reference-audio":
                query = parse_qs(url.query)
                person_id = query.get("person", ["brian"])[0]
                name = query.get("name", [""])[0]
                person = get_person(person_id)
                sample = next(
                    (path for path in person_reference_audio(person) if path.name == name and
                     re.fullmatch(r"voice_ref[1-5]\.wav", name)),
                    None,
                )
                if sample is None:
                    return self._error("That voice sample is not available.", 404)
                self._file(sample)
            elif path == "/api/rebuild/options":
                query = parse_qs(url.query)
                person = get_person(query.get("person", ["brian"])[0])
                paths = person_profile_paths(person, self.engine.kind)
                profile_path, candidate_path = paths["active"], paths["candidate"]
                registry = people_registry()
                source_items = [item for item, _ in profile_sources()]
                selected_sources = registry.get("sources", {}).get(person["id"])
                if selected_sources is None and person["id"] == "brian":
                    selected_sources = [item["path"] for item in source_items]
                elif selected_sources is None:
                    person_sources = [
                        item["path"] for item in source_items
                        if Path(item["path"]).resolve() == person_reference_dir(person).resolve()
                    ]
                    selected_sources = person_sources
                meta_path = profile_path.with_suffix(".json")
                candidate_meta_path = candidate_path.with_suffix(".json")
                profile_meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.is_file() else {}
                candidate_meta = json.loads(candidate_meta_path.read_text(encoding="utf-8")) \
                    if candidate_meta_path.is_file() else {}
                profile_anchor = ""
                if profile_meta.get("anchor"):
                    try:
                        profile_anchor = Path(profile_meta["anchor"]).resolve().relative_to(ROOT.resolve()).as_posix()
                    except ValueError:
                        pass
                self._json({
                    "person": person,
                    "sources": [
                        {**item, "selected": item["path"] in (selected_sources or [])}
                        for item in source_items
                    ],
                    "anchors": [item for item, _ in profile_anchors(selected_sources or [])],
                    "profile": {"path": rel(profile_path), "exists": profile_path.is_file(),
                                "created": profile_meta.get("created"), "clips_used": profile_meta.get("clips_used"),
                                "anchor": profile_anchor},
                    "candidate": {"path": rel(candidate_path), "exists": candidate_path.is_file(),
                                  "created": candidate_meta.get("created"),
                                  "clips_used": candidate_meta.get("clips_used"),
                                  "anchor_agreement": candidate_meta.get("anchor_agreement")},
                })
            elif path == "/api/references":
                query = parse_qs(url.query)
                person_id = query.get("person", ["brian"])[0]
                self._json({"items": reference_items(person_id), "count": REFERENCE_COUNT})
            elif path == "/api/reference-review/audio":
                requested = parse_qs(url.query).get("path", [""])[0]
                person_id = parse_qs(url.query).get("person", ["brian"])[0]
                manifest_path = person_reference_manifest(get_person(person_id))
                if not manifest_path.is_file():
                    return self._error("Start a reference review before previewing clips.", 404)
                manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
                candidate = next((item for item in manifest.get("review_pool", [])
                                  if item.get("path") == requested), None)
                sources = manifest.get("review_sources", [])
                available = {item["path"]: folder for item, folder in profile_sources()}
                if candidate is None or any(source not in available for source in sources):
                    return self._error("That review clip is not available.", 404)
                self._file(safe_path(requested, [available[source] for source in sources]))
            elif path.startswith("/api/compare/"):
                job = self.engine.compare_jobs.get(path.rsplit("/", 1)[-1])
                self._json(self.engine.public_compare(job)) if job else self._error("No such comparison.", 404)
            elif path.startswith("/api/rebuild/"):
                job = self.engine.rebuild_jobs.get(path.rsplit("/", 1)[-1])
                self._json(self.engine.public_rebuild(job)) if job else self._error("No such rebuild job.", 404)
            elif path.startswith("/api/person/dataset/"):
                job = self.engine.dataset_jobs.get(path.rsplit("/", 1)[-1])
                self._json(self.engine.public_dataset(job)) if job else self._error("No such dataset job.", 404)
            elif path == "/api/history":
                q = parse_qs(url.query)
                self._json({"items": history(approved_only=q.get("approved") == ["1"])})
            elif path == "/api/anchor-audio":
                query = parse_qs(url.query)
                requested = query.get("path", [""])[0]
                person = get_person(query.get("person", ["brian"])[0])
                registry = people_registry()
                selected_sources = registry.get("sources", {}).get(person["id"])
                if selected_sources is None and person["id"] == "brian":
                    selected_sources = [item["path"] for item, _ in profile_sources()]
                anchors = {item["path"]: audio_path
                           for item, audio_path in profile_anchors(selected_sources or [])}
                anchor_path = anchors.get(requested)
                if anchor_path is None:
                    return self._error("Choose an available anchor clip.", 404)
                self._file(anchor_path)
            elif path.startswith("/api/jobs/"):
                job = self.engine.jobs.get(path.rsplit("/", 1)[-1])
                self._json(self.engine.public(job)) if job else self._error("No such job.", 404)
            elif path.startswith("/files/"):
                self._file(safe_path(path[len("/files/"):], [OUTPUT, *clip_dirs()]))
            elif path == "/favicon.ico":
                self.send_response(204)
                self.end_headers()
            else:
                self._error("Not found.", 404)
        except ValueError as e:
            self._error(str(e), 404)
        except (BrokenPipeError, ConnectionResetError):
            pass

    def do_POST(self):
        try:
            parsed = urlparse(self.path)
            if parsed.path == "/api/person/upload":
                self._upload_person_file(parsed)
                return
            data = self._body()
            if self.path == "/api/generate":
                voice = data.get("voice") or {}
                if voice.get("type") not in ("profile", "clip") or not voice.get("path"):
                    return self._error("Choose a voice first.")
                job = self.engine.submit(data)
                self._json(self.engine.public(job))
            elif self.path == "/api/people":
                self._json({"person": create_person(data.get("name")), "people": available_people()}, 201)
            elif self.path == "/api/person/reference/delete":
                self._json(delete_uploaded_voice_sample(data.get("person", "brian"), data.get("name")))
            elif self.path == "/api/person/sources":
                person_id = data.get("person", "brian")
                self._json({"sources": save_person_sources(person_id, data.get("sources"))})
            elif self.path == "/api/person/prepare-dataset":
                self._json(self.engine.submit_dataset(data.get("person", "brian")), 202)
            elif self.path == "/api/approve":
                self._json(set_approved(data.get("file", ""), data.get("approved", True)))
            elif self.path == "/api/discard":
                self._json(discard_take(data.get("file", "")))
            elif self.path == "/api/rebuild":
                self._json(self.engine.submit_rebuild(data), 202)
            elif self.path == "/api/reference-review/start":
                self._json(start_reference_review(data.get("sources"), data.get("person", "brian")))
            elif self.path == "/api/reference-review/action":
                self._json(review_reference(data.get("action"), data.get("path"), data.get("sources"),
                                            data.get("person", "brian")))
            elif self.path == "/api/compare":
                self._json(self.engine.submit_compare(data), 202)
            elif self.path == "/api/rebuild/adopt":
                self._json(self.engine.adopt_candidate(data.get("person", "brian")))
            else:
                self._error("Not found.", 404)
        except ValueError as e:
            self._error(str(e))
        except Exception as e:
            traceback.print_exc()
            self._error(f"Something went wrong: {e}", 500)

    def _upload_person_file(self, parsed):
        query = parse_qs(parsed.query)
        person = get_person(query.get("person", [""])[0])
        kind = query.get("kind", [""])[0]
        if any(
            job.get("person") == person["id"] and job["state"] in ("queued", "running")
            for job in self.engine.dataset_jobs.values()
        ):
            return self._error("Wait for video clip preparation to finish before uploading more files.")
        filename = query.get("name", [""])[0].replace("\\", "/").rsplit("/", 1)[-1]
        suffix = Path(filename).suffix.lower()
        stem = re.sub(r"[^A-Za-z0-9_-]+", "_", Path(filename).stem).strip("_")
        if not stem or kind not in {"video", "reference"}:
            return self._error("Choose a valid file and upload type.")
        if kind == "video" and suffix not in VIDEO_UPLOAD_EXTS:
            return self._error("Upload a supported video or audio file.")
        if kind == "reference" and suffix != ".wav":
            return self._error("Voice samples must be WAV files.")
        try:
            length = int(self.headers.get("Content-Length", ""))
        except ValueError:
            return self._error("The upload size is missing or invalid.")
        limit = MAX_VIDEO_UPLOAD_BYTES if kind == "video" else MAX_REFERENCE_UPLOAD_BYTES
        if length <= 0:
            return self._error("The selected file is empty.")
        if length > limit:
            return self._error(
                f"This file exceeds the {limit // (1024 ** 2)} MB upload limit.", 413
            )

        directory = person_video_dir(person) if kind == "video" else person_reference_dir(person)
        directory.mkdir(parents=True, exist_ok=True)
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(dir=directory, suffix=".upload", delete=False) as handle:
                temporary = Path(handle.name)
                remaining = length
                while remaining:
                    chunk = self.rfile.read(min(1024 * 1024, remaining))
                    if not chunk:
                        raise ValueError("The upload ended before the whole file was received.")
                    handle.write(chunk)
                    remaining -= len(chunk)

            if kind == "reference":
                try:
                    duration = sf.info(str(temporary)).duration
                except (OSError, RuntimeError) as e:
                    raise ValueError("The voice sample is not a readable WAV file.") from e
                if not 10 <= duration <= 30:
                    raise ValueError("Voice samples must be 10-30 seconds long.")
                with REFERENCE_LOCK:
                    occupied = {
                        path.name for path in directory.iterdir()
                        if path.is_file() and re.fullmatch(
                            r"(?:voice_ref[1-5]|ref_0[1-6])\.wav", path.name
                        )
                    }
                    if len(occupied) >= 5:
                        raise ValueError("This person already has five voice samples.")
                    slot = next((f"voice_ref{index}.wav" for index in range(1, 6)
                                 if f"voice_ref{index}.wav" not in occupied), None)
                    if slot is None:
                        raise ValueError("This person already has five voice samples.")
                    target = directory / slot
                    temporary.replace(target)
                    registry = people_registry()
                    sources = registry.setdefault("sources", {}).setdefault(person["id"], [])
                    source = rel(directory)
                    if source not in sources:
                        sources.append(source)
                        save_people_registry(registry)
            else:
                target = directory / f"{uuid.uuid4().hex[:8]}_{stem}{suffix}"
                temporary.replace(target)
            temporary = None
            self._json({"file": rel(target), "media": person_media_status(person["id"])}, 201)
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)

    def _file(self, p):
        """Serve a file with byte-range support, so audio players can seek."""
        size = p.stat().st_size
        ctype = mimetypes.guess_type(p.name)[0] or "application/octet-stream"
        rng = self.headers.get("Range")
        start, end = 0, size - 1
        if rng and (m := re.match(r"bytes=(\d*)-(\d*)", rng)):
            if m.group(1):
                start = int(m.group(1))
                end = int(m.group(2)) if m.group(2) else end
            elif m.group(2):
                start = max(0, size - int(m.group(2)))
            end = min(end, size - 1)
            self.send_response(206)
            self.send_header("Content-Range", f"bytes {start}-{end}/{size}")
        else:
            self.send_response(200)
        length = end - start + 1
        self.send_header("Content-Type", ctype)
        self.send_header("Accept-Ranges", "bytes")
        self.send_header("Content-Length", str(length))
        self.send_header("Cache-Control", "no-store")
        if parse_qs(urlparse(self.path).query).get("download"):
            self.send_header("Content-Disposition", f'attachment; filename="{p.name}"')
        self.end_headers()
        with open(p, "rb") as f:
            f.seek(start)
            remaining = length
            while remaining > 0:
                chunk = f.read(min(65536, remaining))
                if not chunk:
                    break
                self.wfile.write(chunk)
                remaining -= len(chunk)


def main():
    p = argparse.ArgumentParser(description="Local voice cloning and generation studio.")
    p.add_argument("--model", choices=["standard", "turbo"], default="standard")
    p.add_argument("--device", default="auto")
    p.add_argument("--host", default="127.0.0.1")
    p.add_argument("--port", type=int, default=7860)
    p.add_argument("--no-browser", action="store_true")
    args = p.parse_args()

    if not HTML.exists():
        raise SystemExit(f"studio.html is missing. Put it next to studio.py ({HTML.parent}).")
    device = speak.pick_device(args.device)
    Handler.engine = Engine(args.model, device)
    server = ThreadingHTTPServer((args.host, args.port), Handler)
    url = f"http://{'127.0.0.1' if args.host in ('0.0.0.0', '') else args.host}:{args.port}"
    print(f"\nVoice Studio running at {url}")
    if args.host not in ("127.0.0.1", "localhost"):
        print("Others on your network can open it at http://<this computer's IP>:"
              f"{args.port}. Anyone who can reach it can generate audio, so only do this on a trusted network.")
    print("The voice model is loading in the background. Press Ctrl+C to stop.\n")
    if not args.no_browser:
        threading.Timer(1.0, lambda: webbrowser.open(url)).start()
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopped.")


if __name__ == "__main__":
    main()