Upload app.py with huggingface_hub

app.py

@@ -11,6 +11,17 @@ import json
 import time
 import torchaudio
 
+# 🛡️ ZeroGPU Support (v68)
+try:
+    import spaces
+    print("✅ ZeroGPU/Spaces detected")
+except ImportError:
+    print("⚠️ Spaces library not found. Using mock decorator for local run.")
+    class spaces:
+        @staticmethod
+        def GPU(f): return f
+
+
 # 🛠️ Monkeypatch torchaudio.backend (DeepFilterNet compatibility)
 # DeepFilterNet uses older torchaudio API structure (torchaudio.backend.common.AudioMetaData)
 # We mock it here before importing df
@@ -49,9 +60,8 @@ if not hasattr(torchaudio, "info"):
     torchaudio.info = mock_info
 
 from df.enhance import enhance, init_df, load_audio, save_audio
-# from df.utils import download_model # Not needed/Not found in this version
 
-# FORCE BUILD TRIGGER: 14:55:00 Jan 20 2026
+# FORCE BUILD TRIGGER: 15:10:00 Jan 20 2026
 
 # 🛠️ Monkeypatch torchaudio.load to bypass TorchCodec requirement
 try:
@@ -100,13 +110,11 @@ def load_models():
     
     if MODELS["translate"] is None:
         print("🌍 Loading Google Translate (deep-translator)...")
-        # No heavy object to load, just a placeholder or class ref
         MODELS["translate"] = "deep-translator-active"
 
     if MODELS["denoiser"] is None:
         print("🧹 Loading DeepFilterNet (Voice Cleaner)...")
         try:
-            # Initialize DeepFilterNet model
             df_ret = init_df()
             if isinstance(df_ret, (list, tuple)) and len(df_ret) > 1:
                 MODELS["denoiser"] = df_ret[0]
@@ -116,7 +124,6 @@ def load_models():
         except Exception as e:
             print(f"⚠️ Failed to load denoiser: {e}")
             try:
-                print("🔄 Final attempt for DeepFilterNet init...")
                 MODELS["denoiser"] = init_df()
             except:
                 pass
@@ -125,7 +132,6 @@ def load_models():
         print("🔊 Loading XTTS-v2 (STRICT GPU PREFERRED)...")
         from TTS.api import TTS
         try:
-            # Try GPU first
             MODELS["tts"] = TTS(model_name="tts_models/multilingual/multi-dataset/xtts_v2", gpu=True)
             print("✨ XTTS-v2 Loaded on GPU")
         except Exception as tts_e:
@@ -137,10 +143,11 @@ def load_models():
                 print(f"❌ FATAL: Could not load XTTS-v2 on any hardware: {cpu_e}")
                 raise cpu_e
 
+@spaces.GPU
 def core_process(request_dict):
     """Internal logic used by both FastAPI and Gradio"""
     action = request_dict.get("action")
-    print(f"--- 🛠️ Processing Action: {action} ---")
+    print(f"--- 🛠️ Processing Action: {action} (ZeroGPU Context) ---")
     start_time = time.time()
     
     if action == "health":
@@ -160,20 +167,18 @@ def core_process(request_dict):
             temp_path = f.name
         try:
             print("🚀 Faster-Whisper Transcription Starting (Instant Mode)...")
-            # beam_size=1 for instantaneous results (Greedy Search)
             segments, info = MODELS["stt"].transcribe(temp_path, language=lang, beam_size=1)
             text = " ".join([segment.text for segment in segments]).strip()
             print(f"✨ Transcription Done: '{text[:50]}...'")
             return {"text": text}
         finally:
-            os.unlink(temp_path)
+            if os.path.exists(temp_path): os.unlink(temp_path)
     
     elif action == "translate":
         text = request_dict.get("text")
         target_lang = request_dict.get("target_lang")
         print(f"🌍 Translate: '{text[:50]}...' to {target_lang}")
         
-        # Map codes to Google standard (ISO 639-1)
         g_lang_map = {
             "en": "en", "fr": "fr", "es": "es", "de": "de", 
             "ar": "ar", "it": "it", "pt": "pt", "ru": "ru", 
@@ -182,7 +187,6 @@ def core_process(request_dict):
         g_target = g_lang_map.get(target_lang, "en")
         
         from deep_translator import GoogleTranslator
-        # deep-translator handles 'auto' source by default
         result = GoogleTranslator(source='auto', target=g_target).translate(text)
         
         print(f"✨ Translation Done: '{result[:50]}...'")
@@ -200,7 +204,6 @@ def core_process(request_dict):
                 f.write(speaker_bytes)
                 speaker_wav_path = f.name
         else:
-             # 🛡️ FALLBACK: Generate dummy speaker if missing
             print("⚠️ No speaker ref provided. Using generated default.")
             import wave, struct, math
             default_path = "default_speaker.wav"
@@ -210,12 +213,9 @@ def core_process(request_dict):
                         wav_file.setnchannels(1)
                         wav_file.setsampwidth(2)
                         wav_file.setframerate(24000)
-                        # Generate 1 sec of silence/noise to satisfy XTTS input requirement
                         data = [struct.pack('<h', int(math.sin(x/100.0)*3000)) for x in range(24000)]
                         wav_file.writeframes(b''.join(data))
-                except Exception as e:
-                    print(f"❌ Failed to create default speaker: {e}")
-            
+                except: pass
             if os.path.exists(default_path):
                 speaker_wav_path = default_path
             
@@ -225,41 +225,29 @@ def core_process(request_dict):
             
             if speaker_wav_path:
                 MODELS["tts"].tts_to_file(text=text, language=lang, file_path=output_path, speaker_wav=speaker_wav_path)
-            else:
-                 # If fallback failed, try cloning from self (hack) or fail gracefully
-                raise RuntimeError("No speaker_wav available for XTTS voice cloning.")
             
-            # --- 🧹 DEEPFILTERNET DENOISING ---
             if MODELS["denoiser"]:
-                print("🧹 Cleaning Audio with DeepFilterNet...")
                 try:
                     noisy_audio, _ = load_audio(output_path, sr=48000)
                     enhanced_audio = enhance(MODELS["denoiser"], noisy_audio, pad=True)
                     save_audio(output_path, enhanced_audio, 48000)
-                    print("✨ Audio Cleaned Successfully")
-                except Exception as e:
-                    print(f"⚠️ Denoising failed, using original: {e}")
-            # ----------------------------------
+                except: pass
 
             with open(output_path, "rb") as f:
                 audio_b64 = base64.b64encode(f.read()).decode()
             print("✨ TTS Done")
             return {"audio": audio_b64}
         finally:
-            if speaker_wav_path and os.path.exists(speaker_wav_path):
+            if speaker_wav_path and os.path.exists(speaker_wav_path) and "default_speaker" not in speaker_wav_path:
                 os.unlink(speaker_wav_path)
-            if os.path.exists(output_path):
-                os.unlink(output_path)
+            if 'output_path' in locals() and os.path.exists(output_path): os.unlink(output_path)
 
     elif action == "s2st":
         audio_b64 = request_dict.get("file")
         source_lang = request_dict.get("source_lang")
         target_lang = request_dict.get("target_lang")
         speaker_wav_b64 = request_dict.get("speaker_wav")
-
-        print(f"🚀 [S2ST] Action Started (Source: {source_lang}, Target: {target_lang})")
         
-        # 1. Decode Audio
         audio_bytes = base64.b64decode(audio_b64)
         with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
             f.write(audio_bytes)
@@ -272,115 +260,47 @@ def core_process(request_dict):
                 sf.write(speaker_bytes)
                 speaker_wav_path = sf.name
         else:
-            # 🛡️ FALLBACK: Use default speaker if input audio is too short/bad for cloning or not provided
             default_path = "default_speaker.wav"
             if os.path.exists(default_path):
-                print(f"⚠️ [S2ST] No valid speaker ref provided. Using default: {default_path}")
                 speaker_wav_path = default_path
         
         try:
-            # 2. STT (Whisper Pro)
-            print("🎙️ [S2ST] Phase 1: Whisper Transcription...")
-            
-            # 🛡️ AUDIO PADDING (1.5s)
+            # Padding & Denoising
             try:
                 waveform, sr = torchaudio.load(temp_path)
-                
-                # --- 🧹 PRE-PROCESS: DeepFilterNet on INPUT ---
                 if MODELS["denoiser"]:
-                    try:
-                        noisy_in, _ = load_audio(temp_path, sr=48000)
-                        clean_in = enhance(MODELS["denoiser"], noisy_in, pad=True)
-                        save_audio(temp_path, clean_in, 48000) # Overwrite temp with clean
-                        print("🧹 [S2ST] Input Audio Cleaned (Pre-ASR/Clone)")
-                        
-                        waveform, sr = torchaudio.load(temp_path)
-                    except Exception as df_e:
-                        print(f"⚠️ Input cleaning failed: {df_e}")
-
-                silence_frames = int(1.5 * sr)
-                silence = torch.zeros((waveform.shape[0], silence_frames))
-                padded_waveform = torch.cat([waveform, silence], dim=1)
+                    noisy_in, _ = load_audio(temp_path, sr=48000)
+                    clean_in = enhance(MODELS["denoiser"], noisy_in, pad=True)
+                    save_audio(temp_path, clean_in, 48000)
+                    waveform, sr = torchaudio.load(temp_path)
                 
-                torchaudio.save(temp_path, padded_waveform, sr)
-                print(f"🛡️ Added 1.5s silence padding to audio (New duration: {padded_waveform.shape[1]/sr:.2f}s)")
-            except Exception as pe:
-                print(f"⚠️ Padding/Cleaning failed: {pe}")
+                silence = torch.zeros((waveform.shape[0], int(1.5 * sr)))
+                padded = torch.cat([waveform, silence], dim=1)
+                torchaudio.save(temp_path, padded, sr)
+            except: pass
 
-            print("🎙️ [S2ST] Phase 1: Faster-Whisper Transcription...")
-            segments, info = MODELS["stt"].transcribe(
-                temp_path, 
-                language=source_lang, 
-                beam_size=1,
-                best_of=1
-            )
+            # STT
+            segments, info = MODELS["stt"].transcribe(temp_path, language=source_lang, beam_size=1)
             text = " ".join([segment.text for segment in segments]).strip()
-            
-            valid_endings = ('.', '!', '?', '…', '。', '！', '？')
-            if text and not text.endswith(valid_endings):
-                print(f"⚠️ Incomplete sentence detected: '{text}' -> Appending ellipsis")
-                text += "..."
-            
-            print(f"✨ [S2ST] Transcribed: '{text[:50]}...'")
-
-            if not text:
-                return {"error": "No speech detected"}
-
-            # 3. Translate
-            print("🌍 [S2ST] Phase 2: Google Translation...")
-            try:
-                g_lang_map = {
-                    "en": "en", "fr": "fr", "es": "es", "de": "de", 
-                    "ar": "ar", "it": "it", "pt": "pt", "ru": "ru", 
-                    "zh": "zh-cn", "ja": "ja", "ko": "ko", "hi": "hi"
-                }
-                g_target = g_lang_map.get(target_lang, "en")
-                from deep_translator import GoogleTranslator
-                translated_text = GoogleTranslator(source='auto', target=g_target).translate(text)
-            except Exception as tr_e:
-                translated_text = text 
+            if text and not text.endswith(('.', '!', '?', '…')): text += "..."
+            if not text: return {"error": "No speech detected"}
 
-            print(f"✨ [S2ST] Translated: '{translated_text[:50]}...'")
+            # Translate
+            g_lang_map = {"en": "en", "fr": "fr", "es": "es", "de": "de", "ar": "ar", "it": "it", "pt": "pt", "ru": "ru", "zh": "zh-cn", "ja": "ja", "ko": "ko", "hi": "hi"}
+            g_target = g_lang_map.get(target_lang, "en")
+            from deep_translator import GoogleTranslator
+            translated_text = GoogleTranslator(source='auto', target=g_target).translate(text)
 
-            # 4. TTS
-            print("🔊 [S2ST] Phase 3: XTTS Synthesis...")
+            # TTS
             with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as output_file:
                 output_path = output_file.name
             
-            if len(translated_text) > 240:
-                print(f"✂️ Text too long ({len(translated_text)}), splitting...")
-                import re
-                sub_segments = re.split(r'(?<=[.!?])\s+', translated_text)
-                combined_audio = []
-                for idx, sub in enumerate(sub_segments):
-                    if not sub.strip(): continue
-                    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as sub_file:
-                        sub_path = sub_file.name
-                    try:
-                        MODELS["tts"].tts_to_file(text=sub, language=target_lang, file_path=sub_path, speaker_wav=speaker_wav_path)
-                        wav, sr = torchaudio.load(sub_path)
-                        combined_audio.append(wav)
-                    finally:
-                        if os.path.exists(sub_path): os.unlink(sub_path)
-                
-                if combined_audio:
-                    final_wav = torch.cat(combined_audio, dim=1)
-                    torchaudio.save(output_path, final_wav, sr)
-                else:
-                    MODELS["tts"].tts_to_file(text="Error", language=target_lang, file_path=output_path, speaker_wav=speaker_wav_path)
-            else:
-                MODELS["tts"].tts_to_file(text=translated_text, language=target_lang, file_path=output_path, speaker_wav=speaker_wav_path)
+            MODELS["tts"].tts_to_file(text=translated_text, language=target_lang, file_path=output_path, speaker_wav=speaker_wav_path)
 
             with open(output_path, "rb") as o:
                 audio_out_b64 = base64.b64encode(o.read()).decode()
             
-            print("🏁 [S2ST] All phases complete!")
-            return {
-                "text": text,
-                "translated": translated_text,
-                "audio": audio_out_b64
-            }
-
+            return {"text": text, "translated": translated_text, "audio": audio_out_b64}
         finally:
             if os.path.exists(temp_path): os.unlink(temp_path)
             if speaker_wav_path and os.path.exists(speaker_wav_path) and "default_speaker" not in speaker_wav_path:
@@ -419,6 +339,7 @@ def create_wav_header(sample_rate=24000, channels=1, bit_depth=16):
     return header
 
 @app.post("/api/v1/tts_stream")
+@spaces.GPU
 async def api_tts_stream(request: Request):
     try:
         load_models() 
@@ -427,8 +348,6 @@ async def api_tts_stream(request: Request):
         lang = data.get("lang")
         speaker_wav_b64 = data.get("speaker_wav")
         
-        print(f"🌊 [TTS Stream] Starting for: '{text[:50]}...' in {lang}")
-        
         speaker_wav_path = None
         if speaker_wav_b64:
             speaker_bytes = base64.b64decode(speaker_wav_b64)
@@ -448,16 +367,14 @@ async def api_tts_stream(request: Request):
                     stream_chunk_size=20
                 ):
                     yield (chunk * 32767).to(torch.int16).cpu().numpy().tobytes()
-                print("✨ [TTS Stream] Generation Complete")
             except Exception as ge:
-                print(f"❌ [TTS Stream] Generator error: {ge}")
+                print(f"❌ [Stream Error]: {ge}")
             finally:
                 if speaker_wav_path and os.path.exists(speaker_wav_path) and "default_speaker" not in speaker_wav_path:
                     os.unlink(speaker_wav_path)
 
         return StreamingResponse(stream_generator(), media_type="audio/wav")
     except Exception as e:
-        print(f"❌ [TTS Stream] Global Error: {traceback.format_exc()}")
         return {"error": str(e)}
 
 @app.get("/health")
@@ -483,5 +400,4 @@ demo = gr.Interface(
 app = gr.mount_gradio_app(app, demo, path="/")
 
 if __name__ == "__main__":
-    print("🚀 Starting FastAPI Server on port 7860...")
     uvicorn.run(app, host="0.0.0.0", port=7860)