app.py

# %%writefile /content/VibeVoice/demo/colab.py
# Original Code: https://github.com/microsoft/VibeVoice/blob/main/demo/gradio_demo.py
"""
VibeVoice Gradio Demo 
"""

import json
import os
import sys
import tempfile
import time
from pathlib import Path
from typing import List, Dict, Any, Iterator
from datetime import datetime
import threading
import numpy as np
import gradio as gr
import librosa
import soundfile as sf
import torch
import os
import traceback
import shutil
import re  
import uuid 

from vibevoice.modular.configuration_vibevoice import VibeVoiceConfig
from vibevoice.modular.modeling_vibevoice_inference import VibeVoiceForConditionalGenerationInference, VibeVoiceGenerationOutput
from vibevoice.processor.vibevoice_processor import VibeVoiceProcessor
from vibevoice.modular.streamer import AudioStreamer
from transformers import set_seed

from pydub import AudioSegment
from pydub.silence import split_on_silence

def drive_save(file_copy):
    drive_path = "/content/gdrive/MyDrive"
    save_folder = os.path.join(drive_path, "VibeVoice_Podcast")

    if os.path.exists(drive_path):
        print("Running on Google Colab and auto-saving to Google Drive...")
        os.makedirs(save_folder, exist_ok=True)
        dest_path = os.path.join(save_folder, os.path.basename(file_copy))
        shutil.copy2(file_copy, dest_path)  
        print(f"File saved to: {dest_path}")
        return dest_path
    else:
        print("Not running on Google Colab (or Google Drive not mounted). Skipping auto-save.")
        return None

import os, requests, urllib.request, urllib.error
from tqdm.auto import tqdm

def download_file(url, download_file_path, redownload=False):
    base_path = os.path.dirname(download_file_path)
    os.makedirs(base_path, exist_ok=True)
    if os.path.exists(download_file_path):
        if redownload:
            os.remove(download_file_path)
            tqdm.write(f"♻️ Redownloading: {os.path.basename(download_file_path)}")
        elif os.path.getsize(download_file_path) > 0:
            tqdm.write(f"✔️ Skipped (already exists): {os.path.basename(download_file_path)}")
            return True
    try:
        request = urllib.request.urlopen(url)
        total = int(request.headers.get('Content-Length', 0))
    except urllib.error.URLError as e:
        print(f"❌ Error: Unable to open URL: {url}")
        print(f"Reason: {e.reason}")
        return False
    with tqdm(total=total, desc=os.path.basename(download_file_path), unit='B', unit_scale=True, unit_divisor=1024) as progress:
        try:
            urllib.request.urlretrieve(
                url, 
                download_file_path,
                reporthook=lambda count, block_size, total_size: progress.update(block_size)
            )
        except urllib.error.URLError as e:
            print(f"❌ Error: Failed to download {url}")
            print(f"Reason: {e.reason}")
            return False
    tqdm.write(f"⬇️ Downloaded: {os.path.basename(download_file_path)}")
    return True

def download_model(repo_id, download_folder="./", redownload=False):
    if not download_folder.strip():
        download_folder = "."
    url = f"https://huggingface.co/api/models/{repo_id}"
    download_dir = os.path.abspath(f"{download_folder.rstrip('/')}/{repo_id.split('/')[-1]}")
    os.makedirs(download_dir, exist_ok=True)
    print(f"📂 Download directory: {download_dir}")
    response = requests.get(url)
    if response.status_code != 200:
        print("❌ Error:", response.status_code, response.text)
        return None
    data = response.json()
    siblings = data.get("siblings", [])
    files = [f["rfilename"] for f in siblings]
    print(f"📦 Found {len(files)} files in repo '{repo_id}'. Checking cache ...")
    for file in tqdm(files, desc="Processing files", unit="file"):
        file_url = f"https://huggingface.co/{repo_id}/resolve/main/{file}"
        file_path = os.path.join(download_dir, file)
        download_file(file_url, file_path, redownload=redownload)
    return download_dir

def generate_file_name(text):
    output_dir = "./podcast_audio"
    os.makedirs(output_dir, exist_ok=True)
    cleaned = re.sub(r"^\s*speaker\s*\d+\s*:\s*", "", text, flags=re.IGNORECASE)
    short = cleaned[:30].strip()
    short = re.sub(r'[^a-zA-Z0-9\s]', '', short)
    short = short.lower().strip().replace(" ", "_")
    if not short:
        short = "podcast_output"
    unique_name = f"{short}_{uuid.uuid4().hex[:6]}"
    return os.path.join(output_dir, unique_name)

class VibeVoiceDemo:
    def __init__(self, model_path: str, device: str = "cuda", inference_steps: int = 5):
        self.model_path = model_path
        self.device = device
        self.inference_steps = inference_steps
        self.is_generating = False  
        self.stop_generation = False  
        self.load_model()
        self.setup_voice_presets()
        self.load_example_scripts()  

    def load_model(self):
        print(f"Loading processor & model from {self.model_path}")
        self.processor = VibeVoiceProcessor.from_pretrained(self.model_path) 
        if self.device == "cuda":
            self.model = VibeVoiceForConditionalGenerationInference.from_pretrained(
                self.model_path,
                torch_dtype=torch.bfloat16,
                device_map=self.device,
            )
        else:
            self.model = VibeVoiceForConditionalGenerationInference.from_pretrained(
                self.model_path,
                torch_dtype=torch.float32, 
                device_map="cpu",
            )
        self.model.eval()
        self.model.model.noise_scheduler = self.model.model.noise_scheduler.from_config(
            self.model.model.noise_scheduler.config,
            algorithm_type='sde-dpmsolver++',
            beta_schedule='squaredcos_cap_v2'
        )
        self.model.set_ddpm_inference_steps(num_steps=self.inference_steps)
        if hasattr(self.model.model, 'language_model'):
            print(f"Language model attention: {self.model.model.language_model.config._attn_implementation}")

    def setup_voice_presets(self):
        voices_dir = os.path.join(os.path.dirname(__file__), "voices")
        if not os.path.exists(voices_dir):
            print(f"Warning: Voices directory not found at {voices_dir}, creating it.")
            os.makedirs(voices_dir, exist_ok=True)
        self.voice_presets = {}
        audio_files = [f for f in os.listdir(voices_dir) if f.lower().endswith(('.wav', '.mp3', '.flac', '.ogg', '.m4a', '.aac')) and os.path.isfile(os.path.join(voices_dir, f))]
        for audio_file in audio_files:
            name = os.path.splitext(audio_file)[0]
            self.voice_presets[name] = os.path.join(voices_dir, audio_file)
        self.voice_presets = dict(sorted(self.voice_presets.items()))
        self.available_voices = {name: path for name, path in self.voice_presets.items() if os.path.exists(path)}
        if not self.available_voices: print("Warning: No voice presets found.")
        print(f"Found {len(self.available_voices)} voice files in {voices_dir}")

    def read_audio(self, audio_path: str, target_sr: int = 24000) -> np.ndarray:
        try:
            wav, sr = sf.read(audio_path)
            if len(wav.shape) > 1: wav = np.mean(wav, axis=1)
            if sr != target_sr: wav = librosa.resample(wav, orig_sr=sr, target_sr=target_sr)
            return wav
        except Exception as e:
            print(f"Error reading audio {audio_path}: {e}")
            return np.array([])

    def trim_silence_from_numpy(self, audio_np: np.ndarray, sample_rate: int, silence_thresh: int = -45, min_silence_len: int = 100, keep_silence: int = 50) -> np.ndarray:
        audio_int16 = (audio_np * 32767).astype(np.int16)
        sound = AudioSegment(data=audio_int16.tobytes(), sample_width=audio_int16.dtype.itemsize, frame_rate=sample_rate, channels=1)
        audio_chunks = split_on_silence(sound, min_silence_len=min_silence_len, silence_thresh=silence_thresh, keep_silence=keep_silence)
        if not audio_chunks: return np.array([0.0], dtype=np.float32)
        combined = sum(audio_chunks)
        samples = np.array(combined.get_array_of_samples())
        return samples.astype(np.float32) / 32767.0

    def generate_podcast_with_timestamps(self,
                                 num_speakers: int,
                                 script: str,
                                 speaker_1: str, speaker_2: str, speaker_3: str, speaker_4: str,
                                 cfg_scale: float,
                                 remove_silence: bool,
                                 progress=gr.Progress()):
        
        # Initial UI state: Clear previous results, show stop button
        yield None, None, None, gr.update(visible=False), gr.update(visible=True)

        final_audio_path, final_json_path = None, None
        try:
            self.stop_generation = False
            self.is_generating = True
            
            if not script.strip(): raise gr.Error("Error: Please provide a script.")
            script = script.replace("’", "'")
            if not 1 <= num_speakers <= 4: raise gr.Error("Error: Number of speakers must be between 1 and 4.")
            selected_speakers = [speaker_1, speaker_2, speaker_3, speaker_4][:num_speakers]
            for i, speaker in enumerate(selected_speakers):
                if not speaker or speaker not in self.available_voices:
                    raise gr.Error(f"Error: Please select a valid speaker for Speaker {i+1}.")
            voice_samples = [self.read_audio(self.available_voices[name]) for name in selected_speakers]
            if any(len(vs) == 0 for vs in voice_samples): raise gr.Error("Error: Failed to load one or more audio files.")

            lines = script.strip().split('\n')
            formatted_script_lines = []
            for line in lines:
                line = line.strip()
                if not line: continue
                if re.match(r'Speaker\s*\d+:', line, re.IGNORECASE):
                    formatted_script_lines.append(line)
                else:
                    speaker_id = len(formatted_script_lines) % num_speakers
                    formatted_script_lines.append(f"Speaker {speaker_id+1}: {line}")

            if not formatted_script_lines: raise gr.Error("Error: Script is empty after formatting.")

            timestamps = {}
            current_time = 0.0
            sample_rate = 24000
            
            base_filename = generate_file_name(formatted_script_lines[0])
            final_audio_path = base_filename + ".wav"
            final_json_path = base_filename + ".json"

            with sf.SoundFile(final_audio_path, 'w', samplerate=sample_rate, channels=1, subtype='PCM_16') as audio_file:
                for i, line in enumerate(formatted_script_lines):
                    if self.stop_generation:
                        print("\n🚫 Generation interrupted by user. Finalizing partial files...")
                        break
                    progress(i / len(formatted_script_lines), desc=f"Generating line {i+1}/{len(formatted_script_lines)}")
                    match = re.match(r'Speaker\s*(\d+):\s*(.*)', line, re.IGNORECASE)
                    if not match: continue
                    speaker_idx = int(match.group(1)) - 1
                    text_content = match.group(2).strip()
                    if not (0 <= speaker_idx < len(voice_samples)): continue
                    
                    inputs = self.processor(text=[line], voice_samples=[voice_samples[speaker_idx]], padding=True, return_tensors="pt")
                    output_waveform = self.model.generate(**inputs, max_new_tokens=None, cfg_scale=cfg_scale, tokenizer=self.processor.tokenizer, generation_config={'do_sample': False}, verbose=False, refresh_negative=True)
                    audio_np = output_waveform.speech_outputs[0].cpu().float().numpy().squeeze()
                    
                    if remove_silence: audio_np = self.trim_silence_from_numpy(audio_np, sample_rate)
                    duration = len(audio_np) / sample_rate
                    audio_file.write((audio_np * 32767).astype(np.int16))
                    timestamps[str(i + 1)] = {"text": text_content, "speaker_id": speaker_idx + 1, "start": current_time, "end": current_time + duration}
                    current_time += duration

            if not timestamps:
                self.is_generating = False
                if os.path.exists(final_audio_path): os.remove(final_audio_path)
                yield None, None, None, gr.update(visible=True), gr.update(visible=False)
                return

            progress(1.0, desc="Saving generated files...")
            with open(final_json_path, "w") as f: json.dump(timestamps, f, indent=2)
            try:
                drive_save(final_audio_path)
                drive_save(final_json_path)
            except Exception as e: print(f"Error saving files to Google Drive: {e}")

            message = "Partial" if self.stop_generation else "Full"
            print(f"\n✨ {message} generation successful!\n🎵 Audio: {final_audio_path}\n📄 Timestamps: {final_json_path}\n")
            
            self.is_generating = False
            yield final_audio_path, final_audio_path, final_json_path, gr.update(visible=True), gr.update(visible=False)

        except Exception as e:
            self.is_generating = False
            print(f"❌ An unexpected error occurred: {str(e)}")
            traceback.print_exc()
            try:
                if final_audio_path and os.path.exists(final_audio_path): os.remove(final_audio_path)
                if final_json_path and os.path.exists(final_json_path): os.remove(final_json_path)
            except Exception as cleanup_e: print(f"Error during cleanup after exception: {cleanup_e}")
            yield None, None, None, gr.update(visible=True), gr.update(visible=False)

    def stop_audio_generation(self):
        if self.is_generating:
            self.stop_generation = True
            print("🛑 Audio generation stop requested")

    def load_example_scripts(self):
        examples_dir = os.path.join(os.path.dirname(__file__), "text_examples")
        self.example_scripts = []
        if not os.path.exists(examples_dir): return
        txt_files = sorted([f for f in os.listdir(examples_dir) if f.lower().endswith('.txt')])
        for txt_file in txt_files:
            try:
                with open(os.path.join(examples_dir, txt_file), 'r', encoding='utf-8') as f:
                    script = f.read().strip()
                if script: self.example_scripts.append([self._get_num_speakers_from_script(script), script])
            except Exception as e: print(f"Error loading example {txt_file}: {e}")

    def _get_num_speakers_from_script(self, script: str) -> int:
        speakers = set(re.findall(r'^Speaker\s+(\d+)\s*:', script, re.MULTILINE | re.IGNORECASE))
        return max(int(s) for s in speakers) if speakers else 1

def create_demo_interface(demo_instance: VibeVoiceDemo):
    with gr.Blocks(title="VibeVoice AI Podcast Generator") as interface:
        gr.HTML("""
        <div style="text-align: center; margin: 20px auto; max-width: 800px;">
            <h1 style="font-size: 2.5em; margin-bottom: 10px;">🎙️ Vibe Podcasting</h1>
            <p style="font-size: 1.2em; color: #555; margin-bottom: 15px;">Generate Long-form Multi-speaker AI Podcasts with VibeVoice</p>
            <a href="https://github.com/NeuralFalconYT/AI-Podcast-Generator" target="_blank" style="display: inline-block; padding: 10px 20px; background-color: #4285F4; color: white; border-radius: 6px; text-decoration: none; font-size: 1em;">🥳 Run on Google Colab</a>
        </div>
        """)
        with gr.Row():
            with gr.Column(scale=1):
                with gr.Group():
                    gr.Markdown("### 🎛️ Podcast Settings")
                    num_speakers = gr.Slider(minimum=1, maximum=4, value=2, step=1, label="Number of Speakers")
                    gr.Markdown("### 🎭 Speaker Selection")
                    speaker_selections = []
                    available_voices = list(demo_instance.available_voices.keys())
                    defaults = ['en-Alice_woman', 'en-Carter_man', 'en-Frank_man', 'en-Maya_woman']
                    for i in range(4):
                        val = defaults[i] if i < len(defaults) and defaults[i] in available_voices else None
                        speaker = gr.Dropdown(choices=available_voices, value=val, label=f"Speaker {i+1}", visible=(i < 2))
                        speaker_selections.append(speaker)
                    with gr.Accordion("🎤 Upload Custom Voices", open=False):
                        upload_audio = gr.File(label="Upload Voice Samples", file_count="multiple", file_types=["audio"])
                        process_upload_btn = gr.Button("Add Uploaded Voices to Speaker Selection")
                    with gr.Accordion("⚙️ Advanced Settings", open=False):
                        cfg_scale = gr.Slider(minimum=1.0, maximum=2.0, value=1.3, step=0.05, label="CFG Scale")
                        remove_silence_checkbox = gr.Checkbox(label="Trim Silence from Podcast", value=False,)
            with gr.Column(scale=2):
                with gr.Group():
                    gr.Markdown("### 📝 Script Input")
                    script_input = gr.Textbox(
                          label="Conversation Script",
                          placeholder="Speaker 1: Hi everyone, I’m Alex, and welcome back.\nSpeaker 2: And I’m lisa. Thanks for tuning in.",
                          lines=10
                      )
                    with gr.Row():
                        random_example_btn = gr.Button("🎲 Random Example", scale=1)
                        generate_btn = gr.Button("🚀 Generate Podcast", variant="primary", scale=2)
                    stop_btn = gr.Button("🛑 Stop Generation", variant="stop", visible=False)
                    gr.Markdown("### 🎵 **Generated Output**")
                    audio_output = gr.Audio(label="Play Generated Podcast")
                    with gr.Accordion("📦 Download Files", open=False):
                      download_file = gr.File(label="Download Audio File (.wav)")
                      json_file_output = gr.File(label="Download Timestamps (.json)")

        with gr.Accordion("💡 Usage Tips & Examples", open=False):
            gr.Markdown("""- **Upload Your Own Voices:** Create your own podcast with custom voice samples.  \n- **Timestamps:** Useful if you want to generate a video using Wan2.2 or other tools. The timestamps let you automatically separate each speaker (splitting the long podcast into smaller chunks), pass the audio clips to your video generation model, and then merge the generated video clips into a full podcast video (e.g., using FFmpeg + any video generation model such as image+audio → video).""")
            gr.Examples(examples=demo_instance.example_scripts, inputs=[num_speakers, script_input], label="Try these example scripts:")

        def process_and_refresh_voices(uploaded_files):
            if not uploaded_files: return [gr.update() for _ in speaker_selections] + [None]
            voices_dir = os.path.join(os.path.dirname(__file__), "voices")
            for f in uploaded_files: shutil.copy(f.name, os.path.join(voices_dir, os.path.basename(f.name)))
            demo_instance.setup_voice_presets()
            new_choices = list(demo_instance.available_voices.keys())
            return [gr.update(choices=new_choices) for _ in speaker_selections] + [None]

        def update_speaker_visibility(num):
            return [gr.update(visible=(i < int(num))) for i in range(4)]

        num_speakers.change(fn=update_speaker_visibility, inputs=num_speakers, outputs=speaker_selections)
        process_upload_btn.click(fn=process_and_refresh_voices, inputs=upload_audio, outputs=speaker_selections + [upload_audio])

        generate_btn.click(
            fn=demo_instance.generate_podcast_with_timestamps,
            inputs=[num_speakers, script_input] + speaker_selections + [cfg_scale, remove_silence_checkbox],
            outputs=[audio_output, download_file, json_file_output, generate_btn, stop_btn],
        )
        
        stop_btn.click(
            fn=demo_instance.stop_audio_generation
        )

        def load_random_example():
            import random
            return random.choice(demo_instance.example_scripts) if demo_instance.example_scripts else (2, "Speaker 0: No examples loaded.")
        random_example_btn.click(fn=load_random_example, outputs=[num_speakers, script_input])

    return interface




def build_conversation_prompt(topic, *speaker_names):
    """
    Generates the final prompt. It takes the topic and a variable number of speaker names.
    """
    names = [name for name in speaker_names if name and name.strip()]

    # Error checking
    if not topic or not topic.strip():
        return "Error: Please provide a topic."
    if not names:
        return "Error: Please provide at least one speaker name."

    num_speakers = len(names)
    speaker_mapping_str = "Speaker mapping (for context only, DO NOT use these names as labels):\n"
    for i, name in enumerate(names):
        speaker_mapping_str += f"- Speaker {i+1} = {name}\n"
    
    speaker_labels = [f"\"Speaker {i+1}:\"" for i in range(num_speakers)]
    
    introductions_str = ""
    for i, name in enumerate(names):
        introductions_str += f"  - Speaker {i+1} introduces themselves by saying: \"I’m {name}...\"\n"
        
    example_str = "STRICT Example (follow this format exactly):\n"
    example_str += f"Speaker 1: Hi everyone, I’m {names[0]}, and I’m excited to be here today.\n"
    if num_speakers > 1:
        for i in range(1, num_speakers):
            example_str += f"Speaker {i+1}: And I’m {names[i]}. Thanks for joining us.\n"
    example_str += "Speaker 1: So, let’s dive into our topic...\n"
    
    prompt = f"""
You are a professional podcast scriptwriter. 
Write a natural, engaging conversation between {num_speakers} speakers on the topic: "{topic}".
{speaker_mapping_str}
Formatting Rules:
- You MUST always format dialogue with {', '.join(speaker_labels)} ONLY. 
- Never replace the labels with real names. The labels stay exactly as they are.
- At the beginning:
{introductions_str}
- During the conversation, they may occasionally mention each other's names ({', '.join(names)}) naturally in the dialogue, but the labels must remain unchanged.
- Do not add narration, descriptions, or any extra formatting.
{example_str}
"""
    return prompt

def update_speaker_name_visibility(num_speakers):
    """
    Shows or hides the speaker name textboxes based on the slider value.
    """
    num = int(num_speakers)
    updates = []
    for i in range(4):
        if i < num:
            updates.append(gr.update(visible=True))
        else:
            updates.append(gr.update(visible=False, value=""))
    
    return tuple(updates) 

def ui2():

    with gr.Blocks(title="Prompt Builder") as demo:
        gr.HTML("""
        <div style="text-align: center; margin: 20px auto; max-width: 800px;">
            <h1 style="font-size: 2.5em; margin-bottom: 5px;">🎙️ Sample Podcast Prompt Generator</h1>
            <p style="font-size: 1.2em; color: #555;">Paste the prompt into any LLM, and customize the propmt if you want.</p>
        </div>""")
        
        with gr.Row():
            with gr.Column(scale=1):
                topic = gr.Textbox(label="Topic", placeholder="e.g., The Future of Artificial Intelligence")
                
                num_speakers = gr.Slider(
                    minimum=1, 
                    maximum=4, 
                    value=2, 
                    step=1, 
                    label="Number of Speakers"
                )
                
                with gr.Group():
                    speaker_textboxes = [
                        gr.Textbox(label=f"Speaker {i+1} Name", visible=(i < 2), placeholder=f"e.g., Speaker {i+1}")
                        for i in range(4)
                    ]
                
                gen_btn = gr.Button("Generate Prompt", variant="primary")


                gr.Examples(
                    examples=[
                        ["The Ethics of Gene Editing", 2, "Dr. Evelyn Reed", "Dr. Ben Carter", "", ""],
                        ["Exploring the Deep Sea", 3, "Maria", "Leo", "Samira", ""],
                        ["The Future of Space Tourism", 4, "Alex", "Zara", "Kenji", "Isla"]
                    ],
                    # The inputs list must match the order of items in the examples list
                    inputs=[topic, num_speakers] + speaker_textboxes,
                    label="Quick Examples"
                )

            with gr.Column(scale=2):
                output_prompt = gr.Textbox(label="Generated Prompt", lines=25, interactive=False, show_copy_button=True)

        
        num_speakers.change(
            fn=update_speaker_name_visibility, 
            inputs=num_speakers, 
            outputs=speaker_textboxes
        )
        
        gen_btn.click(
            fn=build_conversation_prompt, 
            inputs=[topic] + speaker_textboxes, 
            outputs=[output_prompt]
        )

    return demo



import click
@click.command()
@click.option("--model_path", default="microsoft/VibeVoice-1.5B", help="Hugging Face Model Repo ID.")
@click.option("--inference_steps", default=10, show_default=True, type=int, help="Number of inference steps for generation.")
@click.option("--debug", is_flag=True, default=False, help="Enable debug mode.")
@click.option("--share", is_flag=True, default=False, help="Enable sharing of the interface.")
def main(model_path, inference_steps, debug, share):
    # model_folder = download_model(model_path, download_folder="./", redownload=False)
    model_folder=model_path
    device = "cuda" if torch.cuda.is_available() else "cpu"
    set_seed(42)
    print("🎙️ Initializing VibeVoice ...")
    demo_instance = VibeVoiceDemo(model_path=model_folder, device=device, inference_steps=inference_steps)
    custom_css = """.gradio-container { font-family: 'SF Pro Display', -apple-system, BlinkMacSystemFont, sans-serif; }"""
    demo1 = create_demo_interface(demo_instance)
    demo2 = ui2()
    demo = gr.TabbedInterface([demo1, demo2],["Vibe Podcasting","Generate Sample Podcast Script"],title="",theme=gr.themes.Soft(),css=custom_css)
    print("🚀 Launching Gradio Demo...")
    demo.queue().launch(debug=debug, share=share)

if __name__ == "__main__":
    main()