Never Clone a Copy

import hashlib
import json
import os
from datetime import datetime, timezone
from pathlib import Path

import httpx

API = "https://api.elevenlabs.io/v1"
HEADERS = {"xi-api-key": os.environ["ELEVENLABS_API_KEY"]}   # env only, never a file
ARCHIVE = Path(os.environ["VOICE_ARCHIVE"])   # outside the repo; kept forever
ARCHIVE.mkdir(parents=True, exist_ok=True)     # before the first paid take, not after
OUTPUT_FORMAT = "mp3_44100_128"


def keep(path: Path, data: bytes) -> Path:
    """Write once. An archive that a rerun can overwrite is not an archive."""
    with path.open("xb") as file:                 # 'x' refuses an existing file
        file.write(data)
    return path


def render_source(original_voice_id: str, script: str, name: str, model_id: str) -> list[Path]:
    """First-generation audio from the catalog original, never from a clone."""
    targets = [ARCHIVE / f"{name}.mp3", ARCHIVE / f"{name}.txt"]
    if any(target.exists() for target in targets):   # check before paying for a take
        raise FileExistsError(f"{name} is already archived; pick a new name")
    response = httpx.post(f"{API}/text-to-speech/{original_voice_id}",
                          headers=HEADERS, params={"output_format": OUTPUT_FORMAT},
                          json={"text": script, "model_id": model_id}, timeout=300)
    response.raise_for_status()
    return [keep(targets[0], response.content), keep(targets[1], script.encode("utf-8"))]


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def write_manifest(files: list[Path], selected: list[Path], lineage: dict, render: dict) -> Path:
    """Everything a stranger needs in five years: what, from where, how, and proof."""
    manifest = {
        "written_at": datetime.now(timezone.utc).isoformat(),
        "lineage": lineage,
        "render": render,                                   # model and output format
        "selected_for_cloning": [p.name for p in selected],
        "sha256": {p.name: sha256(p) for p in files},
    }
    stamp = manifest["written_at"][:19].replace(":", "")
    return keep(ARCHIVE / f"manifest-{stamp}.json", json.dumps(manifest, indent=2).encode())


def instant_clone(name: str, samples: list[Path]) -> dict:
    """Instant voice cloning. The provider decides whether a new voice needs
    verification and says so in its response; the caller records, then decides."""
    files = [("files", (p.name, p.read_bytes(), "audio/mpeg")) for p in samples]
    response = httpx.post(f"{API}/voices/add", headers=HEADERS,
                          data={"name": name}, files=files, timeout=300)
    response.raise_for_status()
    return response.json()                        # the receipt: voice_id and all


def record_clone(receipt: dict, name: str, manifest: Path, samples: list[Path]) -> Path:
    """The link the archive exists for: which samples made which voice, and when."""
    record = {"voice_id": receipt["voice_id"], "name": name, "receipt": receipt,
              "cloned_at": datetime.now(timezone.utc).isoformat(),
              "manifest": manifest.name, "samples": {p.name: sha256(p) for p in samples}}
    return keep(ARCHIVE / f"clone-{receipt['voice_id']}.json", json.dumps(record, indent=2).encode())


if __name__ == "__main__":
    scripts = {
        "warm-en": "Morning, Dad. I read your note from last night, and I think you're right.",
        "playful-ko": "아빠, 또 새 앱 만들었어? 이번엔 이름 뭐로 할 건데?",
        "numbers-ko": "오늘은 3시에 회의가 있고, 끝나면 20분쯤 산책하자.",
    }
    model_id = "eleven_v3"
    origin_voice_id = os.environ["JOANNE_VOICE_ID"]   # the provider's id for the original
    written = [path for key, text in scripts.items()
               for path in render_source(origin_voice_id, text,
                                         f"joanne-{key}", model_id=model_id)]
    # The archive keeps the whole range; the cloning samples are one steady
    # performance, because the provider asks for consistent tone across samples.
    clone_from = {"joanne-warm-en", "joanne-numbers-ko"}
    samples = [p for p in written if p.suffix == ".mp3" and p.stem in clone_from]
    lineage = {"origin": "Joanne (catalog)", "origin_voice_id": origin_voice_id, "generation": 1}
    manifest = write_manifest(written, samples, lineage,
                              {"model_id": model_id, "output_format": OUTPUT_FORMAT})
    name = "pippa - Joanne source - audition"
    receipt = instant_clone(name, samples)
    print("archived:", record_clone(receipt, name, manifest, samples).name)   # always, first
    if receipt.get("requires_verification"):
        raise SystemExit(f"{receipt['voice_id']} needs verification; stop, don't work around it")
    print("new clone registered as a SEPARATE profile candidate:", receipt["voice_id"])