Update conver.py

conver.py

@@ -9,12 +9,13 @@ import tempfile
 from pydub import AudioSegment
 import base64
 from pathlib import Path
+import numpy as np
 
 @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"
+    model_name: str = "meta-llama/Meta-Llama-3-8B-Instruct"
 
 class URLToAudioConverter:
     def __init__(self, config: ConversationConfig, llm_api_key: str):
@@ -23,6 +24,7 @@ class URLToAudioConverter:
         self.llm_out = None
 
     def fetch_text(self, url: str) -> str:
+        """Obtiene texto desde una URL"""
         if not url:
             raise ValueError("URL cannot be empty")
         full_url = f"{self.config.prefix_url}{url}"
@@ -34,133 +36,173 @@ class URLToAudioConverter:
             raise RuntimeError(f"Failed to fetch URL: {e}")
 
     def extract_conversation(self, text: str) -> Dict:
+        """Convierte texto plano a estructura de conversación"""
         if not text:
             raise ValueError("Input text cannot be empty")
         try:
             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": "..."}]}'
+                f"{text}\nConvert this text into a podcast conversation between two hosts. "
+                "Return ONLY JSON with this structure:\n"
+                '{"conversation": [{"speaker": "Host1", "text": "..."}, {"speaker": "Host2", "text": "..."}]}'
             )
-            chat_completion = self.llm_client.chat.completions.create(
+            response = self.llm_client.chat.completions.create(
                 messages=[{"role": "user", "content": prompt}],
                 model=self.config.model_name,
                 response_format={"type": "json_object"}
             )
-            response_content = chat_completion.choices[0].message.content
-            json_str = response_content.strip()
-            if not json_str.startswith('{'):
-                json_str = json_str[json_str.find('{'):]
-            if not json_str.endswith('}'):
-                json_str = json_str[:json_str.rfind('}')+1]
-            return json.loads(json_str)
+            json_str = response.choices[0].message.content.strip()
+            return json.loads(json_str[json_str.find('{'):json_str.rfind('}')+1])
         except Exception as e:
             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]:
+        """Convierte JSON de conversación a archivos de audio"""
         output_dir = Path(self._create_output_directory())
         filenames = []
         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}")
+                filename = output_dir / f"segment_{i}.mp3"
+                voice = voice_1 if turn["speaker"] == "Host1" else voice_2
+                tmp_path = await self._generate_audio(turn["text"], voice)
                 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}")
+            raise RuntimeError(f"Text-to-speech failed: {e}")
 
-    async def _generate_audio(self, text: str, voice: str, rate: int = 0, pitch: int = 0) -> Tuple[str, Optional[str]]:
+    async def _generate_audio(self, text: str, voice: str) -> str:
+        """Genera audio temporal con edge-tts"""
         if not text.strip():
-            return None, "Text cannot be empty"
-        voice_short_name = voice.split(" - ")[0]
+            raise ValueError("Text cannot be empty")
+        
         communicate = edge_tts.Communicate(
-            text, 
-            voice_short_name, 
-            rate=f"{rate:+d}%", 
-            pitch=f"{pitch:+d}Hz"
+            text,
+            voice.split(" - ")[0],
+            rate="+0%",
+            pitch="+0Hz"
         )
         with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file:
-            tmp_path = tmp_file.name
-            await communicate.save(tmp_path)
-        return tmp_path, None
+            await communicate.save(tmp_file.name)
+            return tmp_file.name
 
     def _create_output_directory(self) -> str:
+        """Crea directorio único para los archivos"""
         folder_name = base64.urlsafe_b64encode(os.urandom(8)).decode("utf-8")
         os.makedirs(folder_name, exist_ok=True)
         return folder_name
 
     def combine_audio_files(self, filenames: List[str]) -> AudioSegment:
+        """Combina segmentos de audio"""
         if not filenames:
-            raise ValueError("No input files provided")
+            raise ValueError("No audio files provided")
+        
         combined = AudioSegment.empty()
         for filename in filenames:
             combined += AudioSegment.from_file(filename, format="mp3")
         return combined
 
+    def _detect_silences(self, audio: AudioSegment, min_len: int = 500, thresh: int = -40) -> List[Tuple[int, int]]:
+        """Detecta intervalos de silencio en el audio"""
+        silent_ranges = []
+        start = None
+        
+        samples = np.array(audio.get_array_of_samples())
+        window_size = int(min_len * audio.frame_rate / 1000)
+        
+        for i in range(0, len(samples) - window_size, window_size):
+            window = samples[i:i+window_size]
+            if np.max(window) < thresh:
+                if start is None:
+                    start = i
+            else:
+                if start is not None:
+                    silent_ranges.append((start, i))
+                    start = None
+        
+        return silent_ranges
+
     def add_background_music_and_tags(
-        self, 
-        speech_audio: AudioSegment, 
-        music_file: str, 
-        tags_files: List[str]
+        self,
+        speech_audio: AudioSegment,
+        music_path: str,
+        tags_paths: List[str]
     ) -> AudioSegment:
-        music = AudioSegment.from_file(music_file)
+        """Mezcla música de fondo y tags inteligentemente"""
+        # 1. Cargar y ajustar música
+        music = AudioSegment.from_file(music_path).fade_out(2000)
+        music = music - 25  # Reducir volumen
+        
+        # 2. Loop inteligente (solo si es necesario)
         if len(music) < len(speech_audio):
-            music = music * (len(speech_audio) // len(music) + 1)
-        music = music[:len(speech_audio)] - 20
-        mixed = speech_audio.overlay(music)
-        for tag_path in tags_files:
-            tag_audio = AudioSegment.from_file(tag_path) - 5
-            mixed = tag_audio + mixed
-        return mixed
-
-    async def url_to_audio(self, url: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        text = self.fetch_text(url)
-        if len(words := text.split()) > 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"]
-        )
-        return await self._process_audio(conversation_json, voice_1, voice_2, conversation_text)
-
-    async def text_to_audio(self, structured_text: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        """Para texto YA estructurado como JSON de conversación."""
-        conversation_json = self.extract_conversation(structured_text)
-        conversation_text = "\n".join(
-            f"{turn['speaker']}: {turn['text']}" 
-            for turn in conversation_json["conversation"]
-        )
-        return await self._process_audio(conversation_json, voice_1, voice_2, conversation_text)
-
-    async def raw_text_to_audio(self, raw_text: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        """Para texto plano directo (sin estructura de diálogo)."""
-        fake_conversation = {"conversation": [{"speaker": "Narrador", "text": raw_text}]}
-        return await self._process_audio(fake_conversation, voice_1, voice_2, raw_text)
+            loops = (len(speech_audio) // len(music)) + 1
+            music = music * loops
+        music = music[:len(speech_audio)]
+        
+        # 3. Mezclar voz y música
+        mixed = speech_audio.overlay(music, position=0)
+        
+        # 4. Insertar tags
+        tag_intro = AudioSegment.from_file(tags_paths[0]) - 10
+        tag_transition = AudioSegment.from_file(tags_paths[1]) - 10
+        
+        # Tag inicial
+        final_audio = tag_intro + mixed
+        
+        # Tags en pausas (opcional)
+        silences = self._detect_silences(speech_audio)
+        for start, end in reversed(silences):
+            if (end - start) > len(tag_transition):
+                final_audio = final_audio.overlay(
+                    tag_transition,
+                    position=start + 100  # Pequeño margen
+                )
+        
+        return final_audio
 
-    async def _process_audio(
-        self, 
-        conversation_json: Dict, 
-        voice_1: str, 
-        voice_2: str, 
-        text: str
+    async def process_content(
+        self,
+        content: str,
+        voice_1: str,
+        voice_2: str,
+        is_url: bool = False
     ) -> Tuple[str, str]:
-        """Método interno para procesamiento común."""
-        audio_files, folder_name = await self.text_to_speech(conversation_json, voice_1, voice_2)
-        combined_audio = self.combine_audio_files(audio_files)
-        final_audio = self.add_background_music_and_tags(
-            combined_audio,
-            "musica.mp3",
-            ["tag.mp3", "tag2.mp3"]
-        )
-        output_file = os.path.join(folder_name, "output.mp3")
-        final_audio.export(output_file, format="mp3")
-        
-        for f in audio_files:
-            os.remove(f)
+        """Procesa contenido (URL o texto) a audio final"""
+        try:
+            # 1. Obtener texto estructurado
+            if is_url:
+                text = self.fetch_text(content)
+                if len(words := text.split()) > self.config.max_words:
+                    text = " ".join(words[:self.config.max_words])
+                conversation = self.extract_conversation(text)
+            else:
+                conversation = self.extract_conversation(content)
+            
+            # 2. Generar audio
+            audio_files, folder_name = await self.text_to_speech(conversation, voice_1, voice_2)
+            combined = self.combine_audio_files(audio_files)
+            
+            # 3. Mezclar con música y tags
+            final_audio = self.add_background_music_and_tags(
+                combined,
+                "musica.mp3",
+                ["tag.mp3", "tag2.mp3"]
+            )
+            
+            # 4. Exportar
+            output_path = os.path.join(folder_name, "podcast_final.mp3")
+            final_audio.export(output_path, format="mp3")
             
-        return output_file, text
+            # 5. Limpieza
+            for f in audio_files:
+                os.remove(f)
+            
+            # Texto de conversación
+            conversation_text = "\n".join(
+                f"{turn['speaker']}: {turn['text']}" 
+                for turn in conversation["conversation"]
+            )
+            
+            return output_path, conversation_text
+            
+        except Exception as e:
+            raise RuntimeError(f"Processing failed: {str(e)}")