Update conver.py

conver.py

@@ -1,15 +1,14 @@
 from dataclasses import dataclass
 from typing import List, Tuple, Dict
 import os
-import httpx
 import json
+import httpx
 from openai import OpenAI
 import edge_tts
 import tempfile
 from pydub import AudioSegment
 import base64
 from pathlib import Path
-import time
 
 @dataclass
 class ConversationConfig:
@@ -18,22 +17,14 @@ class ConversationConfig:
     model_name: str = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo"
 
 class URLToAudioConverter:
-    BASE_OUTPUT_DIR = "outputs"
-
     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
-        self._ensure_base_output_dir()
-
-    def _ensure_base_output_dir(self):
-        if not os.path.exists(self.BASE_OUTPUT_DIR):
-            os.makedirs(self.BASE_OUTPUT_DIR, exist_ok=True)
 
     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)
@@ -45,33 +36,27 @@ class URLToAudioConverter:
     def extract_conversation(self, text: str) -> Dict:
         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": "..."}]}'
             )
-
             chat_completion = 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('{'):
                 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:
             print(f"Error en extract_conversation: {str(e)}")
@@ -81,19 +66,15 @@ class URLToAudioConverter:
     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:
             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}")
@@ -103,110 +84,101 @@ class URLToAudioConverter:
             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:
-        # Crear carpeta única dentro de outputs/
         random_bytes = os.urandom(8)
-        folder_name = base64.urlsafe_b64encode(random_bytes).decode("utf-8").rstrip("=")
-        full_path = os.path.join(self.BASE_OUTPUT_DIR, f"podcast_{folder_name}")
-        os.makedirs(full_path, exist_ok=True)
-        return full_path
+        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:
+    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
-
-            combined.export(output_file, format="mp3")
-
-            # NO eliminar archivos aquí. Solo en limpieza periódica.
-
+            return combined
         except Exception as e:
             raise RuntimeError(f"Failed to combine audio files: {e}")
 
+    def add_background_music_and_tags(
+        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
+        mixed = speech_audio.overlay(music)
+        for i, tag_path in enumerate(tags_files):
+            tag_audio = AudioSegment.from_file(tag_path) - 5
+            if i == 0:
+                mixed = tag_audio + mixed
+            else:
+                mixed = mixed + tag_audio
+        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:
             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)
+        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 = "assets/musica.mp3"
+        tags_paths = ["assets/tag.mp3", "assets/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
 
     async def text_to_audio(self, text: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-        """Procesamiento normal con LLM"""
         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)
+        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 = "assets/musica.mp3"
+        tags_paths = ["assets/tag.mp3", "assets/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
 
     async def raw_text_to_audio(self, text: str, voice_1: str, voice_2: str) -> Tuple[str, str]:
-    """Modo sin LLM (texto directo), convierte el texto en conversación simple y genera audio."""
-    conversation = {
-        "conversation": [
-            {"speaker": "Host", "text": text},
-            {"speaker": "Co-host", "text": "(Continuación del tema)"}
-        ]
-    }
-    audio_files, folder_name = await self.text_to_speech(conversation, voice_1, voice_2)
-    output_file = os.path.join(folder_name, "raw_podcast.mp3")
-    self.combine_audio_files(audio_files, output_file)
-    # DEVUELVE (ruta_audio, texto) para que Gradio lo use bien
-    return output_file, text
-
-
-    def clean_old_files(self, max_age_seconds=86400):
-        """
-        Borra carpetas y archivos en BASE_OUTPUT_DIR que tengan más de max_age_seconds (por defecto 24h)
-        """
-        if not os.path.exists(self.BASE_OUTPUT_DIR):
-            return
-        now = time.time()
-        for folder in os.listdir(self.BASE_OUTPUT_DIR):
-            folder_path = os.path.join(self.BASE_OUTPUT_DIR, folder)
-            if os.path.isdir(folder_path):
-                try:
-                    mtime = os.path.getmtime(folder_path)
-                    if now - mtime > max_age_seconds:
-                        # Borramos carpeta completa
-                        for root, dirs, files in os.walk(folder_path, topdown=False):
-                            for name in files:
-                                os.remove(os.path.join(root, name))
-                            for name in dirs:
-                                os.rmdir(os.path.join(root, name))
-                        os.rmdir(folder_path)
-                except Exception:
-                    pass
+        conversation = {
+            "conversation": [
+                {"speaker": "Host", "text": text},
+                {"speaker": "Co-host", "text": "(Continuación del tema)"}
+            ]
+        }
+        audio_files, folder_name = await self.text_to_speech(conversation, voice_1, voice_2)
+        combined_audio = self.combine_audio_files(audio_files)
+        music_path = "assets/musica.mp3"
+        tags_paths = ["assets/tag.mp3", "assets/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.export(output_file, format="mp3")
+        for f in audio_files:
+            os.remove(f)
+        return text, output_file