app.py

import gradio as gr
import os
from pathlib import Path
from gradio_client import Client, handle_file
from dotenv import load_dotenv
import base64


load_dotenv()

def create_client():
    hf_token = os.getenv("HF_TOKEN")
    if not hf_token:
        raise ValueError("HF_TOKEN environment variable not set")
    return Client("wxcyn/video-analysis", hf_token=hf_token)

def process_video(video_file, question, password, progress=gr.Progress()):
    if video_file is None:
        raise gr.Error("No video file uploaded.")

    video_file_path = video_file  # type="filepath", so this is already a path string
    if not os.path.exists(video_file_path):
        raise gr.Error("Uploaded file could not be accessed.")

    progress(0.2, desc="Connecting to video analysis service...")
    client = create_client()

    progress(0.4, desc="Uploading and processing video...")
    wrapped_file = handle_file(video_file_path)

    # Call the private space
    result = client.predict(
        video_file=wrapped_file,
        question=question,
        password=password,
        api_name="/process_video"
    )

    progress(0.8, desc="Processing results...")
    print("DEBUG: Result from private space:", result)

    # Expecting a tuple: (analysis_text, video_clip_info)
    if isinstance(result, tuple) and len(result) == 2:
        analysis_text, video_clip_info = result

        # If video_clip_info is a string (path)
        if isinstance(video_clip_info, str):
            if os.path.exists(video_clip_info):
                return analysis_text, video_clip_info
            else:
                # Can't find the clip, just return the analysis
                return analysis_text, None

        elif isinstance(video_clip_info, dict):
            # If the dictionary contains a "video" key
            if "video" in video_clip_info:
                clip_path = video_clip_info["video"]
                if os.path.exists(clip_path):
                    return analysis_text, clip_path
                else:
                    # Clip failed, but we still return the analysis
                    return analysis_text, None
            else:
                # No recognized video key, just return the analysis
                return analysis_text, None
        else:
            # Unexpected format, but still return analysis
            return analysis_text, None
    else:
        # Unexpected result format, but we got some result, attempt to show analysis as a string
        if isinstance(result, str):
            return result, None
        # Otherwise, no valid analysis
        raise gr.Error("Unexpected response format from private service.")



def create_interface():
    with gr.Blocks(title="Public Video Content Analysis") as interface:
        gr.Markdown("# Video Content Analysis (Created by Edward&Cynthia)")
        gr.Markdown("# Watch our demo video! [Spacial Intelligence, Reasoning and Counting, Temporal continuity](https://www.youtube.com/watch?v=XuKRg7bRnME)")
        with gr.Row():
            with gr.Column(scale=2):
                password = gr.Textbox(
                    label="Password",
                    type="password",
                    info="Enter your access password.Do not share your password, request from shred-roseate.0b@icloud.com if needed",
                    interactive=True
                )
                video_file = gr.File(
                    label="Upload Video File (1 file under 2.5 minute,up to 30MB)",
                    file_types=['video'],
                    interactive=True,
                    type="filepath"
                )
                question = gr.Textbox(
                    label="Your Question",
                    placeholder="What is happening in this video?",
                    interactive=True
                )
                submit_btn = gr.Button("Analyze Video", variant="primary")

            with gr.Column(scale=3):
                output_text = gr.Textbox(
                    label="Analysis Result",
                    lines=15,
                    interactive=False
                )
                video_output = gr.Video(
                    label="Relevant Video Clip",
                    interactive=False
                )

        submit_btn.click(
            fn=process_video,
            inputs=[video_file, question, password],
            outputs=[output_text, video_output],
        )


    return interface

if __name__ == "__main__":
    Path('data').mkdir(exist_ok=True)
    Path('temp').mkdir(exist_ok=True)
    interface = create_interface()
    interface.queue()
    interface.launch(
        server_name="0.0.0.0",
        server_port=7860,
        show_error=True,
        max_threads=140
    )