Spaces:
Running
Running
| from dataclasses import dataclass | |
| from typing import List, Tuple, Dict | |
| import os | |
| 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 | |
| from threading import Thread | |
| class ConversationConfig: | |
| max_words: int = 3000 | |
| prefix_url: str = "https://r.jina.ai/" | |
| model_name: str = "meta-llama/Llama-3-8b-chat-hf" | |
| custom_prompt_template: str = None | |
| 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 | |
| self._start_cleaner() | |
| def _start_cleaner(self, max_age_hours: int = 24): | |
| def cleaner(): | |
| while True: | |
| now = time.time() | |
| for root, _, files in os.walk("."): | |
| for file in files: | |
| if file.endswith((".mp3", ".wav")): | |
| filepath = os.path.join(root, file) | |
| try: | |
| if now - os.path.getmtime(filepath) > max_age_hours * 3600: | |
| os.remove(filepath) | |
| except: | |
| pass | |
| time.sleep(3600) | |
| Thread(target=cleaner, daemon=True).start() | |
| 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 = self.config.custom_prompt_template.format(text=text) if self.config.custom_prompt_template else ( | |
| f"{text}\nConvierte el texto en un diálogo de podcast en español entre Anfitrión1 y Anfitrión2. " | |
| f"Genera una conversación extensa y natural con al menos 5 intercambios por hablante. " | |
| f"Devuelve SOLO un objeto JSON: " | |
| f'{{"conversation": [{{"speaker": "Anfitrión1", "text": "..."}}, {{"speaker": "Anfitrión2", "text": "..."}}]}}' | |
| ) | |
| print(f"Texto de entrada: {text[:200]}...") | |
| response = self.llm_client.chat.completions.create( | |
| messages=[{"role": "user", "content": prompt}], | |
| model=self.config.model_name, | |
| response_format={"type": "json_object"} | |
| ) | |
| response_content = response.choices[0].message.content | |
| print(f"Respuesta cruda del modelo: {response_content[:500]}...") | |
| 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] | |
| dialogue = json.loads(json_str) | |
| if not dialogue.get("conversation"): | |
| print("Error: No se generó diálogo válido.") | |
| return dialogue | |
| except Exception as e: | |
| raise RuntimeError(f"Failed to parse dialogue: {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: | |
| if not conversation_json["conversation"]: | |
| raise ValueError("No conversation data to process") | |
| for i, turn in enumerate(conversation_json["conversation"]): | |
| filename = output_dir / f"segment_{i}.mp3" | |
| voice = voice_1 if turn["speaker"] == "Anfitrión1" else voice_2 | |
| print(f"Generando audio para {turn['speaker']}: {turn['text'][:50]}... con voz {voice}") | |
| tmp_path = await self._generate_audio(turn["text"], voice) | |
| os.rename(tmp_path, filename) | |
| filenames.append(str(filename)) | |
| if not filenames: | |
| raise ValueError("No audio files generated") | |
| return filenames, str(output_dir) | |
| except Exception as e: | |
| raise RuntimeError(f"Text-to-speech failed: {e}") | |
| async def _generate_audio(self, text: str, voice: str) -> str: | |
| if not text.strip(): | |
| raise ValueError("Text cannot be empty") | |
| communicate = edge_tts.Communicate( | |
| text, | |
| voice.split(" - ")[0], | |
| rate="+0%", | |
| pitch="+0Hz" | |
| ) | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file: | |
| await communicate.save(tmp_file.name) | |
| return tmp_file.name | |
| def _create_output_directory(self) -> str: | |
| 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 audio files provided") | |
| 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_path: str, | |
| tags_paths: List[str], | |
| custom_music_path: str = None | |
| ) -> AudioSegment: | |
| music_file = custom_music_path if custom_music_path and os.path.exists(custom_music_path) else music_path | |
| music = AudioSegment.from_file(music_file).fade_out(2000) - 25 | |
| if len(music) < len(speech_audio): | |
| music = music * ((len(speech_audio) // len(music)) + 1) | |
| music = music[:len(speech_audio)] | |
| mixed = speech_audio.overlay(music) | |
| tag_outro = AudioSegment.from_file(tags_paths[0]) - 10 | |
| tag_trans = AudioSegment.from_file(tags_paths[1]) - 10 | |
| final_audio = mixed + tag_outro | |
| silent_ranges = [] | |
| for i in range(0, len(speech_audio) - 500, 100): | |
| chunk = speech_audio[i:i+500] | |
| if chunk.dBFS < -40: | |
| silent_ranges.append((i, i + 500)) | |
| for start, end in reversed(silent_ranges): | |
| if (end - start) >= len(tag_trans): | |
| final_audio = final_audio.overlay(tag_trans, position=start + 50) | |
| return final_audio | |
| async def url_to_audio(self, url: str, voice_1: str, voice_2: str, custom_music_path: str = None) -> 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 = self.extract_conversation(text) | |
| return await self._process_to_audio(conversation, voice_1, voice_2, custom_music_path) | |
| async def text_to_audio(self, text: str, voice_1: str, voice_2: str, custom_music_path: str = None) -> Tuple[str, str]: | |
| conversation = self.extract_conversation(text) | |
| return await self._process_to_audio(conversation, voice_1, voice_2, custom_music_path) | |
| async def raw_text_to_audio(self, text: str, voice_1: str, voice_2: str, custom_music_path: str = None) -> Tuple[str, str]: | |
| conversation = {"conversation": [{"speaker": "Anfitrión1", "text": text}]} | |
| return await self._process_to_audio(conversation, voice_1, voice_2, custom_music_path) | |
| async def _process_to_audio( | |
| self, | |
| conversation: Dict, | |
| voice_1: str, | |
| voice_2: str, | |
| custom_music_path: str = None | |
| ) -> Tuple[str, str]: | |
| audio_files, folder_name = await self.text_to_speech(conversation, voice_1, voice_2) | |
| combined = self.combine_audio_files(audio_files) | |
| final_audio = self.add_background_music_and_tags( | |
| combined, | |
| "musica.mp3", | |
| ["tag.mp3", "tag2.mp3"], | |
| custom_music_path | |
| ) | |
| output_path = os.path.join(folder_name, "podcast_final.mp3") | |
| final_audio.export(output_path, format="mp3") | |
| for f in audio_files: | |
| os.remove(f) | |
| text_output = "\n".join( | |
| f"{turn['speaker']}: {turn['text']}" | |
| for turn in conversation["conversation"] | |
| ) | |
| return output_path, text_output | |