app.py

import json
import asyncio
import edge_tts
from pydub import AudioSegment
import os
import gradio as gr
from gradio_client import Client
import shutil
import uuid
from dotenv import load_dotenv
import re

load_dotenv()

def sanitize_filename(filename):
    """Convert a string to a safe filename by removing special characters and spaces"""
    safe_filename = re.sub(r'[^a-zA-Z0-9_-]', '', filename.replace(' ', '_'))
    return safe_filename.lower()[:50]

async def get_voices():
    """Get all available English voices from edge-tts"""
    voices = await edge_tts.list_voices()
    english_voices = [
        voice for voice in voices 
        if voice["Locale"].startswith(("en-US", "en-GB", "en-AU", "en-CA", "en-IN"))
    ]
    
    formatted_voices = [
        f"{voice['ShortName']} ({voice['Gender']}, {voice['Locale']})"
        for voice in english_voices
    ]
    return formatted_voices

def extract_voice_name(voice_string):
    """Extract the voice short name from the formatted string"""
    return voice_string.split(" (")[0]

async def generate_audio(text, voice, filename):
    communicate = edge_tts.Communicate(text, extract_voice_name(voice))
    await communicate.save(filename)

async def create_podcast_version(data, speaker_name, speaker_voice, title):
    session_id = str(uuid.uuid4())
    temp_dir = f'temp_{session_id}'
    safe_title = sanitize_filename(title)
    
    if not os.path.exists(temp_dir):
        os.makedirs(temp_dir)
    
    try:
        speaker_version = AudioSegment.empty()
        
        for i, entry in enumerate(data['conversation']):
            if 'speakertext' in entry:
                temp_file = f'{temp_dir}/speaker_{i}.mp3'
                await generate_audio(entry['speakertext'], speaker_voice, temp_file)
                audio = AudioSegment.from_file(temp_file)
                speaker_version += audio
                os.remove(temp_file)
        
        speaker_path = f"{safe_title}_{speaker_name.lower()}_only.mp3"
        speaker_version.export(speaker_path, format="mp3")
        
        return speaker_path, temp_dir
    except Exception as e:
        if os.path.exists(temp_dir):
            shutil.rmtree(temp_dir)
        raise e

def generate_podcast(title, channel_name, speaker_name, speaker_voice):
    try:
        if not all([title, channel_name, speaker_name, speaker_voice]):
            raise ValueError("All fields must be filled out")
        
        client = Client(os.getenv('API_URL'))
        result = client.predict(
            message=f"""{os.getenv('API_MESSAGE')} {{
              "title": "{title}",
              "channel": "{channel_name}",
              "speaker": "{speaker_name}",
              "conversation": [
                {{
                  "speakertext": ""
                }}
              ]
            }}
            
            give 36 sentences for the speaker.
            """,
            request=os.getenv('API_REQUEST'),
            param_3=0.5,
            param_4=8100,
            param_5=0.5,
            param_6=0,
            api_name="/chat"
        )

        try:
            podcast_data = json.loads(result)
        except json.JSONDecodeError:
            json_start = result.find('```') + 3
            json_end = result.rfind('```')
            
            if json_start > 2 and json_end > json_start:
                if result[json_start:json_start+4] == 'json':
                    json_start = result.find('\n', json_start) + 1
                json_str = result[json_start:json_end].strip()
                podcast_data = json.loads(json_str)
            else:
                raise ValueError("Could not parse JSON from response")

        speaker_path, temp_dir = asyncio.run(
            create_podcast_version(
                podcast_data, 
                speaker_name,
                speaker_voice,
                title
            )
        )
        
        if os.path.exists(temp_dir):
            shutil.rmtree(temp_dir)
        
        return [speaker_path, podcast_data]
    except Exception as e:
        return [None, f"Error: {str(e)}"]

with gr.Blocks(theme=gr.themes.Soft()) as interface:
    available_voices = asyncio.run(get_voices())
    
    gr.Markdown("# Easy Podcast (Single Speaker)")
    gr.Markdown("Generate a podcast with one speaker. Choose a voice and customize details to create your perfect audio.<br>To use elevelabs voices or cloned voices, or to automate the podcast video creation with avatar contact me at aheedsajid@gmail.com<br>Support me USDT (TRC-20) (TAe7hsSVWtMEYz3G5V1UiUdYPQVqm28bKx)")
    
    with gr.Row():
        with gr.Column():
            title = gr.Textbox(
                label="Podcast Topic",
                placeholder="e.g., The Future of AI",
                show_label=True
            )
            channel_name = gr.Textbox(
                label="Channel Name",
                placeholder="e.g., TechTalks",
                value="WeePakistan",
                show_label=True
            )
        with gr.Column():
            speaker_name = gr.Textbox(
                label="Speaker Name",
                placeholder="e.g., John",
                value="Andrew",
                show_label=True
            )
    
    with gr.Row():
        speaker_voice = gr.Dropdown(
            choices=available_voices,
            value=next((v for v in available_voices if "Andrew" in v), available_voices[0]),
            label="Speaker Voice",
            info="Select voice for the speaker"
        )
    
    generate_btn = gr.Button("Generate Podcast", variant="primary")
    
    with gr.Row():
        speaker_audio = gr.Audio(label="Speaker Audio")
    
    conversation_json = gr.JSON(label="Generated Conversation")
    
    generate_btn.click(
        fn=generate_podcast,
        inputs=[title, channel_name, speaker_name, speaker_voice],
        outputs=[speaker_audio, conversation_json]
    )

if __name__ == "__main__":
    interface.launch()