import cv2 import base64 import os import argparse import sys from openai import OpenAI def extract_frames(video_path, num_frames=5): """Extract a few evenly spaced frames from a video and encode to base64.""" cap = cv2.VideoCapture(video_path) total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) frames = [] if total_frames == 0: return frames step = max(1, total_frames // num_frames) for i in range(num_frames): cap.set(cv2.CAP_PROP_POS_FRAMES, min(i * step, total_frames - 1)) ret, frame = cap.read() if ret: # encode to jpeg _, buffer = cv2.imencode('.jpg', frame) b64_img = base64.b64encode(buffer).decode('utf-8') frames.append(f"data:image/jpeg;base64,{b64_img}") cap.release() return frames def analyze_video(video_source, prompt, api_key=None, model="MiniMaxAI/MiniMax-M3", base_url="https://api.featherless.ai/v1"): api_key = api_key or os.getenv("FEATHERLESS_API_KEY") if not api_key: raise ValueError("API key must be provided or set in FEATHERLESS_API_KEY env var") client = OpenAI(base_url=base_url, api_key=api_key) print(f"[+] Extracting frames from {video_source}...") frames = extract_frames(video_source, num_frames=5) if not frames: raise RuntimeError("Failed to extract any frames from the video.") content = [{"type": "text", "text": prompt}] for frame in frames: content.append({"type": "image_url", "image_url": {"url": frame}}) messages = [ { "role": "user", "content": content } ] print(f"[+] Sending {len(frames)} frames to vision model...") response = client.chat.completions.create( model=model, messages=messages ) return response.choices[0].message.content def main(): parser = argparse.ArgumentParser(description="Analyze video using Featherless AI vision models.") parser.add_argument("--video-url", required=True, help="Video file path or URL.") parser.add_argument("--prompt", required=True, help="Prompt for analysis.") parser.add_argument("--api-key", help="Featherless API key (overrides FEATHERLESS_API_KEY env var).") parser.add_argument("--model", default="MiniMaxAI/MiniMax-M3", help="Model name.") parser.add_argument("--base-url", default="https://api.featherless.ai/v1", help="Base API URL.") args = parser.parse_args() try: description = analyze_video( video_source=args.video_url, prompt=args.prompt, api_key=args.api_key, model=args.model, base_url=args.base_url, ) print("\n--- Video Analysis ---\n") print(description) print("\n----------------------\n") except Exception as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()