renderer_app.py

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import gradio as gr
import uvicorn

from api import api, job_manager, project_manager, settings
from renderer import RenderEngine
from renderer.core.models import AIReelsRequest
from renderer.scenes import Timeline
from renderer.studio import capability_catalog
from renderer.templates import apply_creative_style, apply_preset, list_creative_styles, list_platform_profiles, list_templates


def _mcp_description(summary: str, example: dict[str, Any]) -> str:
    """Build a concise MCP description with a machine-readable JSON example."""
    payload = json.dumps(example, indent=2, ensure_ascii=True)
    return f"{summary}\n\nExample JSON:\n```json\n{payload}\n```"


def create_dashboard() -> gr.Blocks:
    with gr.Blocks(title="Ava2lon Studio AI") as dashboard:
        gr.Markdown(
            "# Ava2lon Studio AI\n"
            "CPU-first CapCut-class video automation studio with REST API parity, async jobs, webhooks, and project JSON."
        )
        with gr.Tab("Projects"):
            project_name = gr.Textbox(label="Project name", value="Untitled Ava2lon Project")
            project_metadata = gr.Textbox(label="Metadata JSON", lines=5, value=json.dumps({"platform": "tiktok"}, indent=2))
            project_create = gr.Button("Create Project", variant="primary")
            project_list = gr.Button("Refresh Projects")
            project_output = gr.JSON(label="Projects")
            project_create.click(fn=create_project, inputs=[project_name, project_metadata], outputs=project_output, api_name="create_project", api_description=_mcp_description("Create a project from a name and JSON metadata object.", {"name": "Campaign 01", "metadata_json": '{"platform":"tiktok"}'}))
            project_list.click(fn=list_projects, outputs=project_output, api_name="list_projects", api_description=_mcp_description("List saved studio projects.", {}))

        with gr.Tab("Assets"):
            gr.Markdown("Use `/upload` or `/assets/upload` for multipart assets, then attach them to a project with `/project/assets/add`.")
            asset_project_id = gr.Textbox(label="Project ID")
            asset_json = gr.Textbox(label="Asset JSON", lines=6, value=json.dumps({"path": "clip.mp4", "kind": "video"}, indent=2))
            asset_button = gr.Button("Attach Asset", variant="primary")
            asset_output = gr.JSON(label="Project")
            asset_button.click(fn=add_project_asset, inputs=[asset_project_id, asset_json], outputs=asset_output, api_name="add_project_asset", api_description=_mcp_description("Attach a media asset JSON object to an existing project.", {"project_id": "project_123", "asset_json": '{"path":"clip.mp4","kind":"video"}'}))

        with gr.Tab("Timeline"):
            timeline_project_id = gr.Textbox(label="Project ID")
            timeline_track_type = gr.Dropdown(choices=["video", "audio", "text", "overlay", "sticker", "subtitle"], value="video", label="Track type")
            timeline_item = gr.Textbox(
                label="Timeline item JSON",
                lines=8,
                value=json.dumps({"media": "clip.mp4", "start": 0, "duration": 5, "caption": "Hook"}, indent=2),
            )
            timeline_add_button = gr.Button("Add To Timeline", variant="primary")
            timeline_output = gr.JSON(label="Project")
            timeline_add_button.click(fn=add_timeline_item, inputs=[timeline_project_id, timeline_track_type, timeline_item], outputs=timeline_output, api_name="add_timeline_item", api_description=_mcp_description("Add a typed item to a project timeline.", {"project_id": "project_123", "track_type": "video", "item_json": '{"media":"clip.mp4","start":0,"duration":5,"caption":"Hook"}'}))

            timeline_operation_json = gr.Textbox(
                label="Operation JSON",
                lines=8,
                value=json.dumps({"operation": "split", "item_id": "clip_123", "params": {"offset": 2.5}}, indent=2),
            )
            timeline_operation_button = gr.Button("Apply Operation")
            timeline_operation_button.click(fn=apply_timeline_operation, inputs=[timeline_project_id, timeline_operation_json], outputs=timeline_output, api_name="apply_timeline_operation", api_description=_mcp_description("Apply a timeline operation such as split, trim, insert, or ripple delete.", {"project_id": "project_123", "operation_json": '{"operation":"split","item_id":"clip_123","params":{"offset":2.5}}'}))

        with gr.Tab("Templates"):
            template_button = gr.Button("Load Template Catalog")
            template_output = gr.JSON(label="Templates")
            template_button.click(fn=load_template_catalog, outputs=template_output, api_name="load_template_catalog", api_description=_mcp_description("Load available render and caption templates.", {}))

        with gr.Tab("Effects"):
            effect_button = gr.Button("Load Effect Catalog")
            effect_output = gr.JSON(label="Effects")
            effect_button.click(fn=load_effect_catalog, outputs=effect_output, api_name="load_effect_catalog", api_description=_mcp_description("Load available video effects.", {}))

        with gr.Tab("Filters"):
            filter_button = gr.Button("Load Filter Catalog")
            filter_output = gr.JSON(label="Filters")
            filter_button.click(fn=load_filter_catalog, outputs=filter_output, api_name="load_filter_catalog", api_description=_mcp_description("Load available video filters.", {}))

        with gr.Tab("Captions"):
            caption_text = gr.Textbox(label="Caption source text", lines=6)
            caption_button = gr.Button("Submit Caption Job", variant="primary")
            caption_output = gr.JSON(label="Caption Job")
            caption_button.click(fn=submit_caption_job, inputs=caption_text, outputs=caption_output, api_name="submit_caption_job", api_description=_mcp_description("Submit a caption-generation job from source text.", {"text": "Launch faster with automated rendering."}))

        with gr.Tab("Audio"):
            audio_button = gr.Button("Load Audio Catalog")
            audio_output = gr.JSON(label="Audio")
            audio_button.click(fn=load_audio_catalog, outputs=audio_output, api_name="load_audio_catalog", api_description=_mcp_description("Load audio-processing and music-generation capabilities.", {}))

        with gr.Tab("AI Tools"):
            ai_tool = gr.Dropdown(choices=capability_catalog()["ai_editing"], value="auto_viral_score", label="AI tool")
            ai_payload = gr.Textbox(label="AI payload JSON", lines=8, value=json.dumps({"platform": "tiktok", "text": "A strong opening hook"}, indent=2))
            ai_button = gr.Button("Submit AI Tool", variant="primary")
            ai_output = gr.JSON(label="AI Job")
            ai_button.click(fn=submit_ai_tool, inputs=[ai_tool, ai_payload], outputs=ai_output, api_name="submit_ai_tool", api_description=_mcp_description("Submit a configured AI editing tool job.", {"tool": "auto_viral_score", "payload": '{"platform":"tiktok","text":"A strong opening hook"}'}))

        with gr.Tab("Rendering"):
            render_json = gr.Textbox(
                label="Render JSON",
                lines=14,
                value="",
                placeholder="Paste a production render request JSON object with absolute or uploaded asset paths.",
            )
            render_button = gr.Button("Submit Render", variant="primary")
            render_output = gr.JSON(label="Submission")
            render_button.click(fn=submit_render_job, inputs=render_json, outputs=render_output, api_name="submit_render_job", api_description=_mcp_description("Submit a validated JSON video render job.", {"payload": '{"scenes":[{"start":0,"duration":5,"media":"clip.mp4","caption":"Opening hook"}],"platform":"tiktok","output_name":"render.mp4"}'}))

        with gr.Tab("AI Reels"):
            script = gr.Textbox(label="Script", lines=6)
            voiceover = gr.File(label="Voiceover", file_types=["audio"], type="filepath")
            assets = gr.File(label="Assets", file_count="multiple", type="filepath")
            template = gr.Dropdown(choices=list_templates(), value="tiktok_classic", label="Caption Template")
            creative_style = gr.Dropdown(choices=list_creative_styles(), value="viral_shorts", label="Creative Style")
            platform = gr.Dropdown(choices=list_platform_profiles(), value="tiktok", label="Platform")
            music = gr.File(label="Background Music", file_types=["audio"], type="filepath")
            ai_button = gr.Button("Submit AI Reel", variant="primary")
            ai_output = gr.JSON(label="Submission")
            ai_button.click(fn=submit_ai_reel, inputs=[script, voiceover, assets, template, creative_style, platform, music], outputs=ai_output, api_name="submit_ai_reel", api_description=_mcp_description("Create an AI reel from a script, voiceover, and visual assets. Upload files first and use the returned paths.", {"script": "Three automation tips.", "voiceover": "/tmp/voice.wav", "assets": ["/tmp/a.jpg", "/tmp/b.mp4"], "template": "tiktok_classic", "creative_style": "viral_shorts", "platform": "tiktok", "music": None}))

        with gr.Tab("Batch Render"):
            batch_json = gr.Textbox(label="Batch JSON", lines=14, value=json.dumps({"jobs": []}, indent=2))
            batch_button = gr.Button("Submit Batch", variant="primary")
            batch_output = gr.JSON(label="Batch Submission")
            batch_button.click(fn=submit_batch_render, inputs=batch_json, outputs=batch_output, api_name="submit_batch_render", api_description=_mcp_description("Submit multiple render jobs from a JSON batch.", {"payload": '{"jobs":[{"scenes":[{"start":0,"duration":3,"media":"clip.mp4"}],"output_name":"clip-a.mp4"}]}'}))

        with gr.Tab("Job Status"):
            status_job_id = gr.Textbox(label="Job ID")
            status_button = gr.Button("Refresh")
            status_output = gr.JSON(label="Status")
            status_button.click(fn=get_job_status, inputs=status_job_id, outputs=status_output, api_name="get_job_status", api_description=_mcp_description("Retrieve the current state and metrics for a job.", {"job_id": "job_abc123"}))

        with gr.Tab("Logs"):
            logs_job_id = gr.Textbox(label="Job ID")
            logs_button = gr.Button("Load Logs")
            logs_output = gr.Textbox(label="Logs", lines=20)
            logs_button.click(fn=get_job_logs, inputs=logs_job_id, outputs=logs_output, api_name="get_job_logs", api_description=_mcp_description("Retrieve execution logs for a job.", {"job_id": "job_abc123"}))

        with gr.Tab("Downloads"):
            download_job_id = gr.Textbox(label="Job ID")
            download_button = gr.Button("Get Output")
            download_output = gr.File(label="Rendered Video")
            download_button.click(fn=get_download_path, inputs=download_job_id, outputs=download_output, api_name="get_download_path", api_description=_mcp_description("Return the completed artifact path for a job.", {"job_id": "job_abc123"}))

        with gr.Tab("Transcribe"):
            transcribe_audio = gr.File(label="Audio or Video", file_types=["audio", "video"], type="filepath")
            transcribe_model = gr.Dropdown(
                choices=["tiny", "base", "small", "medium", "large-v3"],
                value=settings.whisper_model_size,
                label="Whisper Model",
            )
            transcribe_language = gr.Textbox(label="Language", placeholder="Optional ISO code, e.g. en")
            transcribe_button = gr.Button("Transcribe", variant="primary")
            transcribe_output = gr.JSON(label="Transcript")
            transcribe_button.click(
                fn=transcribe_audio_file,
                inputs=[transcribe_audio, transcribe_model, transcribe_language],
                outputs=transcribe_output,
                api_name="transcribe_audio_file",
                api_description=_mcp_description("Transcribe an uploaded audio or video file with word timestamps. Upload the file first and use its returned path.", {"path": "/tmp/interview.mp3", "model_size": "tiny", "language": "en"}),
            )

        with gr.Tab("Asset Inspector"):
            asset_path = gr.Textbox(label="Asset path")
            inspect_button = gr.Button("Inspect")
            inspect_output = gr.JSON(label="Metadata")
            inspect_button.click(fn=inspect_media_asset, inputs=asset_path, outputs=inspect_output, api_name="inspect_media_asset", api_description=_mcp_description("Inspect media codecs, streams, duration, and dimensions.", {"path": "/app/storage/video.mp4"}))

        with gr.Tab("AI Analysis"):
            analysis_media = gr.Textbox(label="Media URL or path")
            analysis_transcript = gr.Textbox(label="Transcript", lines=5)
            analysis_platform = gr.Dropdown(choices=list_platform_profiles(), value="tiktok", label="Target Platform")
            analysis_button = gr.Button("Submit Analysis", variant="primary")
            analysis_output = gr.JSON(label="Analysis Job")
            analysis_button.click(
                fn=submit_media_analysis,
                inputs=[analysis_media, analysis_transcript, analysis_platform],
                outputs=analysis_output,
                api_name="submit_media_analysis",
                api_description=_mcp_description("Analyze media for highlights, pacing, platform fit, and engagement signals.", {"media": "https://example.com/video.mp4", "transcript": "A strong opening hook and useful explanation.", "platform": "tiktok"}),
            )

        with gr.Tab("Clip Generator"):
            clip_media = gr.Textbox(label="Media URL or path")
            clip_json = gr.Textbox(label="Clip JSON", lines=6, value=json.dumps([{"start": 0, "end": 8}], indent=2))
            clip_button = gr.Button("Generate Clips", variant="primary")
            clip_output = gr.JSON(label="Clip Job")
            clip_button.click(fn=submit_clip_job, inputs=[clip_media, clip_json], outputs=clip_output, api_name="submit_clip_job", api_description=_mcp_description("Generate one or more clips from media and timed clip JSON.", {"media": "https://example.com/video.mp4", "clips_json": '[{"start":0,"end":8},{"start":20,"end":32}]'}))

        with gr.Tab("Metadata"):
            metadata_topic = gr.Textbox(label="Topic or transcript", lines=5)
            metadata_platform = gr.Dropdown(choices=list_platform_profiles(), value="tiktok", label="Platform")
            metadata_button = gr.Button("Generate Metadata", variant="primary")
            metadata_output = gr.JSON(label="Metadata Job")
            metadata_button.click(fn=submit_metadata_job, inputs=[metadata_topic, metadata_platform], outputs=metadata_output, api_name="submit_metadata_job", api_description=_mcp_description("Generate platform-aware title, description, hashtags, and chapters.", {"topic": "Five n8n video automation mistakes", "platform": "youtube_shorts"}))

        with gr.Tab("Publishing"):
            publish_media = gr.Textbox(label="Media URL or rendered output path")
            publish_title = gr.Textbox(label="Title")
            publish_platforms = gr.Textbox(label="Platforms", value="youtube,tiktok,instagram")
            publish_button = gr.Button("Create Publish Draft", variant="primary")
            publish_output = gr.JSON(label="Publish Job")
            publish_button.click(fn=create_publish_draft, inputs=[publish_media, publish_title, publish_platforms], outputs=publish_output, api_name="create_publish_draft", api_description=_mcp_description("Create a publishing draft for one or more platforms.", {"media": "/app/exports/render.mp4", "title": "Automation Tips", "platforms": "youtube,tiktok,instagram"}))

        with gr.Tab("Settings"):
            settings_button = gr.Button("Load Settings")
            settings_output = gr.JSON(label="Settings")
            settings_button.click(fn=get_runtime_settings, outputs=settings_output, api_name="get_runtime_settings", api_description=_mcp_description("Return renderer configuration and runtime capability settings.", {}))

        with gr.Tab("Queue Monitor"):
            queue_button = gr.Button("Refresh Queue")
            queue_output = gr.JSON(label="Queue")
            queue_button.click(fn=get_queue_status, outputs=queue_output, api_name="get_queue_status", api_description=_mcp_description("Return queue counts and active render jobs.", {}))

    return dashboard


def submit_render_job(payload: str) -> dict[str, Any]:
    """Submit a validated JSON timeline for asynchronous video rendering."""
    data = apply_creative_style(apply_preset(json.loads(payload)))
    request = Timeline.request_from_payload(data)
    job_id = job_manager.submit_render(request)
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def create_project(name: str, metadata_json: str) -> dict[str, Any]:
    """Create a persistent studio project with optional JSON metadata."""
    metadata = json.loads(metadata_json or "{}")
    return {"project": project_manager.create(name, metadata=metadata)}


def list_projects() -> dict[str, Any]:
    """List all persistent studio projects."""
    return {"projects": project_manager.list()}


def add_project_asset(project_id: str, asset_json: str) -> dict[str, Any]:
    """Attach a media asset JSON object to an existing project."""
    return {"project": project_manager.add_asset(project_id, json.loads(asset_json or "{}"))}


def add_timeline_item(project_id: str, track_type: str, item_json: str) -> dict[str, Any]:
    """Add a JSON item to a video, audio, text, overlay, sticker, or subtitle track."""
    return {"project": project_manager.add_to_timeline(project_id, json.loads(item_json or "{}"), track_type=track_type)}


def apply_timeline_operation(project_id: str, operation_json: str) -> dict[str, Any]:
    """Apply a JSON timeline operation such as split, trim, insert, or ripple delete."""
    data = json.loads(operation_json or "{}")
    return {
        "project": project_manager.timeline_operation(
            project_id,
            data.get("operation", "drag"),
            item_id=data.get("item_id"),
            params=data.get("params", {}),
        )
    }


def _catalog_section(section: str) -> dict[str, Any]:
    catalog = capability_catalog()
    return {section: catalog.get(section)}


def load_template_catalog() -> dict[str, Any]:
    """Return available render and caption template capabilities."""
    return _catalog_section("templates")


def load_effect_catalog() -> dict[str, Any]:
    """Return available video effect capabilities."""
    return _catalog_section("effects")


def load_filter_catalog() -> dict[str, Any]:
    """Return available video filter capabilities."""
    return _catalog_section("filters")


def load_audio_catalog() -> dict[str, Any]:
    """Return audio-processing and music-generation capabilities."""
    catalog = capability_catalog()
    return {"audio": catalog["audio"], "music": catalog["music_generator"]}


def submit_caption_job(text: str) -> dict[str, Any]:
    """Submit an asynchronous caption-generation job from source text."""
    from renderer.studio import StudioTaskProcessor

    job_id = job_manager.submit_task(lambda task_id, log: StudioTaskProcessor(settings, log=log).caption_generate({"text": text}, task_id))
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def submit_ai_tool(tool: str, payload: str) -> dict[str, Any]:
    """Submit an AI editing tool using its name and a JSON payload."""
    from renderer.studio import StudioTaskProcessor

    data = json.loads(payload or "{}")
    job_id = job_manager.submit_task(lambda task_id, log: StudioTaskProcessor(settings, log=log).ai_tool(tool, data, task_id))
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def submit_batch_render(payload: str) -> dict[str, Any]:
    """Submit multiple asynchronous render jobs from a JSON jobs array."""
    data = json.loads(payload)
    requests = [Timeline.request_from_payload(apply_creative_style(apply_preset(job))) for job in data.get("jobs", [])]
    return {"job_ids": job_manager.submit_batch(requests)}


def submit_ai_reel(
    script: str,
    voiceover: str,
    assets: list[str],
    template: str,
    creative_style: str,
    platform: str,
    music: str | None,
) -> dict[str, Any]:
    """Create an AI reel from a script, voiceover, visual assets, style, and music."""
    request = AIReelsRequest(
        script=script,
        voiceover=voiceover,
        assets=assets or [],
        template=template,
        creative_style=creative_style,
        platform=platform,
        background_music=music,
    )
    job_id = job_manager.submit_ai_reels(request)
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def get_job_status(job_id: str) -> dict[str, Any]:
    """Return state, metrics, artifact metadata, and errors for a job ID."""
    return job_manager.get(job_id).__dict__


def get_job_logs(job_id: str) -> str:
    """Return execution log messages for a job ID."""
    return "\n\n".join(job_manager.get(job_id).logs)


def get_download_path(job_id: str) -> str | None:
    """Return the local artifact path when a job has completed."""
    record = job_manager.get(job_id)
    if record.state != "COMPLETED":
        return None
    return record.output_path


def inspect_media_asset(path: str) -> dict[str, Any]:
    """Inspect a local media asset for streams, codecs, duration, and dimensions."""
    return RenderEngine(settings).inspect_asset(path)


def transcribe_audio_file(path: str, model_size: str, language: str) -> dict[str, Any]:
    """Transcribe an uploaded audio or video file with optional language selection."""
    return RenderEngine(settings).transcribe(
        path,
        model_size=model_size,
        language=language.strip() or None,
        word_timestamps=True,
    )


def submit_media_analysis(media: str, transcript: str, platform: str) -> dict[str, Any]:
    """Analyze media for highlights, pacing, retention, engagement, and platform fit."""
    from renderer.platform import PlatformProcessor

    job_id = job_manager.submit_task(
        lambda task_id, log: PlatformProcessor(settings, log=log).analyze(media, task_id, transcript=transcript, platform=platform)
    )
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def submit_clip_job(media: str, clips_json: str) -> dict[str, Any]:
    """Generate timed clips from a media path or URL and a JSON clip specification."""
    from renderer.platform import PlatformProcessor

    clips = json.loads(clips_json)
    job_id = job_manager.submit_task(lambda task_id, log: PlatformProcessor(settings, log=log).clips(media, task_id, clips))
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def submit_metadata_job(topic: str, platform: str) -> dict[str, Any]:
    """Generate platform-aware title, description, hashtags, keywords, and chapters."""
    from renderer.platform import PlatformProcessor

    job_id = job_manager.submit_task(lambda task_id, log: PlatformProcessor(settings, log=log).metadata(task_id, topic=topic, platform=platform))
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def create_publish_draft(media: str, title: str, platforms: str) -> dict[str, Any]:
    """Create an asynchronous publishing draft for comma-separated platforms."""
    from renderer.platform import PlatformProcessor

    payload = {"media": media, "title": title, "platforms": [item.strip() for item in platforms.split(",") if item.strip()], "draft": True}
    job_id = job_manager.submit_task(lambda task_id, log: PlatformProcessor(settings, log=log).publish(payload, task_id))
    return {"job_id": job_id, "status": f"/status/{job_id}", "download": f"/download/{job_id}"}


def get_queue_status() -> dict[str, Any]:
    """Return queue totals, state counts, worker limits, and active jobs."""
    return job_manager.summary()


def get_runtime_settings() -> dict[str, Any]:
    """Return renderer paths, worker settings, Whisper settings, and capabilities."""
    return {
        "product": "Ava2lon Studio AI",
        "base_dir": str(settings.base_dir),
        "temp_dir": str(settings.temp_dir),
        "exports_dir": str(settings.exports_dir),
        "storage_dir": str(settings.storage_dir),
        "max_workers": settings.max_workers,
        "whisper_model_size": settings.whisper_model_size,
        "whisper_device": settings.whisper_device,
        "capabilities": capability_catalog()["principles"],
    }


app = gr.mount_gradio_app(
    api,
    create_dashboard(),
    path="/dashboard",
    root_path="/dashboard",
    mcp_server=True,
)


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)