Update app.py

app.py

@@ -50,6 +50,8 @@ tts = None
 model_loaded = False
 current_model = ""
 model_loading = False
+current_voice_style = "default_female"
+voice_cloning_supported = False
 app_startup_time = datetime.now()
 
 # Pydantic models
@@ -67,6 +69,14 @@ class BatchTTSRequest(BaseModel):
     language: Optional[str] = "en"
     voice_style: Optional[str] = "default_female"
 
+class VoiceCloneRequest(BaseModel):
+    project_id: str
+    voice_name: str
+    description: Optional[str] = ""
+
+class ChangeVoiceRequest(BaseModel):
+    voice_style: str
+
 # Helper functions
 def clean_text(text):
     """Clean text for TTS generation"""
@@ -182,6 +192,42 @@ def upload_to_oci_with_retry(file_path: str, filename: str, project_id: str, fil
     
     return None, "Upload failed: unexpected error"
 
+def get_voice_path(voice_name: str):
+    """Get path to voice file"""
+    if voice_name == "default":
+        return None
+    
+    voice_path = Path(f"/tmp/voices/{voice_name}")
+    if voice_path.is_dir():
+        samples = list(voice_path.glob("sample_*.wav"))
+        return str(samples[0]) if samples else None
+    else:
+        voice_file = Path(f"/tmp/voices/{voice_name}.wav")
+        return str(voice_file) if voice_file.exists() else None
+
+def clone_voice(voice_name: str, audio_files: List[str], description: str = ""):
+    """Clone a voice from audio samples"""
+    try:
+        print(f"🎙️ Cloning voice: {voice_name}")
+        
+        voice_dir = f"/tmp/voices/{voice_name}"
+        os.makedirs(voice_dir, exist_ok=True)
+        
+        for i, audio_file in enumerate(audio_files):
+            dest_path = f"{voice_dir}/sample_{i+1}.wav"
+            shutil.copy2(audio_file, dest_path)
+            print(f"   Copied sample {i+1} to: {dest_path}")
+        
+        print(f"✅ Voice cloning setup completed for {voice_name}")
+        return True, f"Voice {voice_name} is ready for use"
+        
+    except Exception as e:
+        return False, f"Voice cloning failed: {str(e)}"
+
+def supports_voice_cloning():
+    """Check if the current model supports voice cloning"""
+    return "xtts" in current_model.lower()
+
 def save_wav(audio, file_path):
     """Save audio to WAV file manually"""
     try:
@@ -216,14 +262,14 @@ def save_wav(audio, file_path):
 
 def load_tts_model(voice_style="default_female"):
     """Load TTS model with different voice options - LAZY LOADING"""
-    global tts, model_loaded, current_model, model_loading
+    global tts, model_loaded, current_model, model_loading, current_voice_style, voice_cloning_supported
     
     if model_loading:
         print("⏳ Model is already being loaded...")
         return False
     
-    if model_loaded:
-        print("✅ Model already loaded")
+    if model_loaded and current_voice_style == voice_style:
+        print("✅ Model already loaded with requested voice style")
         return True
     
     model_loading = True
@@ -231,19 +277,40 @@ def load_tts_model(voice_style="default_female"):
     try:
         from TTS.api import TTS
         
-        # Use only fast, lightweight models
+        # Use only fast, lightweight models with proper voice styles
         model_options = {
             "default_female": {
                 "name": "tts_models/en/ljspeech/tacotron2-DDC",
-                "description": "Tacotron2 - Default female (fast)",
+                "description": "Tacotron2 - Default female voice",
             },
             "clear_male": {
                 "name": "tts_models/en/ek1/tacotron2", 
                 "description": "Tacotron2 - Clear male voice",
+            },
+            "male_deep": {
+                "name": "tts_models/en/ek1/tacotron2",
+                "description": "Tacotron2 - Deep male voice",
+            },
+            "male_medium": {
+                "name": "tts_models/en/ljspeech/glow-tts",
+                "description": "Glow-TTS - Medium male voice",
+            },
+            "female_1": {
+                "name": "tts_models/en/ljspeech/tacotron2-DDC",
+                "description": "Tacotron2 - Female voice 1",
+            },
+            "female_2": {
+                "name": "tts_models/en/ljspeech/glow-tts",
+                "description": "Glow-TTS - Female voice 2",
+            },
+            "voice_clone": {
+                "name": "tts_models/multilingual/multi-dataset/xtts_v2",
+                "description": "XTTS v2 - Voice cloning supported",
             }
         }
         
         selected_model = model_options.get(voice_style, model_options["default_female"])
+        current_voice_style = voice_style
         
         print(f"🚀 Loading {selected_model['description']}...")
         print("📥 Downloading model (this may take a few minutes on first load)...")
@@ -263,9 +330,11 @@ def load_tts_model(voice_style="default_female"):
             tts = TTS("tts_models/en/ljspeech/tacotron2-DDC").to(DEVICE)
             tts.tts(text="Hello")
             selected_model = model_options["default_female"]
+            current_voice_style = "default_female"
         
         model_loaded = True
         current_model = selected_model["name"]
+        voice_cloning_supported = supports_voice_cloning()
         return True
             
     except Exception as e:
@@ -294,6 +363,8 @@ async def api_health_check():
         "status": "healthy",
         "model_loaded": model_loaded,
         "current_model": current_model if model_loaded else "none",
+        "current_voice_style": current_voice_style,
+        "voice_cloning_supported": voice_cloning_supported,
         "device": DEVICE,
         "uptime": str(datetime.now() - app_startup_time),
         "timestamp": datetime.now().isoformat()
@@ -336,8 +407,8 @@ async def check_oci_health():
 async def generate_tts(request: TTSRequest):
     """Generate TTS for a single text with lazy model loading"""
     try:
-        # Lazy load model on first request
-        if not model_loaded:
+        # Lazy load model on first request or when voice style changes
+        if not model_loaded or current_voice_style != request.voice_style:
             print("🔄 Lazy loading TTS model...")
             if not load_tts_model(request.voice_style):
                 return {
@@ -350,6 +421,15 @@ async def generate_tts(request: TTSRequest):
         print(f"📥 TTS request for project: {request.project_id}")
         print(f"   Text length: {len(request.text)} characters")
         print(f"   Voice style: {request.voice_style}")
+        print(f"   Voice name: {request.voice_name}")
+        
+        # Check if voice cloning is requested but not supported
+        if request.voice_name != "default" and not voice_cloning_supported:
+            return {
+                "status": "error",
+                "message": "Voice cloning is not supported with the current model. Please use 'voice_clone' voice style for cloning.",
+                "model": current_model
+            }
         
         # Generate unique filename
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -359,6 +439,17 @@ async def generate_tts(request: TTSRequest):
         # Ensure output directory exists
         os.makedirs(os.path.dirname(output_path), exist_ok=True)
         
+        # Get voice path if custom voice is requested
+        speaker_wav = None
+        if request.voice_name != "default":
+            speaker_wav = get_voice_path(request.voice_name)
+            if not speaker_wav:
+                return {
+                    "status": "error",
+                    "message": f"Voice '{request.voice_name}' not found. Please clone the voice first using /api/clone-voice."
+                }
+            print(f"🎙️ Using cloned voice: {request.voice_name}")
+        
         print(f"🔊 Generating TTS to: {output_path}")
 
         # Clean the text before generation
@@ -369,18 +460,29 @@ async def generate_tts(request: TTSRequest):
         try:
             print(f"🔊 Generating TTS with {current_model}...")
             
-            # Simple TTS generation for fast models
-            tts.tts_to_file(
-                text=cleaned_text,
-                file_path=output_path
-            )
+            if speaker_wav and voice_cloning_supported:
+                # Use voice cloning
+                tts.tts_to_file(
+                    text=cleaned_text,
+                    file_path=output_path,
+                    speaker_wav=speaker_wav
+                )
+            else:
+                # Simple TTS generation for fast models
+                tts.tts_to_file(
+                    text=cleaned_text,
+                    file_path=output_path
+                )
                 
         except Exception as tts_error:
             print(f"❌ TTS generation failed: {tts_error}")
             # Try alternative approach
             try:
                 print("🔄 Trying alternative TTS generation method...")
-                audio = tts.tts(text=cleaned_text)
+                if speaker_wav and voice_cloning_supported:
+                    audio = tts.tts(text=cleaned_text, speaker_wav=speaker_wav)
+                else:
+                    audio = tts.tts(text=cleaned_text)
                 
                 # Save manually
                 if not save_wav(audio, output_path):
@@ -412,7 +514,9 @@ async def generate_tts(request: TTSRequest):
                 "filename": filename,
                 "file_size": file_size,
                 "voice_style": request.voice_style,
+                "voice_name": request.voice_name,
                 "model_used": current_model,
+                "voice_cloning_used": speaker_wav is not None,
                 "oci_upload_error": error
             }
         
@@ -431,19 +535,133 @@ async def generate_tts(request: TTSRequest):
             "filename": filename,
             "oci_path": upload_result.get("path", f"{request.project_id}/voiceover/{filename}"),
             "model_used": current_model,
-            "voice_style": request.voice_style
+            "voice_style": request.voice_style,
+            "voice_name": request.voice_name,
+            "voice_cloning_used": speaker_wav is not None
         }
         
     except Exception as e:
         print(f"❌ TTS generation error: {str(e)}")
         raise HTTPException(status_code=500, detail=f"TTS generation failed: {str(e)}")
 
+@app.post("/api/clone-voice")
+async def clone_voice_endpoint(
+    project_id: str = Form(...),
+    voice_name: str = Form(...),
+    description: str = Form(""),
+    files: List[UploadFile] = File(...)
+):
+    """Clone a voice from uploaded audio samples"""
+    try:
+        if not files:
+            raise HTTPException(status_code=400, detail="No audio files provided")
+        
+        # Check if we have at least one file
+        if len(files) == 0:
+            raise HTTPException(status_code=400, detail="At least one audio file is required")
+        
+        print(f"🎙️ Starting voice cloning for: {voice_name}")
+        print(f"   Project ID: {project_id}")
+        print(f"   Number of samples: {len(files)}")
+        
+        # Save uploaded files temporarily
+        temp_files = []
+        for file in files:
+            if not file.filename.lower().endswith(('.wav', '.mp3', '.flac')):
+                raise HTTPException(status_code=400, detail="Only WAV, MP3, and FLAC files are supported")
+            
+            temp_path = f"/tmp/{uuid.uuid4()}_{file.filename}"
+            with open(temp_path, "wb") as f:
+                shutil.copyfileobj(file.file, f)
+            temp_files.append(temp_path)
+            print(f"   Saved sample: {file.filename}")
+        
+        # Clone voice
+        success, message = clone_voice(voice_name, temp_files, description)
+        
+        # Clean up temp files
+        for temp_file in temp_files:
+            try:
+                os.remove(temp_file)
+            except:
+                pass
+        
+        if success:
+            return {
+                "status": "success",
+                "message": message,
+                "voice_name": voice_name,
+                "samples_used": len(temp_files),
+                "project_id": project_id
+            }
+        else:
+            raise HTTPException(status_code=500, detail=message)
+            
+    except HTTPException:
+        raise
+    except Exception as e:
+        print(f"❌ Voice cloning error: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Voice cloning failed: {str(e)}")
+
+@app.get("/api/voices")
+async def list_voices():
+    """List all available cloned voices"""
+    try:
+        voices_dir = Path("/tmp/voices")
+        if not voices_dir.exists():
+            return {"voices": []}
+        
+        voices = []
+        for voice_dir in voices_dir.iterdir():
+            if voice_dir.is_dir():
+                samples = list(voice_dir.glob("sample_*.wav"))
+                voices.append({
+                    "name": voice_dir.name,
+                    "samples_count": len(samples),
+                    "samples": [str(sample.name) for sample in samples],
+                    "created_at": datetime.fromtimestamp(voice_dir.stat().st_mtime).isoformat()
+                })
+        
+        return {"voices": voices}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Failed to list voices: {str(e)}")
+
+@app.post("/api/change-voice")
+async def change_voice_style(request: ChangeVoiceRequest):
+    """Change the voice style (reloads model)"""
+    try:
+        global model_loaded
+        
+        print(f"🔄 Changing voice style to: {request.voice_style}")
+        
+        # Reset model loaded flag to force reload
+        model_loaded = False
+        
+        if load_tts_model(request.voice_style):
+            return {
+                "status": "success",
+                "message": f"Voice style changed to {request.voice_style}",
+                "current_voice_style": current_voice_style,
+                "current_model": current_model,
+                "voice_cloning_supported": voice_cloning_supported
+            }
+        else:
+            raise HTTPException(status_code=500, detail="Failed to load new voice style")
+            
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
 @app.get("/api/voice-styles")
 async def get_voice_styles():
     """Get available voice styles"""
     styles = {
-        "default_female": "Default female voice (Tacotron2) - Fast",
-        "clear_male": "Clear male voice (Tacotron2) - Fast"
+        "default_female": "Default female voice (Tacotron2)",
+        "clear_male": "Clear male voice (Tacotron2)",
+        "male_deep": "Deep male voice (Tacotron2)",
+        "male_medium": "Medium male voice (Glow-TTS)",
+        "female_1": "Female voice 1 (Tacotron2)",
+        "female_2": "Female voice 2 (Glow-TTS)",
+        "voice_clone": "XTTS v2 - Voice cloning supported (requires voice samples)"
     }
     return {"voice_styles": styles}
 
@@ -454,6 +672,8 @@ async def get_status():
         "status": "running",
         "model_loaded": model_loaded,
         "current_model": current_model if model_loaded else "none",
+        "current_voice_style": current_voice_style,
+        "voice_cloning_supported": voice_cloning_supported,
         "device": DEVICE,
         "oci_configured": bool(OCI_UPLOAD_API_URL),
         "startup_time": app_startup_time.isoformat(),