app.py

import traceback
import gradio as gr
import requests
import time
import os

# Read from environment variables with default values if not set
LETTERCAST_API_CAST_TEMPLATE = (
    f'{os.environ.get("LETTERCAST_API_CAST")}/feeds/{{feed_id}}/casts'
)
TTS_BACKEND = "openai-tts-1-hd"  # Default TTS backend
WAIT_TIME_S = 30


def create_cast(token, feed_id, url, file, detail_level, create_video, hosts):
    if not url and file is None:
        return {"error": "Please provide a URL or upload a file."}, None, None
    if not token or not feed_id:
        return {"error": "Please provide both Token and Feed ID."}, None, None
    headers = {"Authorization": f"Bearer {token}"}
    data = {
        "hosts": hosts,
        "detail_level": detail_level,
        "create_video": str(create_video).lower(),
        "tts_backend": TTS_BACKEND,
    }
    if url:
        data["url"] = url
    endpoint_url = LETTERCAST_API_CAST_TEMPLATE.format(feed_id=feed_id)

    if file is not None:
        with open(file, "rb") as f:
            files = {"file": (os.path.basename(file), f, "application/pdf")}
            response = requests.post(
                endpoint_url, headers=headers, data=data, files=files
            )
    else:
        response = requests.post(endpoint_url, headers=headers, data=data)

    if response.status_code == 200:
        resp_data = response.json()
        audio_file_url = resp_data.get("audio_file_url")
        video_file_url = resp_data.get("video_file_url")
        return resp_data, audio_file_url, video_file_url
    else:
        error_message = f"Error {response.status_code}: {response.text}"
        return {"error": error_message}, None, None


def poll_until_ready(
    audio_file_url, video_file_url, create_video, progress=gr.Progress()
):
    max_retries = 60  # Max wait time of 5 minutes (assuming 5s per retry)
    wait_time = WAIT_TIME_S  # seconds
    audio_ready = False
    video_ready = False
    for i in range(max_retries):
        progress(i / max_retries)
        if audio_file_url and not audio_ready:
            try:
                response = requests.get(audio_file_url)
                if response.status_code == 200:
                    audio_ready = True
            except Exception:
                traceback.print_exc()  # Ignore exceptions and retry
        if create_video and video_file_url and not video_ready:
            try:
                response = requests.get(video_file_url)
                if response.status_code == 200:
                    video_ready = True
            except Exception:
                traceback.print_exc()
        if audio_ready and (not create_video or video_ready):
            status_text = "Media is ready."
            audio_player_html = (
                create_audio_player(audio_file_url) if audio_ready else ""
            )
            video_player_html = (
                create_video_player(video_file_url) if video_ready else ""
            )
            progress(1.0)
            return status_text, audio_player_html, video_player_html
        else:
            time.sleep(wait_time)
    return "Media not available yet.", "", ""


def create_audio_player(audio_url):
    if audio_url:
        return f"""
        <audio controls controlsList="nodownload">
            <source src="{audio_url}" type="audio/mpeg">
            Your browser does not support the audio element.
        </audio>
        """
    else:
        return ""


def create_video_player(video_url):
    if video_url:
        return f"""
        <video controls controlsList="nodownload" width="100%">
            <source src="{video_url}" type="video/mp4">
            Your browser does not support the video tag.
        </video>
        """
    else:
        return ""


with gr.Blocks() as demo:
    gr.Markdown("## Test the Lettercast.ai API (beta)")
    # gr.Markdown(
    #     "To get your access token visit [developer.lettercast.ai](https://developer.lettercast.ai)"
    # )
    with gr.Row():
        token = gr.Textbox(label="Token", placeholder="Enter your token")
        feed_id = gr.Textbox(label="Feed ID", placeholder="Enter your feed ID")
    with gr.Row():
        url = gr.Textbox(label="URL", placeholder="Enter a URL")
        file = gr.File(label="Upload a PDF file", type="filepath")
    with gr.Row():
        detail_level = gr.Dropdown(
            choices=["summary", "full"], label="Detail Level", value="summary"
        )
        hosts = gr.Dropdown(
            choices=["single_host", "two_hosts"], label="Hosts", value="single_host"
        )
        create_video = gr.Checkbox(label="Create Video (PDFs)", value=False)

    # Button for creating pod and showing the JSON response
    create_pod_btn = gr.Button("Create Pod")
    output = gr.JSON(label="Response")
    audio_pod_data = gr.State()
    video_pod_data = gr.State()

    # Button for starting the polling process
    start_polling_btn = gr.Button("Start Polling")
    status = gr.Textbox(label="Status")
    audio_player = gr.HTML(label="Generated Audio")
    video_player = gr.HTML(label="Generated Video")

    # Handle pod creation
    create_pod_btn.click(
        fn=create_cast,
        inputs=[
            token,
            feed_id,
            url,
            file,
            detail_level,
            create_video,
            hosts,
        ],
        outputs=[output, audio_pod_data, video_pod_data],
        show_progress="minimal",
        concurrency_limit=1,
    )

    # Handle polling for media readiness
    start_polling_btn.click(
        fn=lambda audio_pod_data, video_pod_data, create_video: poll_until_ready(
            audio_pod_data, video_pod_data, create_video
        ),
        inputs=[audio_pod_data, video_pod_data, create_video],
        outputs=[status, audio_player, video_player],
        show_api=False,
    )

demo.launch()