257 lines
7.8 KiB
Python
257 lines
7.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Analyze screen recordings using OpenRouter + Gemini Vision.
|
|
|
|
Sends the full video file directly to Gemini via OpenRouter for a
|
|
UX research-style analysis. Saves results as markdown in docs/research/.
|
|
|
|
Usage:
|
|
uv run python -m app.analyze_videos # analyze all .mp4 in videos/
|
|
uv run python -m app.analyze_videos videos/file.mp4 # single video
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from dotenv import load_dotenv
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
load_dotenv()
|
|
|
|
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
|
|
if not OPENROUTER_API_KEY:
|
|
print(
|
|
"ERROR: OpenRouter API key not found.\n"
|
|
" Put OPENROUTER_API_KEY=sk-... in .env (repo root) or set the env var.\n"
|
|
" Get one at https://openrouter.ai/keys",
|
|
file=sys.stderr,
|
|
)
|
|
sys.exit(1)
|
|
|
|
OPENROUTER_BASE = "https://openrouter.ai/api/v1"
|
|
DEFAULT_MODEL = os.getenv("OPENROUTER_MODEL", "~google/gemini-flash-latest")
|
|
|
|
UX_PROMPT = """\
|
|
Analyze this screen recording like a UX researcher.
|
|
|
|
Track:
|
|
- user goals
|
|
- hesitation
|
|
- repeated actions
|
|
- likely confusion
|
|
- unnecessary clicks
|
|
- context switching
|
|
- inefficient workflow patterns
|
|
- UI discoverability issues
|
|
- moments where expectations appear violated
|
|
|
|
Output:
|
|
1. overall workflow summary
|
|
2. friction timeline
|
|
3. inferred user intent
|
|
4. UX issues ranked by severity
|
|
5. suggested improvements
|
|
|
|
Be specific about UI elements, button labels, menu paths, and exact behaviors
|
|
you observe throughout the video.
|
|
"""
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def probe_duration(video_path: Path) -> float:
|
|
"""Get video duration in seconds."""
|
|
try:
|
|
dur = subprocess.check_output(
|
|
["ffprobe", "-v", "error", "-show_entries", "format=duration",
|
|
"-of", "default=noprint_wrappers=1:nokey=1", str(video_path)],
|
|
stderr=subprocess.DEVNULL,
|
|
).decode().strip()
|
|
return float(dur)
|
|
except (subprocess.CalledProcessError, FileNotFoundError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def read_video(video_path: Path) -> tuple[bytes, str]:
|
|
"""Read a video file and return (bytes, MIME type)."""
|
|
ext = video_path.suffix.lstrip(".").lower()
|
|
mime_map = {
|
|
"mp4": "video/mp4",
|
|
"mov": "video/quicktime",
|
|
"webm": "video/webm",
|
|
"mkv": "video/x-matroska",
|
|
}
|
|
mime = mime_map.get(ext, f"video/{ext}")
|
|
with open(video_path, "rb") as f:
|
|
data = f.read()
|
|
return data, mime
|
|
|
|
|
|
def build_payload(video_path: Path, duration: float) -> dict:
|
|
"""Build the OpenRouter chat completion payload with a video attachment."""
|
|
video_data, mime = read_video(video_path)
|
|
encoded = base64.b64encode(video_data).decode()
|
|
|
|
content = [
|
|
{"type": "text", "text": f"{UX_PROMPT}\n\n(Duration: {int(duration//60)}m{int(duration%60):02}s)"},
|
|
{
|
|
"type": "video_url",
|
|
"video_url": {
|
|
"url": f"data:{mime};base64,{encoded}",
|
|
},
|
|
},
|
|
]
|
|
|
|
return {
|
|
"model": DEFAULT_MODEL,
|
|
"messages": [{"role": "user", "content": content}],
|
|
"max_tokens": 8192,
|
|
"temperature": 0.3,
|
|
}
|
|
|
|
|
|
def call_openrouter(payload: dict) -> str:
|
|
"""Send request to OpenRouter and return the assistant's reply."""
|
|
headers = {
|
|
"Authorization": f"Bearer {OPENROUTER_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
"HTTP-Referer": "https://github.com/notid/e-filing",
|
|
"X-Title": "eFiling Video Analyzer",
|
|
}
|
|
|
|
with httpx.Client(timeout=300.0) as client:
|
|
resp = client.post(
|
|
f"{OPENROUTER_BASE}/chat/completions",
|
|
headers=headers,
|
|
json=payload,
|
|
)
|
|
# Print full error for debugging
|
|
if resp.status_code >= 400:
|
|
print(f" API status {resp.status_code}: {resp.text[:1000]}", file=sys.stderr)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
|
|
choices = data.get("choices", [])
|
|
if not choices:
|
|
raise ValueError(f"No choices in OpenRouter response: {json.dumps(data, indent=2)[:500]}")
|
|
return choices[0]["message"]["content"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Output
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def write_report(video_path: Path, analysis: str, model: str, duration: float) -> Path:
|
|
"""Write the analysis as a markdown file in docs/research/."""
|
|
output_dir = Path(__file__).resolve().parent.parent / "docs" / "research"
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
safe_name = re.sub(r"[^\w\s\-]", "", video_path.stem)
|
|
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d")
|
|
out_file = output_dir / f"{safe_name}_{timestamp}.md"
|
|
|
|
dur_min = int(duration // 60)
|
|
dur_sec = int(duration % 60)
|
|
header = f"""\
|
|
# eFiling — UX Analysis: {video_path.name}
|
|
|
|
| Field | Value |
|
|
|-------|-------|
|
|
| **Source video** | `{video_path.name}` |
|
|
| **Duration** | {dur_min}m {dur_sec}s |\n| **Analysis date** | {datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC")} |
|
|
| **Model** | {model} |
|
|
|
|
---
|
|
|
|
"""
|
|
out_file.write_text(header + analysis)
|
|
return out_file
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Analyze screen recordings with Gemini via OpenRouter",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""\
|
|
examples:
|
|
# analyze one specific video
|
|
python -m app.analyze_videos "videos/E-Filing in Filevine.mp4"
|
|
|
|
# analyze all videos in videos/
|
|
python -m app.analyze_videos
|
|
""",
|
|
)
|
|
parser.add_argument(
|
|
"videos",
|
|
nargs="*",
|
|
default=[],
|
|
help="Video files to analyze (defaults to all .mp4 in videos/)",
|
|
)
|
|
parser.add_argument(
|
|
"--model",
|
|
type=str,
|
|
default=os.getenv("OPENROUTER_MODEL", DEFAULT_MODEL),
|
|
help=f"OpenRouter model (default: {DEFAULT_MODEL})",
|
|
)
|
|
args = parser.parse_args()
|
|
model_override = args.model # local var avoids scoping conflict
|
|
videos_dir = Path(__file__).resolve().parent.parent / "videos"
|
|
if args.videos:
|
|
video_paths = [Path(v) for v in args.videos]
|
|
elif videos_dir.exists():
|
|
video_paths = sorted(videos_dir.glob("*"))
|
|
else:
|
|
print("No videos found. Pass paths explicitly or put files in videos/", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if not video_paths:
|
|
print("No video files to analyze.", file=sys.stderr)
|
|
sys.exit(0)
|
|
|
|
# Patch module-level so build_payload picks it up
|
|
globals()['DEFAULT_MODEL'] = model_override
|
|
print(f"Analyzing {len(video_paths)} video(s) with model '{model_override}'...")
|
|
print("Mode: full-video upload (no frame extraction)")
|
|
print()
|
|
|
|
for i, vp in enumerate(video_paths, 1):
|
|
print(f"[{i}/{len(video_paths)}] {vp.name}")
|
|
|
|
duration = probe_duration(vp)
|
|
if duration <= 0:
|
|
print(f" SKIP — could not determine duration", file=sys.stderr)
|
|
continue
|
|
|
|
try:
|
|
payload = build_payload(vp, duration)
|
|
analysis = call_openrouter(payload)
|
|
out_file = write_report(vp, analysis, args.model, duration)
|
|
print(f" ✅ Saved to {out_file}")
|
|
except Exception as exc:
|
|
print(f" ❌ Error: {exc}", file=sys.stderr)
|
|
continue
|
|
|
|
print()
|
|
|
|
print("Done.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|