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)}")