Update conver.py

conver.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List, Tuple, Dict
+from typing import List, Tuple, Dict, Optional
 import os
 import json
 import httpx
@@ -50,17 +50,11 @@ class URLToAudioConverter:
             response_content = chat_completion.choices[0].message.content
             json_str = response_content.strip()
             if not json_str.startswith('{'):
-                start = json_str.find('{')
-                if start != -1:
-                    json_str = json_str[start:]
+                json_str = json_str[json_str.find('{'):]
             if not json_str.endswith('}'):
-                end = json_str.rfind('}')
-                if end != -1:
-                    json_str = json_str[:end+1]
+                json_str = json_str[:json_str.rfind('}')+1]
             return json.loads(json_str)
         except Exception as e:
-            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]:
@@ -79,105 +73,94 @@ class URLToAudioConverter:
         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]:
+    async def _generate_audio(self, text: str, voice: str, rate: int = 0, pitch: int = 0) -> Tuple[str, Optional[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)
+        communicate = edge_tts.Communicate(
+            text, 
+            voice_short_name, 
+            rate=f"{rate:+d}%", 
+            pitch=f"{pitch:+d}Hz"
+        )
         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")
+        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:
         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
-            return combined
-        except Exception as e:
-            raise RuntimeError(f"Failed to combine audio files: {e}")
+        combined = AudioSegment.empty()
+        for filename in filenames:
+            combined += AudioSegment.from_file(filename, format="mp3")
+        return combined
 
     def add_background_music_and_tags(
-        self, speech_audio: AudioSegment, music_file: str, tags_files: List[str]
+        self, 
+        speech_audio: AudioSegment, 
+        music_file: str, 
+        tags_files: List[str]
     ) -> AudioSegment:
         music = AudioSegment.from_file(music_file)
         if len(music) < len(speech_audio):
-            loops = (len(speech_audio) // len(music)) + 1
-            music = music * loops
-        music = music[:len(speech_audio)] - 20  # bajar volumen música
+            music = music * (len(speech_audio) // len(music) + 1)
+        music = music[:len(speech_audio)] - 20
         mixed = speech_audio.overlay(music)
-        for i, tag_path in enumerate(tags_files):
+        for tag_path in tags_files:
             tag_audio = AudioSegment.from_file(tag_path) - 5
-            if i == 0:
-                mixed = tag_audio + mixed
-            else:
-                mixed = mixed + tag_audio
+            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)
-        words = text.split()
-        if len(words) > self.config.max_words:
+        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"]
+            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)
-        combined_audio = self.combine_audio_files(audio_files)
-        music_path = "musica.mp3"
-        tags_paths = ["tag.mp3", "tag2.mp3"]
-        final_audio = self.add_background_music_and_tags(combined_audio, music_path, tags_paths)
-        final_output = os.path.join(folder_name, "combined_output_with_music.mp3")
-        final_audio.export(final_output, format="mp3")
-        for f in audio_files:
-            os.remove(f)
-        return final_output, conversation_text
+        return await self._process_audio(conversation_json, voice_1, voice_2, conversation_text)
 
-    async def text_to_audio(self, text: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        conversation_json = self.extract_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"]
+            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)
-        combined_audio = self.combine_audio_files(audio_files)
-        music_path = "musica.mp3"
-        tags_paths = ["tag.mp3", "tag2.mp3"]
-        final_audio = self.add_background_music_and_tags(combined_audio, music_path, tags_paths)
-        final_output = os.path.join(folder_name, "combined_output_with_music.mp3")
-        final_audio.export(final_output, format="mp3")
-        for f in audio_files:
-            os.remove(f)
-        return final_output, conversation_text
+        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)
 
-    async def generate_podcast(self, conversation, voice_1, voice_2):
-        audio_files, folder_name = await self.text_to_speech(conversation, voice_1, voice_2)
+    async def _process_audio(
+        self, 
+        conversation_json: Dict, 
+        voice_1: str, 
+        voice_2: str, 
+        text: str
+    ) -> 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)
-        music_path = "musica.mp3"
-        tags_paths = ["tag.mp3", "tag2.mp3"]
-        final_audio = self.add_background_music_and_tags(combined_audio, music_path, tags_paths)
-        output_file = os.path.join(folder_name, "raw_podcast_with_music.mp3")
+        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)
             
-        conversation_text = "\n".join(
-            f"{turn['speaker']}: {turn['text']}" for turn in conversation["conversation"]
-        )
-        return conversation_text, output_file
+        return output_file, text