openfeedback.py

#!/usr/bin/env python3
"""OpenFeedback: AI-powered presentation and talk feedback from video."""

import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
from typing import Optional

from google import genai
from google.genai import types

# --- Constants ---

DEFAULT_MODEL = "gemini-2.5-flash-preview-05-20"
MAX_POLL_SECONDS = 300
POLL_INITIAL_INTERVAL = 1.0
POLL_BACKOFF_FACTOR = 2.0

ANALYSIS_PROMPT = """\
You are an expert presentation coach and public speaking analyst.

Analyze this video of a presentation or talk. Provide structured, actionable feedback.

## Required output format

### 1. Content Summary
Summarize the talk in 3-5 bullet points. Include timestamps for key sections.

### 2. Delivery Analysis
Evaluate these dimensions with specific observations and timestamps:
- **Energy & enthusiasm**: How engaged does the speaker sound and look?
- **Confidence**: Posture, eye contact, voice steadiness
- **Pacing**: Too fast, too slow, good variation?
- **Body language**: Gestures, movement, stance
- **Filler words**: Note any repeated fillers ("um", "uh", "like", "so", "you know") with approximate frequency

### 3. Strongest Moments
List 2-3 moments where the speaker was most effective. Include timestamps and explain why they worked.

### 4. Weakest Moments
List 2-3 moments that need the most improvement. Include timestamps and explain what went wrong.

### 5. Top 10 Actionable Fixes
Numbered list of specific, concrete improvements. Prioritize by impact. Each fix should reference a specific moment or pattern from the video.

### 6. Ratings
Rate each dimension on a 1-10 scale (10 = world-class TED talk level):

| Dimension | Score | One-line justification |
|-----------|-------|----------------------|
| Content | /10 | |
| Delivery | /10 | |
| Visuals/Slides | /10 | |
| Demo Execution | /10 | |
| Audience Engagement | /10 | |
| **Overall** | **/10** | |

If the video has no slides or no demo, mark those as N/A instead of guessing.

Be honest and direct. The speaker wants to improve, not hear flattery.
"""


def log(msg: str) -> None:
    """Print a progress message to stderr."""
    print(f"  {msg}", file=sys.stderr, flush=True)


def check_dependencies() -> None:
    """Verify required tools are available."""
    if not os.environ.get("GEMINI_API_KEY"):
        print("Error: GEMINI_API_KEY environment variable is not set.", file=sys.stderr)
        print("Get one at https://aistudio.google.com/apikey", file=sys.stderr)
        sys.exit(1)

    if not shutil.which("yt-dlp"):
        print("Error: yt-dlp is not installed. Install with: pip install yt-dlp", file=sys.stderr)
        sys.exit(1)


def is_url(source: str) -> bool:
    """Check if the source looks like a URL."""
    return source.startswith("http://") or source.startswith("https://")


def download_video(
    url: str,
    tmpdir: str,
    start: Optional[str] = None,
    end: Optional[str] = None,
) -> str:
    """Download a video from YouTube using yt-dlp. Returns the path to the downloaded file."""
    log("Downloading video...")

    output_template = os.path.join(tmpdir, "%(title).50s.%(ext)s")
    cmd = [
        "yt-dlp",
        "-f", "bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best",
        "--merge-output-format", "mp4",
        "-o", output_template,
        "--no-playlist",
        "--quiet",
        "--progress",
    ]

    if start and end:
        cmd += ["--download-sections", f"*{start}-{end}"]
    elif start:
        cmd += ["--download-sections", f"*{start}-inf"]
    elif end:
        cmd += ["--download-sections", f"*0:00-{end}"]

    cmd.append(url)

    try:
        subprocess.run(cmd, check=True, capture_output=True, text=True)
    except subprocess.CalledProcessError as e:
        print(f"Error downloading video: {e.stderr.strip()}", file=sys.stderr)
        sys.exit(1)

    # Find the downloaded file
    files = list(Path(tmpdir).glob("*.mp4"))
    if not files:
        # Try any video file
        files = list(Path(tmpdir).glob("*.*"))
        files = [f for f in files if f.suffix in (".mp4", ".mkv", ".webm", ".mov")]

    if not files:
        print("Error: No video file found after download.", file=sys.stderr)
        sys.exit(1)

    result = str(files[0])
    size_mb = os.path.getsize(result) / (1024 * 1024)
    log(f"Downloaded: {Path(result).name} ({size_mb:.1f} MB)")
    return result


def upload_to_gemini(client: genai.Client, video_path: str) -> types.File:
    """Upload a video file to the Gemini Files API and wait for it to become ACTIVE."""
    log("Uploading to Gemini...")

    file_size_mb = os.path.getsize(video_path) / (1024 * 1024)
    log(f"File size: {file_size_mb:.1f} MB")

    uploaded_file = client.files.upload(
        file=video_path,
        config=types.UploadFileConfig(
            display_name=Path(video_path).stem,
        ),
    )

    log("Waiting for Gemini to process the video...")
    elapsed = 0.0
    interval = POLL_INITIAL_INTERVAL

    while elapsed < MAX_POLL_SECONDS:
        file_info = client.files.get(name=uploaded_file.name)

        if file_info.state == "ACTIVE":
            log("Video processed successfully.")
            return file_info

        if file_info.state == "FAILED":
            error_msg = file_info.error if file_info.error else "Unknown error"
            print(f"Error: Gemini failed to process the video: {error_msg}", file=sys.stderr)
            sys.exit(1)

        time.sleep(interval)
        elapsed += interval
        interval = min(interval * POLL_BACKOFF_FACTOR, 30.0)

    print(f"Error: Timed out waiting for Gemini to process the video ({MAX_POLL_SECONDS}s).", file=sys.stderr)
    sys.exit(1)


def analyze_video(client: genai.Client, file_info: types.File, model: str) -> str:
    """Send the uploaded video to Gemini for analysis. Returns the response text."""
    log(f"Analyzing with {model}...")

    response = client.models.generate_content(
        model=model,
        contents=[
            types.Content(
                parts=[
                    types.Part.from_uri(file_uri=file_info.uri, mime_type=file_info.mime_type),
                    types.Part.from_text(text=ANALYSIS_PROMPT),
                ],
            ),
        ],
        config=types.GenerateContentConfig(
            temperature=0.3,
            max_output_tokens=8192,
        ),
    )

    if not response.text:
        print("Error: Gemini returned an empty response.", file=sys.stderr)
        sys.exit(1)

    return response.text


def cleanup_gemini_file(client: genai.Client, file_info: types.File) -> None:
    """Delete the uploaded file from Gemini to avoid storage costs."""
    try:
        client.files.delete(name=file_info.name)
        log("Cleaned up Gemini file.")
    except Exception:
        # Non-fatal: file will auto-expire after 48 hours anyway
        log("Warning: Could not delete Gemini file (it will auto-expire in 48h).")


def main() -> None:
    parser = argparse.ArgumentParser(
        prog="openfeedback",
        description="Get AI-powered feedback on your presentation or talk from a video.",
        epilog="Example: openfeedback 'https://youtube.com/watch?v=...' --start 38:00 --end 1:09:00",
    )
    parser.add_argument(
        "source",
        help="YouTube URL or path to a local video file",
    )
    parser.add_argument(
        "--start",
        help="Start timestamp for analysis (e.g., 38:00 or 0:38:00). Only applies to YouTube URLs.",
    )
    parser.add_argument(
        "--end",
        help="End timestamp for analysis (e.g., 1:09:00). Only applies to YouTube URLs.",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        dest="json_output",
        help="Output raw response as JSON (with metadata).",
    )
    parser.add_argument(
        "--model",
        default=DEFAULT_MODEL,
        help=f"Gemini model to use (default: {DEFAULT_MODEL}).",
    )

    args = parser.parse_args()

    check_dependencies()

    client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
    file_info = None
    tmpdir = None

    try:
        if is_url(args.source):
            tmpdir = tempfile.mkdtemp(prefix="openfeedback_")
            video_path = download_video(args.source, tmpdir, args.start, args.end)
        else:
            video_path = args.source
            if not os.path.isfile(video_path):
                print(f"Error: File not found: {video_path}", file=sys.stderr)
                sys.exit(1)
            if args.start or args.end:
                log("Warning: --start and --end are ignored for local files. Use ffmpeg to trim first.")

        file_info = upload_to_gemini(client, video_path)
        feedback = analyze_video(client, file_info, args.model)

        if args.json_output:
            output = {
                "source": args.source,
                "model": args.model,
                "feedback": feedback,
            }
            print(json.dumps(output, indent=2))
        else:
            print(feedback)

        log("Done!")

    finally:
        if file_info:
            cleanup_gemini_file(client, file_info)
        if tmpdir:
            shutil.rmtree(tmpdir, ignore_errors=True)


if __name__ == "__main__":
    main()