Add application file

app.py

@@ -0,0 +1,127 @@
+import gradio as gr
+import clip
+import torch
+from qdrant_client import QdrantClient
+import subprocess
+import os
+import uuid
+
+# Setup CLIP
+device = "cuda" if torch.cuda.is_available() else "cpu"
+model, preprocess = clip.load("ViT-B/32", device=device)
+
+# Setup Qdrant
+client = QdrantClient(
+    url="https://265484ec-5f64-40ec-a619-c7c9dffc2dd9.us-east-1-0.aws.cloud.qdrant.io:6333",
+    api_key="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhY2Nlc3MiOiJtIn0.I2MgcVnOKkWmOXwFlqJqEqm6LFQIF4cjxU5up4wxwyw"
+)
+COLLECTION_NAME = "video_segments"
+
+# Paths
+VIDEO_BASE_DIR = "/project/phan/tqn/RAG-VideoReferencing/"
+CLIP_OUTPUT_DIR = "generated_clips"
+os.makedirs(CLIP_OUTPUT_DIR, exist_ok=True)
+
+DEFAULT_VIDEO_FILENAME = "temp_video_0.mp4"
+DEFAULT_VIDEO_PATH = os.path.join(VIDEO_BASE_DIR, DEFAULT_VIDEO_FILENAME)
+
+
+def extract_video_clip(input_path, start_time, end_time):
+    """
+    Use ffmpeg to extract a clip from the video.
+    """
+    clip_name = f"clip_{uuid.uuid4().hex}.mp4"
+    output_path = os.path.join(CLIP_OUTPUT_DIR, clip_name)
+
+    command = [
+        "ffmpeg",
+        "-ss", str(start_time),
+        "-i", input_path,
+        "-to", str(end_time - start_time),
+        "-c", "copy",
+        output_path,
+        "-y"  # Overwrite if exists
+    ]
+
+    try:
+        subprocess.run(command, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        return output_path
+    except subprocess.CalledProcessError as e:
+        print(f"[ERROR] FFmpeg failed: {e}")
+        return None
+
+def time_to_seconds(time_str):
+    h, m, s = time_str.split(':')
+    return int(h) * 3600 + int(m) * 60 + float(s)
+
+def search_and_clip_video(text_query: str):
+    print(f"[INFO] Searching for: {text_query}")
+
+    # Encode query
+    with torch.no_grad():
+        text_tokens = clip.tokenize([text_query]).to(device)
+        text_features = model.encode_text(text_tokens)
+        text_features /= text_features.norm(dim=1, keepdim=True)
+
+    # Query Qdrant
+    search_result = client.search(
+        collection_name=COLLECTION_NAME,
+        query_vector=text_features.cpu().numpy()[0].tolist(),
+        limit=1,
+    )
+
+    if not search_result:
+        print("[WARN] No result found.")
+        return DEFAULT_VIDEO_PATH
+
+    hit = search_result[0]
+    start = hit.payload.get("start", 0)
+    end = hit.payload.get("end", 0)
+    start = time_to_seconds(start)
+    end = time_to_seconds(end)
+    video_filename = hit.payload.get("video_path", DEFAULT_VIDEO_FILENAME)
+
+    full_video_path = os.path.join(VIDEO_BASE_DIR, video_filename)
+
+    print(f"[INFO] Found: {video_filename} ({start} - {end})")
+
+    # Extract clip using ffmpeg
+    clip_path = extract_video_clip(full_video_path, float(start), float(end))
+    if clip_path and os.path.exists(clip_path):
+        print(f"[INFO] Returning clip: {clip_path}")
+        return clip_path
+    else:
+        print("[WARN] Failed to extract clip, returning default video.")
+        return DEFAULT_VIDEO_PATH
+
+
+# Fallback test interface
+def get_test_video():
+    print("[INFO] Returning test video path")
+    return DEFAULT_VIDEO_PATH
+
+
+# Gradio Interfaces
+search_demo = gr.Interface(
+    fn=search_and_clip_video,
+    inputs=gr.Textbox(label="Enter search query", value="sample query"),
+    outputs=gr.Video(label="Video Result"),
+    title="🎥 Semantic Video Search with Clip Extraction",
+    description="Returns a clipped video segment matching your query."
+)
+
+test_demo = gr.Interface(
+    fn=get_test_video,
+    inputs=None,
+    outputs=gr.Video(label="Test Video"),
+    title="Simple Video Test",
+    description="Always displays the default video to verify video player works."
+)
+
+demo = gr.TabbedInterface(
+    [search_demo, test_demo],
+    ["Search Video", "Test Video Player"]
+)
+
+if __name__ == "__main__":
+    demo.launch(share=True)