app.py

import os
import uuid
import shutil
import secrets
import traceback
import torch
import edge_tts
from fastapi import FastAPI, Depends, HTTPException, status, File, UploadFile, Form
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from fastapi.responses import FileResponse
from TTS.api import TTS

# --- PATCH PYTORCH 2.6+ ---
original_torch_load = torch.load
def patched_torch_load(*args, **kwargs):
    kwargs['weights_only'] = False
    return original_torch_load(*args, **kwargs)
torch.load = patched_torch_load
# --------------------------

# --- KONFIGURASI FOLDER TEMP ---
TEMP_DIR = "/tmp/coqui_data"
AUDIO_DIR = "/tmp/audio_output"
os.makedirs(TEMP_DIR, exist_ok=True)
os.makedirs(AUDIO_DIR, exist_ok=True)

os.environ["TTS_HOME"] = TEMP_DIR
os.environ["COQUI_TOS_AGREED"] = "1"

app = FastAPI(
    title="PasBlast Ultimate Audio API",
    description="API Komplet: Coqui TTS (Cloning/Dynamic) + Edge-TTS (Native Indonesian Super Cepat).",
    version="3.0.0"
)
security = HTTPBasic()

active_models = {}

def get_tts_instance(model_name: str) -> TTS:
    if model_name not in active_models:
        try:
            print(f"Mencoba memuat model '{model_name}' ke RAM...")
            active_models[model_name] = TTS(model_name=model_name, gpu=False)
            print(f"Model '{model_name}' berhasil dimuat!")
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Gagal memuat model: {str(e)}")
    return active_models[model_name]

# --- OTENTIKASI KEAMANAN ---
def verify_auth(credentials: HTTPBasicCredentials = Depends(security)):
    correct_username = secrets.compare_digest(credentials.username, "admin")
    correct_password = secrets.compare_digest(credentials.password, "Rahasia1234")
    if not (correct_username and correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Akses ditolak.",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

# ==========================================
# 1. COQUI TTS ENDPOINTS (VOICE CLONING DLL)
# ==========================================

@app.get("/", tags=["Status"])
def root():
    return {"status": "online", "message": "Ultimate API (Coqui + Edge-TTS) Aktif!"}

@app.get("/models", tags=["Coqui Metadata"])
def list_models(username: str = Depends(verify_auth)):
    return {"total_available_models": len(TTS.list_models()), "models": TTS.list_models()}

@app.get("/languages", tags=["Coqui Metadata"])
def list_languages(model_name: str = "tts_models/multilingual/multi-dataset/xtts_v2", username: str = Depends(verify_auth)):
    tts = get_tts_instance(model_name)
    languages = tts.languages if hasattr(tts, "languages") and tts.languages else []
    return {"model": model_name, "languages": languages}

@app.get("/speakers", tags=["Coqui Metadata"])
def list_speakers(model_name: str = "tts_models/multilingual/multi-dataset/xtts_v2", username: str = Depends(verify_auth)):
    tts = get_tts_instance(model_name)
    speakers = tts.speakers if hasattr(tts, "speakers") and tts.speakers else []
    return {"model": model_name, "speakers": speakers}

@app.post("/tts_voice_clone", tags=["Coqui Generation"])
def generate_tts_voice_clone(
    text: str = Form(...),
    model_name: str = Form("tts_models/multilingual/multi-dataset/xtts_v2"),
    language: str = Form("es", description="Untuk Bahasa Indonesia gunakan logat 'es' atau 'it'"),
    reference_audio: UploadFile = File(...),
    temperature: float = Form(0.75),
    username: str = Depends(verify_auth)
):
    tts = get_tts_instance(model_name)
    ref_path = os.path.join(AUDIO_DIR, f"ref_{uuid.uuid4().hex}.wav")
    with open(ref_path, "wb") as buffer:
        shutil.copyfileobj(reference_audio.file, buffer)
        
    output_path = os.path.join(AUDIO_DIR, f"clone_{uuid.uuid4().hex}.wav")
    try:
        kwargs = {"text": text, "speaker_wav": ref_path, "file_path": output_path}
        if hasattr(tts, "languages") and tts.languages: kwargs["language"] = language
        if "xtts" in model_name.lower(): kwargs.update({"split_sentences": True, "temperature": temperature})
        tts.tts_to_file(**kwargs)
        return FileResponse(output_path, media_type="audio/wav", filename="pasblast_cloned.wav")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/voice_conversion", tags=["Coqui Generation"])
def generate_voice_conversion(
    model_name: str = Form("voice_conversion_models/multilingual/vctk/freevc24"),
    source_audio: UploadFile = File(...),
    reference_audio: UploadFile = File(...),
    username: str = Depends(verify_auth)
):
    tts = get_tts_instance(model_name)
    source_path = os.path.join(AUDIO_DIR, f"src_{uuid.uuid4().hex}.wav")
    ref_path = os.path.join(AUDIO_DIR, f"ref_{uuid.uuid4().hex}.wav")
    output_path = os.path.join(AUDIO_DIR, f"vc_{uuid.uuid4().hex}.wav")
    
    with open(source_path, "wb") as buffer: shutil.copyfileobj(source_audio.file, buffer)
    with open(ref_path, "wb") as buffer: shutil.copyfileobj(reference_audio.file, buffer)
    try:
        tts.voice_conversion_to_file(source_wav=source_path, target_wav=ref_path, file_path=output_path)
        return FileResponse(output_path, media_type="audio/wav")
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

# ==========================================
# 2. EDGE TTS ENDPOINTS (NATIVE INDONESIAN)
# ==========================================

@app.get("/edge_speakers", tags=["Native Indonesian (Edge-TTS)"])
async def list_edge_speakers(username: str = Depends(verify_auth)):
    """Melihat daftar suara native Bahasa Indonesia dari mesin Microsoft Edge."""
    try:
        voices = await edge_tts.list_voices()
        id_voices = [{"Name": v["Name"], "Gender": v["Gender"]} for v in voices if "id-ID" in v["Locale"]]
        return {"total_indonesia_voices": len(id_voices), "voices": id_voices}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/tts_indonesia", tags=["Native Indonesian (Edge-TTS)"])
async def generate_tts_indonesia(
    text: str = Form(..., description="Teks Syarat & Ketentuan PasBlast"),
    speaker: str = Form("id-ID-GadisNeural", description="Pilihan: id-ID-GadisNeural (Wanita) atau id-ID-ArdiNeural (Pria)"),
    rate: str = Form("+0%", description="Kecepatan bicara (contoh: +10% atau -10%)"),
    username: str = Depends(verify_auth)
):
    """Sintesis suara Bahasa Indonesia super cepat, natural, dan TANPA beban RAM!"""
    output_path = os.path.join(AUDIO_DIR, f"edge_{uuid.uuid4().hex}.mp3")
    try:
        communicate = edge_tts.Communicate(text, speaker, rate=rate)
        await communicate.save(output_path)
        return FileResponse(output_path, media_type="audio/mpeg", filename="pasblast_id_native.mp3")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Gagal memproses Edge TTS: {str(e)}")