Update app.py

app.py

@@ -19,7 +19,12 @@ os.makedirs("/tmp/voices", exist_ok=True)
 os.makedirs("/tmp/output", exist_ok=True)
 
 # Initialize FastAPI app
-app = FastAPI(title="TTS API", description="API for text-to-speech with Coqui TTS and voice cloning")
+app = FastAPI(
+    title="TTS API", 
+    description="API for text-to-speech with Coqui TTS",
+    docs_url="/",
+    redoc_url=None
+)
 
 # Add CORS middleware
 app.add_middleware(
@@ -30,10 +35,9 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
-# Configuration - FIXED OCI URL HANDLING
+# Configuration
 OCI_UPLOAD_API_URL = os.getenv("OCI_UPLOAD_API_URL", "").strip()
 if OCI_UPLOAD_API_URL:
-    # Remove trailing slash if present
     OCI_UPLOAD_API_URL = OCI_UPLOAD_API_URL.rstrip('/')
 
 DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
@@ -45,10 +49,7 @@ print(f"🔧 OCI Upload URL: {OCI_UPLOAD_API_URL or 'Not configured - uploads wi
 tts = None
 model_loaded = False
 current_model = ""
-voice_cloning_supported = False
 model_loading = False
-model_load_attempts = 0
-current_voice_style = "default_female"
 app_startup_time = datetime.now()
 
 # Pydantic models
@@ -66,14 +67,6 @@ 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"""
@@ -106,12 +99,10 @@ def clean_text(text):
 def upload_to_oci(file_path: str, filename: str, project_id: str, file_type="voiceover"):
     """Upload file to OCI using your existing API with subfolder support"""
     try:
-        # Check if OCI URL is properly configured
         if not OCI_UPLOAD_API_URL:
             print("⚠️ OCI upload skipped - OCI_UPLOAD_API_URL not configured")
             return {"status": "skipped", "message": "OCI upload disabled"}, None
         
-        # Validate URL format
         if not OCI_UPLOAD_API_URL.startswith(('http://', 'https://')):
             return None, f"Invalid OCI URL format: {OCI_UPLOAD_API_URL}"
         
@@ -157,24 +148,12 @@ def upload_to_oci(file_path: str, filename: str, project_id: str, file_type="voi
                 print(f"❌ {error_msg}")
                 return None, error_msg
                 
-    except requests.exceptions.Timeout:
-        error_msg = "OCI upload timeout - server took too long to respond"
-        print(f"❌ {error_msg}")
-        return None, error_msg
-    except requests.exceptions.ConnectionError:
-        error_msg = f"Cannot connect to OCI API at {OCI_UPLOAD_API_URL} - check if the server is running and accessible"
-        print(f"❌ {error_msg}")
-        return None, error_msg
-    except requests.exceptions.RequestException as e:
-        error_msg = f"Network error during upload: {str(e)}"
-        print(f"❌ {error_msg}")
-        return None, error_msg
     except Exception as e:
-        error_msg = f"Unexpected upload error: {str(e)}"
+        error_msg = f"Upload error: {str(e)}"
         print(f"❌ {error_msg}")
         return None, error_msg
 
-def upload_to_oci_with_retry(file_path: str, filename: str, project_id: str, file_type="voiceover", max_retries=3):
+def upload_to_oci_with_retry(file_path: str, filename: str, project_id: str, file_type="voiceover", max_retries=2):
     """Upload file to OCI with retry logic"""
     for attempt in range(max_retries):
         try:
@@ -203,42 +182,6 @@ 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:
@@ -273,62 +216,37 @@ 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, voice_cloning_supported, model_loading, model_load_attempts, current_voice_style
+    global tts, model_loaded, current_model, model_loading
     
     if model_loading:
         print("⏳ Model is already being loaded...")
         return False
     
-    if model_loaded and current_voice_style == voice_style:
-        print("✅ Model already loaded with requested voice style")
+    if model_loaded:
+        print("✅ Model already loaded")
         return True
     
     model_loading = True
-    model_load_attempts += 1
     
     try:
         from TTS.api import TTS
         
-        # Use smaller, faster models for initial load
+        # Use only fast, lightweight models
         model_options = {
             "default_female": {
                 "name": "tts_models/en/ljspeech/tacotron2-DDC",
                 "description": "Tacotron2 - Default female (fast)",
-                "speaker": None
             },
             "clear_male": {
                 "name": "tts_models/en/ek1/tacotron2", 
                 "description": "Tacotron2 - Clear male voice",
-                "speaker": None
-            },
-            # Fallbacks for other voice styles
-            "male_deep": {
-                "name": "tts_models/en/ljspeech/tacotron2-DDC",
-                "description": "Tacotron2 - Default female (fallback)",
-                "speaker": None
-            },
-            "male_medium": {
-                "name": "tts_models/en/ljspeech/tacotron2-DDC",
-                "description": "Tacotron2 - Default female (fallback)", 
-                "speaker": None
-            },
-            "female_1": {
-                "name": "tts_models/en/ljspeech/tacotron2-DDC",
-                "description": "Tacotron2 - Default female (fallback)",
-                "speaker": None
-            },
-            "female_2": {
-                "name": "tts_models/en/ljspeech/tacotron2-DDC",
-                "description": "Tacotron2 - Default female (fallback)",
-                "speaker": None
             }
         }
         
         selected_model = model_options.get(voice_style, model_options["default_female"])
-        current_voice_style = voice_style
         
         print(f"🚀 Loading {selected_model['description']}...")
-        print(f"📥 This may take a few minutes on first load...")
+        print("📥 Downloading model (this may take a few minutes on first load)...")
         
         # Load the selected model
         tts = TTS(selected_model["name"]).to(DEVICE)
@@ -348,8 +266,6 @@ def load_tts_model(voice_style="default_female"):
         
         model_loaded = True
         current_model = selected_model["name"]
-        voice_cloning_supported = False
-        model_load_attempts = 0
         return True
             
     except Exception as e:
@@ -360,18 +276,6 @@ def load_tts_model(voice_style="default_female"):
         model_loading = False
 
 # Health check endpoints - CRITICAL FOR DEPLOYMENT
-@app.get("/")
-async def root():
-    """Root endpoint - always responds quickly"""
-    return {
-        "status": "running", 
-        "service": "TTS API",
-        "startup_time": app_startup_time.isoformat(),
-        "model_loaded": model_loaded,
-        "device": DEVICE,
-        "oci_configured": bool(OCI_UPLOAD_API_URL)
-    }
-
 @app.get("/health")
 async def health_check():
     """Health check endpoint - must respond quickly"""
@@ -405,7 +309,6 @@ async def check_oci_health():
         }
     
     try:
-        # Test connection to OCI service with short timeout
         test_url = f"{OCI_UPLOAD_API_URL}/api/health"
         response = requests.get(test_url, timeout=5)
         
@@ -434,7 +337,7 @@ 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 or current_voice_style != request.voice_style:
+        if not model_loaded:
             print("🔄 Lazy loading TTS model...")
             if not load_tts_model(request.voice_style):
                 return {
@@ -448,14 +351,6 @@ async def generate_tts(request: TTSRequest):
         print(f"   Text length: {len(request.text)} characters")
         print(f"   Voice style: {request.voice_style}")
         
-        # Check if voice cloning is requested but not supported
-        if request.voice_name != "default" and not supports_voice_cloning():
-            return {
-                "status": "error",
-                "message": "Voice cloning is not supported with the current model. Only the default voice is available.",
-                "model": current_model
-            }
-        
         # Generate unique filename
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
         filename = f"voiceover_{timestamp}.wav"
@@ -464,16 +359,6 @@ 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"
-                }
-        
         print(f"🔊 Generating TTS to: {output_path}")
 
         # Clean the text before generation
@@ -553,218 +438,12 @@ async def generate_tts(request: TTSRequest):
         print(f"❌ TTS generation error: {str(e)}")
         raise HTTPException(status_code=500, detail=f"TTS generation failed: {str(e)}")
 
-@app.post("/api/batch-tts")
-async def batch_generate_tts(request: BatchTTSRequest):
-    """Generate TTS for multiple texts with sequential naming"""
-    try:
-        # Lazy load model on first request
-        if not model_loaded or current_voice_style != request.voice_style:
-            print("🔄 Lazy loading TTS model for batch processing...")
-            if not load_tts_model(request.voice_style):
-                raise HTTPException(status_code=500, detail="TTS model failed to load")
-        
-        print(f"📥 Batch TTS request for project: {request.project_id}")
-        print(f"   Number of texts: {len(request.texts)}")
-        print(f"   Voice style: {request.voice_style}")
-        
-        # Check if voice cloning is requested but not supported
-        if request.voice_name != "default" and not supports_voice_cloning():
-            raise HTTPException(
-                status_code=400, 
-                detail="Voice cloning is not supported with the current model. Only the default voice is available."
-            )
-        
-        # 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:
-                raise HTTPException(status_code=400, detail=f"Voice '{request.voice_name}' not found")
-        
-        results = []
-        
-        for i, text in enumerate(request.texts):
-            print(f"   Processing text {i+1}/{len(request.texts)}")
-            
-            # Generate sequential filename
-            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
-            filename = f"voiceover_{timestamp}_{i+1:03d}.wav"
-            output_path = f"/tmp/output/{filename}"
-            
-            # Clean text
-            cleaned_text = clean_text(text)
-            
-            # Generate TTS
-            try:
-                tts.tts_to_file(text=cleaned_text, file_path=output_path)
-                
-                # Verify file was created
-                if not os.path.exists(output_path):
-                    raise Exception(f"Failed to create output file: {output_path}")
-                
-                file_size = os.path.getsize(output_path)
-                
-                # Upload to OCI
-                upload_result, error = upload_to_oci_with_retry(
-                    output_path, filename, request.project_id, "voiceover"
-                )
-                
-                if error:
-                    results.append({
-                        "status": "success_local",
-                        "filename": filename,
-                        "local_file": output_path,
-                        "file_size": file_size,
-                        "upload_error": error,
-                        "text_index": i
-                    })
-                else:
-                    # Clean up local file on successful upload
-                    try:
-                        os.remove(output_path)
-                    except:
-                        pass
-                    
-                    results.append({
-                        "status": "success",
-                        "filename": filename,
-                        "oci_path": upload_result.get("path", f"{request.project_id}/voiceover/{filename}"),
-                        "file_size": file_size,
-                        "text_index": i
-                    })
-                    
-            except Exception as e:
-                results.append({
-                    "status": "error",
-                    "filename": filename,
-                    "error": str(e),
-                    "text_index": i
-                })
-        
-        # Count successes and errors
-        success_count = len([r for r in results if r["status"] in ["success", "success_local"]])
-        error_count = len([r for r in results if r["status"] == "error"])
-        
-        return {
-            "status": "completed",
-            "message": f"Processed {len(request.texts)} texts: {success_count} successful, {error_count} errors",
-            "results": results,
-            "summary": {
-                "total": len(request.texts),
-                "successful": success_count,
-                "errors": error_count
-            }
-        }
-        
-    except Exception as e:
-        print(f"❌ Batch TTS error: {str(e)}")
-        raise HTTPException(status_code=500, detail=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")
-        
-        # 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)
-        
-        # 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)
-            }
-        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),
-                    "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
-            }
-        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",
-        "male_deep": "Deep male voice (Fallback to default)",
-        "male_medium": "Medium male voice (Fallback to default)",
-        "female_1": "Female voice 1 (Fallback to default)", 
-        "female_2": "Female voice 2 (Fallback to default)"
+        "clear_male": "Clear male voice (Tacotron2) - Fast"
     }
     return {"voice_styles": styles}
 
@@ -775,12 +454,10 @@ 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,
         "device": DEVICE,
         "oci_configured": bool(OCI_UPLOAD_API_URL),
         "startup_time": app_startup_time.isoformat(),
-        "uptime": str(datetime.now() - app_startup_time),
-        "model_load_attempts": model_load_attempts
+        "uptime": str(datetime.now() - app_startup_time)
     }
 
 # Startup event - NO MODEL LOADING to avoid timeouts