Update app.py

app.py

@@ -1,187 +1,54 @@
-from dataclasses import dataclass
-from typing import List, Tuple, Dict
+import gradio as gr
 import os
-import re
-import httpx
-import json
-from openai import OpenAI
-import edge_tts
-import tempfile
-from pydub import AudioSegment
-import base64
-from pathlib import Path
-import shutil  # Importamos shutil para manejo de directorios
+import asyncio
+from conver import ConversationConfig, URLToAudioConverter
+from dotenv import load_dotenv
 
-@dataclass
-class ConversationConfig:
-    max_words: int = 3000
-    prefix_url: str = "https://r.jina.ai/"
-    model_name: str = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo"
+load_dotenv()
 
-class URLToAudioConverter:
-    def __init__(self, config: ConversationConfig, llm_api_key: str):
-        self.config = config
-        self.llm_client = OpenAI(api_key=llm_api_key, base_url="https://api.together.xyz/v1")
-        self.llm_out = None
+async def synthesize(article_url, text_input, language="en"):
+    if not article_url and not text_input:
+        return "Error: Ingresa una URL o texto", None
 
-    def fetch_text(self, url: str) -> str:
-        if not url:
-            raise ValueError("URL cannot be empty")
-
-        full_url = f"{self.config.prefix_url}{url}"
-        try:
-            response = httpx.get(full_url, timeout=60.0)
-            response.raise_for_status()
-            return response.text
-        except httpx.HTTPError as e:
-            raise RuntimeError(f"Failed to fetch URL: {e}")
-
-    def extract_conversation(self, text: str) -> Dict:
-        if not text:
-            raise ValueError("Input text cannot be empty")
-
-        try:
-            # Prompt mejorado para obtener JSON consistente
-            prompt = (
-                f"{text}\nConvert the provided text into a short informative podcast conversation "
-                f"between two experts. Return ONLY a JSON object with the following structure:\n"
-                '{"conversation": [{"speaker": "Speaker1", "text": "..."}, {"speaker": "Speaker2", "text": "..."}]}'
-            )
-            
-            chat_completion = self.llm_client.chat.completions.create(
-                messages=[{"role": "user", "content": prompt}],
-                model=self.config.model_name,
-                response_format={"type": "json_object"}  # Fuerza formato JSON
-            )
-            
-            # Extracción robusta de JSON
-            response_content = chat_completion.choices[0].message.content
-            json_str = response_content.strip()
-            
-            # Limpieza de texto alrededor del JSON
-            if not json_str.startswith('{'):
-                start = json_str.find('{')
-                if start != -1:
-                    json_str = json_str[start:]
-            
-            if not json_str.endswith('}'):
-                end = json_str.rfind('}')
-                if end != -1:
-                    json_str = json_str[:end+1]
-            
-            return json.loads(json_str)
-        except Exception as e:
-            # Debug: Imprime la respuesta del modelo para diagnóstico
-            print(f"Error en extract_conversation: {str(e)}")
-            print(f"Respuesta del modelo: {response_content}")
-            raise RuntimeError(f"Failed to extract conversation: {str(e)}")
-
-    async def text_to_speech(self, conversation_json: Dict, voice_1: str, voice_2: str) -> Tuple[List[str], str]:
-        output_dir = Path(self._create_output_directory())
-        filenames = []
+    try:
+        config = ConversationConfig()
+        converter = URLToAudioConverter(config, llm_api_key=os.environ.get("TOGETHER_API_KEY"))
         
-        try:
-            for i, turn in enumerate(conversation_json["conversation"]):
-                filename = output_dir / f"output_{i}.mp3"
-                voice = voice_1 if i % 2 == 0 else voice_2
-
-                tmp_path, error = await self._generate_audio(turn["text"], voice)
-                if error:
-                    raise RuntimeError(f"Text-to-speech failed: {error}")
-
-                os.rename(tmp_path, filename)
-                filenames.append(str(filename))
-
-            return filenames, str(output_dir)
-        except Exception as e:
-            raise RuntimeError(f"Failed to convert text to speech: {e}")
-
-    async def _generate_audio(self, text: str, voice: str, rate: int = 0, pitch: int = 0) -> Tuple[str, str]:
-        if not text.strip():
-            return None, "Text cannot be empty"
-        if not voice:
-            return None, "Voice cannot be empty"
-
-        voice_short_name = voice.split(" - ")[0]
-        rate_str = f"{rate:+d}%"
-        pitch_str = f"{pitch:+d}Hz"
-        communicate = edge_tts.Communicate(text, voice_short_name, rate=rate_str, pitch=pitch_str)
-
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
-            tmp_path = tmp_file.name
-            await communicate.save(tmp_path)
-
-        return tmp_path, None
-
-    def _create_output_directory(self) -> str:
-        random_bytes = os.urandom(8)
-        folder_name = base64.urlsafe_b64encode(random_bytes).decode("utf-8")
-        os.makedirs(folder_name, exist_ok=True)
-        return folder_name
-
-    def combine_audio_files(self, filenames: List[str], output_file: str) -> None:
-        if not filenames:
-            raise ValueError("No input files provided")
-
-        try:
-            combined = AudioSegment.empty()
-            for filename in filenames:
-                audio_segment = AudioSegment.from_file(filename, format="mp3")
-                combined += audio_segment
-
-            combined.export(output_file, format="mp3")
-
-            # Limpieza mejorada y robusta
-            dir_path = os.path.dirname(filenames[0])
+        # Voces humanizadas
+        voices = {
+            "en": ("en-US-AvaMultilingualNeural", "en-US-AndrewMultilingualNeural"),
+            "es": ("es-ES-AlvaroNeural", "es-ES-ElviraNeural")
+        }
+        voice1, voice2 = voices.get(language, voices["en"])
+
+        if text_input:
+            output_file, conversation = await converter.text_to_audio(text_input, voice1, voice2)
+        else:
+            output_file, conversation = await converter.url_to_audio(article_url, voice1, voice2)
             
-            # Eliminar todos los archivos en el directorio
-            for file in os.listdir(dir_path):
-                file_path = os.path.join(dir_path, file)
-                if os.path.isfile(file_path):
-                    try:
-                        os.remove(file_path)
-                    except Exception as e:
-                        print(f"Warning: Could not remove file {file_path}: {str(e)}")
-            
-            # Intentar eliminar el directorio (no crítico si falla)
-            try:
-                os.rmdir(dir_path)
-            except OSError as e:
-                print(f"Info: Could not remove directory {dir_path}: {str(e)}")
-                # No es crítico, el espacio puede continuar
-                
-        except Exception as e:
-            raise RuntimeError(f"Failed to combine audio files: {e}")
-
-    async def url_to_audio(self, url: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        text = self.fetch_text(url)
-
-        words = text.split()
-        if len(words) > self.config.max_words:
-            text = " ".join(words[:self.config.max_words])
-
-        conversation_json = self.extract_conversation(text)
-        conversation_text = "\n".join(
-            f"{turn['speaker']}: {turn['text']}" for turn in conversation_json["conversation"]
-        )
-        self.llm_out = conversation_json
-        audio_files, folder_name = await self.text_to_speech(
-            conversation_json, voice_1, voice_2
-        )
-
-        final_output = os.path.join(folder_name, "combined_output.mp3")
-        self.combine_audio_files(audio_files, final_output)
-        return final_output, conversation_text
-
-    async def text_to_audio(self, text: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        """Método para procesar texto directo"""
-        conversation_json = self.extract_conversation(text)
-        conversation_text = "\n".join(
-            f"{turn['speaker']}: {turn['text']}" for turn in conversation_json["conversation"]
-        )
-        audio_files, folder_name = await self.text_to_speech(
-            conversation_json, voice_1, voice_2
-        )
-        final_output = os.path.join(folder_name, "combined_output.mp3")
-        self.combine_audio_files(audio_files, final_output)
-        return final_output, conversation_text
+        return conversation, output_file
+    except Exception as e:
+        return f"Error: {str(e)}", None
+
+def synthesize_sync(article_url, text_input, language):
+    return asyncio.run(synthesize(article_url, text_input, language))
+
+with gr.Blocks(theme='gstaff/sketch') as demo:
+    gr.Markdown("# 🎙 Podcast Converter (Human Voices)")
+    with gr.Group():
+        text_url = gr.Textbox(label="URL (opcional)", placeholder="https://...")
+        text_input = gr.Textbox(label="O texto directo", lines=5)
+        language = gr.Dropdown(["en", "es"], label="Idioma", value="en")
+        btn = gr.Button("Generar Podcast", variant="primary")
+    
+    with gr.Row():
+        conv_display = gr.Textbox(label="Conversación", interactive=False, lines=10)
+        aud = gr.Audio(label="Audio Generado", interactive=False)
+    
+    btn.click(
+        synthesize_sync,
+        inputs=[text_url, text_input, language],
+        outputs=[conv_display, aud]
+    )
+
+demo.launch()