""" cantrell-kokoro-engine — StoryVoice Cloning Backend Docker Space, Python 3.12, FastAPI only KokoClone zero-shot voice cloning via Kanade voice conversion """ import os import sys from fastapi import FastAPI, HTTPException from fastapi.responses import Response from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import uvicorn app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ── KokoClone voice cloning (lazy-loaded) ──────────────────────────────────── BASE_DIR = os.path.dirname(os.path.abspath(__file__)) VOICES_DIR = os.path.join(BASE_DIR, "voices") sys.path.insert(0, BASE_DIR) _kokoclone = None def get_kokoclone(): global _kokoclone if _kokoclone is None: from core.cloner import KokoClone print("[StoryVoice] Loading KokoClone...") _kokoclone = KokoClone() print("[StoryVoice] KokoClone ready.") return _kokoclone def find_voice_audio(voice_id: str): """Find a reference MP3/WAV in the Space's voices/ folder.""" voice_id_clean = voice_id.replace(".mp3", "").replace(".wav", "").strip() for ext in (".mp3", ".wav"): p = os.path.join(VOICES_DIR, f"{voice_id_clean}{ext}") if os.path.exists(p): return p if os.path.isdir(VOICES_DIR): for f in os.listdir(VOICES_DIR): name, ext = os.path.splitext(f) if ext.lower() in (".mp3", ".wav") and name.lower() == voice_id_clean.lower(): return os.path.join(VOICES_DIR, f) return None # ── Routes ──────────────────────────────────────────────────────────────────── @app.get("/") def index(): html = """ Nyako StoryVoice™

Nyako StoryVoice™

Voice Clone — Cantrell Creatives
""" return Response(content=html, media_type="text/html") @app.get("/health") def health(): return {"status": "ok", "engine": "kokoclone"} @app.get("/my-voices") def my_voices(): """Returns list of custom voice MP3s available for cloning.""" found = [] if os.path.isdir(VOICES_DIR): for f in sorted(os.listdir(VOICES_DIR)): name, ext = os.path.splitext(f) if ext.lower() in (".mp3", ".wav"): found.append({"voice_id": name, "display_name": name, "file": f}) return found # ── KokoClone endpoints ─────────────────────────────────────────────────────── class CloneRequest(BaseModel): text: str voice_id: str # name of your reference MP3 in the voices/ folder speed: float = 1.0 lang: str = "en" @app.post("/clone") def clone(req: CloneRequest): """Generate speech cloned to match a reference voice MP3.""" if not req.text.strip(): raise HTTPException(status_code=400, detail="text is required") ref_path = find_voice_audio(req.voice_id) if not ref_path: raise HTTPException(status_code=404, detail=f"Voice reference not found: {req.voice_id}") try: import tempfile cloner = get_kokoclone() with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: out_path = tmp.name cloner.generate( text=req.text.strip(), lang=req.lang, reference_audio=ref_path, output_path=out_path ) with open(out_path, "rb") as f: audio_bytes = f.read() os.remove(out_path) return Response(content=audio_bytes, media_type="audio/wav") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) class ConvertRequest(BaseModel): voice_id: str # target reference voice MP3 source_audio_b64: str # base64-encoded source WAV/MP3 @app.post("/convert") def convert(req: ConvertRequest): """Re-voice existing audio to match a reference voice MP3.""" import base64, tempfile ref_path = find_voice_audio(req.voice_id) if not ref_path: raise HTTPException(status_code=404, detail=f"Voice reference not found: {req.voice_id}") try: cloner = get_kokoclone() src_bytes = base64.b64decode(req.source_audio_b64) with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as src_tmp: src_tmp.write(src_bytes) src_path = src_tmp.name with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as out_tmp: out_path = out_tmp.name cloner.convert( source_audio=src_path, reference_audio=ref_path, output_path=out_path ) with open(out_path, "rb") as f: audio_bytes = f.read() os.remove(src_path) os.remove(out_path) return Response(content=audio_bytes, media_type="audio/wav") except Exception as e: raise HTTPException(status_code=500, detail=str(e)) if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)