run_analysis.py

"""CLI for running the local analysis pipeline on a video file.

Examples
--------
    # Analyze a local video at the default 2s frame interval:
    python run_analysis.py path/to/lecture.mp4

    # Analyze straight from a URL (YouTube, Vimeo, direct link, ...):
    python run_analysis.py "https://www.youtube.com/watch?v=XXXXXXXXXXX"

    # Custom id + frame interval, and print the resulting bookmarks:
    python run_analysis.py lecture.mp4 --video-id my_video --interval 3 --show-bookmarks

Artifacts are written to ``data/outputs/<video_id>/``. Requires the local
dependencies (``pip install -r requirements-local.txt``) and ffmpeg. URL
ingestion additionally uses yt-dlp — only download videos you have the right to
use, and respect each platform's Terms of Service.
"""

from __future__ import annotations

import argparse
import json

from src.config import get_config
from src.video_source import is_url


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Analyze a short video locally, from a file path or a URL."
    )
    parser.add_argument("source", type=str, help="Local video path or http(s) URL.")
    parser.add_argument("--video-id", type=str, default=None, help="Output id.")
    parser.add_argument(
        "--interval", type=float, default=None, help="Frame sampling interval (s)."
    )
    parser.add_argument(
        "--show-bookmarks", action="store_true", help="Print bookmarks at the end."
    )
    parser.add_argument(
        "--as-sample",
        action="store_true",
        help="Promote the result straight into data/sample_outputs/ as a demo "
        "sample (instead of data/outputs/), so the dashboard/Space can serve it.",
    )
    args = parser.parse_args()

    from pathlib import Path

    if not is_url(args.source) and not Path(args.source).exists():
        raise SystemExit(f"Video not found (and not a URL): {args.source}")

    # Force live mode for the CLI regardless of demo defaults.
    config = get_config()
    config.demo_mode = False
    config.use_precomputed = False
    if args.interval is not None:
        config.frame_interval_sec = args.interval

    from src import storage
    from src.pipeline import analyze_video  # lazy import of heavy deps

    # When promoting to a sample we persist ourselves (to sample_outputs).
    artifacts = analyze_video(
        args.source, video_id=args.video_id, config=config, persist=not args.as_sample
    )
    metrics = artifacts["metrics"]
    vid = metrics["video_id"]

    print(f"\n✅ Analysis complete for '{vid}'")
    print(f"   total processing: {metrics['total_processing_sec']} s")
    print(f"   counts: {json.dumps(metrics['counts'])}")

    if args.as_sample:
        storage.save_as_sample(vid, artifacts)
        print(f"\n📦 Promoted to demo sample: data/sample_outputs/{vid}_*.json")
        print("   To publish it to your Space (only for CC/public-domain/your-own")
        print("   /permissioned content):")
        print(f"     git add data/sample_outputs/{vid}_*.json")
        print(f'     git commit -m "Add sample: {vid}"')
        print("     git push hf main")
    else:
        print(f"   output dir: data/outputs/{vid}/")

    if args.show_bookmarks:
        print("\nBookmarks:")
        for b in artifacts["bookmarks"]:
            print(f"  {b['timestamp']}  {b['title']}  ({b['reason']})")


if __name__ == "__main__":
    main()