# app.py ##################################################### Import necessary libraries ##################################################### import os, shutil, time, json, sys, warnings import gradio as gr from pathlib import Path sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) warnings.filterwarnings("ignore") from KeyFrameSelection.FeatureExtraction import process_video, save_records from KeyFrameSelection.Similarties import hash_filter, clip_filter from FrameProcessor.utils.io_utils import get_frames_from_folder, save_description_to_csv from FrameProcessor.graph.workflow import frame_processor from config.paths import output_csv_file, output_json_file ##################################################### Define the main summarization function ##################################################### def summarize_video(video_path): keyframe_dir = "outputs/keyframes" csv_path = "outputs/keyframes.csv" if os.path.exists("outputs"): shutil.rmtree("outputs") os.makedirs("outputs/final_output", exist_ok=True) start = time.time() # Step 1: Extract raw keyframes records, fps = process_video(video_path, interval_sec=10) # Step 2: Filter min_frames = 10 max_iterations = 20 iteration = 0 hash_threshold = 5 ssim_threshold = 0.95 clip_threshold = 0.90 filtered = records while len(filtered) >= min_frames and iteration < max_iterations: filtered = hash_filter(filtered, hash_threshold, ssim_threshold, 5) filtered = clip_filter(filtered, clip_threshold, 5) hash_threshold = max(1, hash_threshold - 1) ssim_threshold = max(0.5, ssim_threshold - 0.05) clip_threshold = min(0.99, clip_threshold + 0.03) iteration += 1 save_records(filtered, keyframe_dir, csv_path, fps) frame_paths = get_frames_from_folder(keyframe_dir) # Step 3: Graph processing on each frame results = [] for frame_path in frame_paths: state = { "frame_path": frame_path, "frame_data": {}, "frame_features": {}, "importance": "not_important", "reason": "", "description": {}, "next_step": "describe_frame" } try: result = frame_processor.invoke(state) results.append(result) # time.sleep(4.1) # ✅ Add delay here # this is to avoid rate limits if result["importance"] == "important": save_description_to_csv(result) except Exception as e: results.append({ "frame_path": frame_path, "importance": "error", "reason": str(e) }) important = [r for r in results if r["importance"] == "important"] with open(output_json_file, "w") as f: json.dump(results, f, indent=2, ensure_ascii=False) end = time.time() return f"✅ Processed {len(important)} important frames out of {len(results)} in {end - start:.2f}s." ###################################################### Gradio UI ##################################################### def process_uploaded_video(video_file): if not video_file: raise gr.Error("Please upload a video first.") return summarize_video(video_file) with gr.Blocks(theme=gr.themes.Soft()) as demo: google_key = os.getenv("GOOGLE_API_KEY") gemini_key = os.getenv("GEMINI_API_KEY") google_short = google_key if google_key else "Not found" gemini_short = gemini_key if gemini_key else "Not found" gr.Markdown( """
Upload your lecture or tutorial video
Click Summarize to extract important frames and their content