from scenedetect import VideoManager, SceneManager from scenedetect.detectors import ContentDetector import cv2 import os import numpy as np import ffmpeg from moviepy.editor import VideoFileClip import time from utils import read_files from grok_analyze import analyze_image_with_grok_f import json import random import string import random def extract_scenes_scenedetect(video_path, threshold=30.0, min_scene_duration=5.0): vid_obj={"type":"video", "path":video_path} vid_obj["scenes"]=[ ] # Initialize video and scene manager video = VideoManager([video_path]) scene_manager = SceneManager() scene_manager.add_detector(ContentDetector(threshold=threshold)) # Detect scenes video.start() scene_manager.detect_scenes(video) scene_list = scene_manager.get_scene_list() # Get video FPS for time calculations cap = cv2.VideoCapture(video_path) fps = cap.get(cv2.CAP_PROP_FPS) min_frames = int(min_scene_duration * fps) # Convert 25 seconds to frames # Filter scenes to ensure at least 25 seconds (min_frames) between them filtered_scenes = [] last_frame = -min_frames # Allow the first scene to start at frame 0 for scene in scene_list: print(scene) start_frame = scene[0].get_frames() if start_frame >= last_frame + min_frames: filtered_scenes.append(scene) last_frame = start_frame print(f"Total scenes detected: {len(scene_list)}") print(f"Filtered scenes (at least {min_scene_duration} seconds apart): {len(filtered_scenes)}") if not filtered_scenes: print("No scenes detected; extracting the first frame of the video.") cap.set(cv2.CAP_PROP_POS_FRAMES, 0) ret, frame = cap.read() if ret: # Calculate timestamp for frame 0 timestamp = 0 / fps minutes = int(timestamp // 60) seconds = int(timestamp % 60) milliseconds = int((timestamp % 1) * 1000) print(f"First Frame - Timestamp: {minutes:02d}:{seconds:02d}.{milliseconds:03d}") # Analyze frame and store scene info description = analyze_image_with_grok_f(frame) scene_info = { "timestamp": f"{minutes:02d}:{seconds:02d}.{milliseconds:03d}", "description": description } vid_obj["scenes"].append(scene_info) else: print("Failed to read the first frame of the video.") cap.release() video.release() return vid_obj else: # Extract frames from filtered scenes for i, scene in enumerate(filtered_scenes): # Calculate and format timestamp start_frame = scene[0].get_frames() timestamp = start_frame / fps minutes = int(timestamp // 60) seconds = int(timestamp % 60) milliseconds = int((timestamp % 1) * 1000) print(f"Scene {i:04d} - Timestamp: {minutes:02d}:{seconds:02d}.{milliseconds:03d}") # Seek to the start of the scene cap.set(cv2.CAP_PROP_POS_FRAMES, start_frame) ret, frame = cap.read() description=analyze_image_with_grok_f(frame) scene={"timestamp":f"{minutes:02d}:{seconds:02d}.{milliseconds:03d}", "description":description} vid_obj["scenes"].append(scene) # if ret: # output_path = os.path.join(output_folder, f"scene_{i:04d}.jpg") # cv2.imwrite(output_path, frame) print(vid_obj) cap.release() video.release() return vid_obj def get_metadata(all_files): all_files_info=[] for file_path in all_files: print(file_path) if file_path.split(".")[-1] in ["MOV","mp4"]: vid_info=extract_scenes_scenedetect(file_path) all_files_info.append(vid_info) random_string = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8)) file_name = f"all_files_info_{random_string}.json" with open(file_name, "w") as file: json.dump(all_files_info, file, indent=4) # upload_to_s3(file_name,"bucket","s3key") return all_files_info,file_name def get_scenes_metadata(metadata, no_scenes=10): selected = [] while len(selected) < no_scenes and metadata: # Pick 3, stop if list empties item = random.choice(metadata) selected.append(item) metadata.remove(item) # print(selected) # print(len(selected)) return(selected) if __name__ == "__main__": with open('/Users/georgia.bucea/products/ShortsAI/all_files_metadata.json', 'r') as file: data = json.load(file) print(len(data)) get_scenes_metadata(data) # all_files_info=[] # # dir_path="/Users/georgia.bucea/products/ShortsAI/21_aug" # # all_files=read_files(dir_path) # dir_path="/Users/georgia.bucea/products/ShortsAI/16_aug" # # all_files2=read_files(dir_path) # all_files=read_files(dir_path) # # all_files=all_files+all_files2 # for file in all_files: # file_path=dir_path+"/"+file # if file_path.split(".")[-1] in ["MOV","mp4"]: # vid_info=extract_scenes_scenedetect(file_path) # all_files_info.append(vid_info) # with open("all_files_info_21_aug_16_aug.json", "w") as file: # json.dump(all_files_info, file, indent=4)