Update app.py

app.py

@@ -13,7 +13,6 @@ from fastapi import FastAPI, HTTPException, Form, UploadFile, File
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 import torch
-import numpy as np
 
 # Global state
 tts = None
@@ -78,13 +77,6 @@ class TTSRequest(BaseModel):
     language: Optional[str] = "en"
     voice_style: Optional[str] = "default_female"
 
-class BatchTTSRequest(BaseModel):
-    texts: List[str]
-    project_id: str
-    voice_name: Optional[str] = "default"
-    language: Optional[str] = "en"
-    voice_style: Optional[str] = "default_female"
-
 class VoiceCloneRequest(BaseModel):
     project_id: str
     voice_name: str
@@ -101,112 +93,47 @@ def clean_text(text):
     if not text or not isinstance(text, str):
         return "Hello"
     
-    # Remove any non-ASCII characters
     text = text.encode('ascii', 'ignore').decode('ascii')
-    
-    # Remove any problematic characters but keep basic punctuation
     text = re.sub(r'[^\w\s\.\,\!\?\-\'\"\:]', '', text)
-    
-    # Replace multiple spaces with single space
     text = re.sub(r'\s+', ' ', text)
     
-    # Ensure text ends with punctuation if it's a sentence
     if len(text) > 10 and not re.search(r'[\.\!\?]$', text):
         text = text + '.'
     
     text = text.strip()
     
-    # If text is empty after cleaning, use default
     if not text:
         text = "Hello world"
     
     return 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"""
+    """Upload file to OCI"""
     try:
         if not OCI_UPLOAD_API_URL:
             print("⚠️ OCI upload skipped - OCI_UPLOAD_API_URL not configured")
             return {"status": "skipped", "message": "OCI upload disabled"}, None
         
-        if not OCI_UPLOAD_API_URL.startswith(('http://', 'https://')):
-            return None, f"Invalid OCI URL format: {OCI_UPLOAD_API_URL}"
-        
         url = f"{OCI_UPLOAD_API_URL}/api/upload"
         print(f"🔗 Attempting upload to: {url}")
         
         with open(file_path, "rb") as f:
             files = {"file": (filename, f, "audio/wav")}
-            data = {
-                "project_id": project_id,
-                "subfolder": "voiceover"
-            }
-            
-            headers = {
-                "User-Agent": "TTS-API/1.0",
-                "Accept": "application/json"
-            }
+            data = {"project_id": project_id, "subfolder": "voiceover"}
             
-            response = requests.post(
-                url, 
-                files=files, 
-                data=data, 
-                headers=headers,
-                timeout=30
-            )
-            
-            print(f"📡 Upload response status: {response.status_code}")
+            response = requests.post(url, files=files, data=data, timeout=30)
             
             if response.status_code == 200:
-                try:
-                    result = response.json()
-                    if result.get("status") == "success":
-                        print(f"✅ Upload successful: {result}")
-                        return result, None
-                    else:
-                        error_msg = result.get("message", "Upload failed - unknown error")
-                        print(f"❌ Upload failed: {error_msg}")
-                        return None, error_msg
-                except ValueError as e:
-                    return None, f"Invalid JSON response: {str(e)}"
-            else:
-                error_msg = f"Upload failed with status {response.status_code}: {response.text}"
-                print(f"❌ {error_msg}")
-                return None, error_msg
-                
-    except Exception as 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=2):
-    """Upload file to OCI with retry logic"""
-    for attempt in range(max_retries):
-        try:
-            print(f"🔄 Upload attempt {attempt + 1} of {max_retries} for {filename}")
-            result, error = upload_to_oci(file_path, filename, project_id, file_type)
-            
-            if error:
-                if attempt < max_retries - 1:
-                    wait_time = 2 ** attempt
-                    print(f"⏳ Upload failed, retrying in {wait_time}s: {error}")
-                    time.sleep(wait_time)
-                    continue
+                result = response.json()
+                if result.get("status") == "success":
+                    return result, None
                 else:
-                    return None, error
+                    return None, result.get("message", "Upload failed")
             else:
-                return result, None
+                return None, f"Upload failed with status {response.status_code}"
                 
-        except Exception as e:
-            if attempt < max_retries - 1:
-                wait_time = 2 ** attempt
-                print(f"⏳ Upload exception, retrying in {wait_time}s: {str(e)}")
-                time.sleep(wait_time)
-                continue
-            else:
-                return None, f"Upload failed after {max_retries} attempts: {str(e)}"
-    
-    return None, "Upload failed: unexpected error"
+    except Exception as e:
+        return None, f"Upload error: {str(e)}"
 
 def get_voice_path(voice_name: str):
     """Get path to voice file"""
@@ -217,9 +144,7 @@ def get_voice_path(voice_name: str):
     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
+    return None
 
 def clone_voice(voice_name: str, audio_files: List[str], description: str = ""):
     """Clone a voice from audio samples"""
@@ -244,40 +169,8 @@ 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:
-        # Try soundfile first
-        try:
-            import soundfile as sf
-            sf.write(file_path, audio, 22050)
-            return True
-        except ImportError:
-            print("⚠️ soundfile not available, using fallback method")
-        
-        # Fallback: use wave library
-        import wave
-        import numpy as np
-        
-        if isinstance(audio, list):
-            audio = np.array(audio)
-        
-        audio_int16 = (audio * 32767).astype(np.int16)
-        
-        with wave.open(file_path, 'wb') as wav_file:
-            wav_file.setnchannels(1)
-            wav_file.setsampwidth(2)
-            wav_file.setframerate(22050)
-            wav_file.writeframes(audio_int16.tobytes())
-        
-        return True
-        
-    except Exception as e:
-        print(f"❌ Failed to save WAV: {e}")
-        return False
-
 def load_tts_model(voice_style="default_female"):
-    """Load TTS model with different voice options - LAZY LOADING"""
+    """Load TTS model with lazy loading"""
     global tts, model_loaded, current_model, model_loading, current_voice_style, voice_cloning_supported
     
     if model_loading:
@@ -293,7 +186,6 @@ def load_tts_model(voice_style="default_female"):
     try:
         from TTS.api import TTS
         
-        # Use only fast, lightweight models with proper voice styles
         model_options = {
             "default_female": {
                 "name": "tts_models/en/ljspeech/tacotron2-DDC",
@@ -303,22 +195,6 @@ def load_tts_model(voice_style="default_female"):
                 "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",
@@ -329,24 +205,11 @@ def load_tts_model(voice_style="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)...")
         
-        # Load the selected model
         tts = TTS(selected_model["name"]).to(DEVICE)
         
         # Quick test
-        try:
-            test_text = "Hello"
-            audio = tts.tts(text=test_text)
-            print(f"✅ {selected_model['description']} loaded successfully!")
-        except Exception as test_error:
-            print(f"❌ Model test failed: {test_error}")
-            # Try fallback to default
-            print("🔄 Trying fallback model...")
-            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"
+        tts.tts_to_file(text="Hello", file_path="/tmp/test.wav")
         
         model_loaded = True
         current_model = selected_model["name"]
@@ -360,110 +223,60 @@ def load_tts_model(voice_style="default_female"):
     finally:
         model_loading = False
 
-# Health check endpoints - CRITICAL FOR DEPLOYMENT
+# Health check endpoints
 @app.get("/")
 async def root():
-    """Root endpoint - redirect to docs"""
+    """Root endpoint"""
     return {
         "status": "running", 
         "service": "TTS API",
-        "message": "Visit /docs for API documentation"
+        "message": "Visit /docs for API documentation",
+        "model_loaded": model_loaded
     }
 
 @app.get("/health")
 async def health_check():
-    """Health check endpoint - must respond quickly"""
+    """Health check endpoint"""
     return {
         "status": "healthy",
         "timestamp": datetime.now().isoformat(),
-        "startup_time": app_startup_time.isoformat(),
         "model_loaded": model_loaded,
         "service": "TTS API"
     }
 
 @app.get("/api/health")
 async def api_health_check():
-    """API health check with model status"""
+    """API health check"""
     return {
         "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()
+        "device": DEVICE
     }
 
-@app.get("/api/oci-health")
-async def check_oci_health():
-    """Check if OCI upload service is available"""
-    if not OCI_UPLOAD_API_URL:
-        return {
-            "status": "not_configured",
-            "message": "OCI_UPLOAD_API_URL environment variable is not set"
-        }
-    
-    try:
-        test_url = f"{OCI_UPLOAD_API_URL}/api/health"
-        response = requests.get(test_url, timeout=5)
-        
-        if response.status_code == 200:
-            return {
-                "status": "healthy",
-                "oci_url": OCI_UPLOAD_API_URL,
-                "message": "OCI service is reachable"
-            }
-        else:
-            return {
-                "status": "unhealthy", 
-                "oci_url": OCI_UPLOAD_API_URL,
-                "message": f"OCI service returned status {response.status_code}"
-            }
-    except Exception as e:
-        return {
-            "status": "error",
-            "oci_url": OCI_UPLOAD_API_URL,
-            "message": f"Cannot connect to OCI service: {str(e)}"
-        }
-
-# API endpoints with lazy loading
+# API endpoints
 @app.post("/api/tts")
 async def generate_tts(request: TTSRequest):
-    """Generate TTS for a single text with lazy model loading"""
+    """Generate TTS for a single text"""
     try:
-        # 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 {
                     "status": "error",
-                    "message": "TTS model failed to load. Please try again in a moment.",
-                    "requires_tos_acceptance": True,
-                    "tos_url": "https://coqui.ai/cpml.txt"
+                    "message": "TTS model failed to load. Please try again."
                 }
         
         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")
         filename = f"voiceover_{timestamp}.wav"
         output_path = f"/tmp/output/{filename}"
         
-        # Ensure output directory exists
         os.makedirs(os.path.dirname(output_path), exist_ok=True)
         
+        cleaned_text = clean_text(request.text)
+        
         # Get voice path if custom voice is requested
         speaker_wav = None
         if request.voice_name != "default":
@@ -471,102 +284,43 @@ async def generate_tts(request: TTSRequest):
             if not speaker_wav:
                 return {
                     "status": "error",
-                    "message": f"Voice '{request.voice_name}' not found. Please clone the voice first using /api/clone-voice."
+                    "message": f"Voice '{request.voice_name}' not found."
                 }
-            print(f"🎙️ Using cloned voice: {request.voice_name}")
         
-        print(f"🔊 Generating TTS to: {output_path}")
-
-        # Clean the text before generation
-        cleaned_text = clean_text(request.text)
-        print(f"📝 Cleaned text: '{cleaned_text}'")
-        
-        # Generate TTS with error handling
-        try:
-            print(f"🔊 Generating TTS with {current_model}...")
-            
-            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...")
-                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):
-                    raise Exception("Failed to save audio file")
-                    
-            except Exception as alt_error:
-                print(f"❌ Alternative method also failed: {alt_error}")
-                raise alt_error
+        if speaker_wav and voice_cloning_supported:
+            tts.tts_to_file(text=cleaned_text, file_path=output_path, speaker_wav=speaker_wav)
+        else:
+            tts.tts_to_file(text=cleaned_text, file_path=output_path)
         
-        # Verify the file was created
         if not os.path.exists(output_path):
-            raise Exception(f"TTS failed to create output file: {output_path}")
+            raise Exception("TTS failed to create output file")
         
         file_size = os.path.getsize(output_path)
-        print(f"✅ TTS generated: {output_path} ({file_size} bytes)")
         
-        # Upload to OCI
-        upload_result, error = upload_to_oci_with_retry(
-            output_path, filename, request.project_id, "voiceover"
-        )
+        upload_result, error = upload_to_oci(output_path, filename, request.project_id)
         
         if error:
-            print(f"❌ OCI upload failed: {error}")
-            # Return success with local file info
             return {
                 "status": "success_local",
                 "message": f"TTS generated locally (upload failed: {error})",
                 "local_file": output_path,
                 "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
+                "file_size": file_size
             }
         
-        print(f"✅ Upload successful: {filename}")
-        
-        # Clean up local file
         try:
             os.remove(output_path)
-            print(f"🧹 Cleaned up local file: {output_path}")
-        except Exception as cleanup_error:
-            print(f"⚠️ Could not clean up file: {cleanup_error}")
+        except:
+            pass
             
         return {
             "status": "success",
             "message": "TTS generated and uploaded successfully",
             "filename": filename,
-            "oci_path": upload_result.get("path", f"{request.project_id}/voiceover/{filename}"),
-            "model_used": current_model,
-            "voice_style": request.voice_style,
-            "voice_name": request.voice_name,
-            "voice_cloning_used": speaker_wav is not None
+            "oci_path": upload_result.get("path", f"{request.project_id}/voiceover/{filename}")
         }
         
     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")
@@ -581,15 +335,6 @@ async def clone_voice_endpoint(
         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')):
@@ -599,12 +344,9 @@ async def clone_voice_endpoint(
             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)
@@ -616,16 +358,12 @@ async def clone_voice_endpoint(
                 "status": "success",
                 "message": message,
                 "voice_name": voice_name,
-                "samples_used": len(temp_files),
-                "project_id": project_id
+                "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")
@@ -642,51 +380,20 @@ async def list_voices():
                 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()
+                    "samples_count": len(samples)
                 })
         
         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)",
         "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)"
+        "voice_clone": "XTTS v2 - Voice cloning supported"
     }
     return {"voice_styles": styles}
 
@@ -697,12 +404,7 @@ 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(),
-        "uptime": str(datetime.now() - app_startup_time)
+        "device": DEVICE
     }
 
 if __name__ == "__main__":