src/storage.py

"""JSON storage helpers and the on-disk layout for analysis artifacts.

Layout
------
data/
    sample_outputs/                 # committed, precomputed demo artifacts
        <video_id>_timeline.json
        <video_id>_transcript.json
        <video_id>_visual_events.json
        <video_id>_bookmarks.json
        <video_id>_metrics.json
    work/<video_id>/                # transient artifacts from live analysis
        frames/
        audio.wav
    outputs/<video_id>/             # JSON produced by a live local run

The dashboard and API both read artifacts through the helpers here so the
storage convention lives in exactly one place.
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List, Optional

from src.config import CONFIG

# Artifact "kinds" and their filename suffixes.
ARTIFACTS = {
    "timeline": "_timeline.json",
    "transcript": "_transcript.json",
    "visual_events": "_visual_events.json",
    "bookmarks": "_bookmarks.json",
    "metrics": "_metrics.json",
}


# --------------------------------------------------------------------------- #
# Low-level JSON IO
# --------------------------------------------------------------------------- #
def read_json(path: Path) -> Any:
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def write_json(path: Path, data: Any) -> Path:
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(data, fh, ensure_ascii=False, indent=2)
    return path


# --------------------------------------------------------------------------- #
# Path resolution
# --------------------------------------------------------------------------- #
def sample_path(video_id: str, kind: str) -> Path:
    return CONFIG.sample_outputs_dir / f"{video_id}{ARTIFACTS[kind]}"


def output_path(video_id: str, kind: str) -> Path:
    return CONFIG.data_dir / "outputs" / video_id / f"{video_id}{ARTIFACTS[kind]}"


def work_dir(video_id: str) -> Path:
    return CONFIG.workdir / video_id


# --------------------------------------------------------------------------- #
# Discovery
# --------------------------------------------------------------------------- #
def list_sample_video_ids() -> List[str]:
    """Return the ids of every precomputed sample that has a timeline file."""
    d = CONFIG.sample_outputs_dir
    if not d.exists():
        return []
    ids = sorted(
        p.name[: -len(ARTIFACTS["timeline"])]
        for p in d.glob(f"*{ARTIFACTS['timeline']}")
    )
    return ids


def list_sample_videos() -> List[tuple]:
    """Return ``[(video_id, title)]`` for every sample, for friendly UI labels.

    The title is read from the sample's ``*_metrics.json``
    (``video_metadata.title``) and falls back to the id when absent.
    """
    out: List[tuple] = []
    for vid in list_sample_video_ids():
        title = vid
        metrics = load_metrics(vid)
        if metrics:
            title = (metrics.get("video_metadata") or {}).get("title") or vid
        out.append((vid, title))
    return out


def _resolve(video_id: str, kind: str) -> Optional[Path]:
    """Prefer a live local output, then fall back to the committed sample."""
    out = output_path(video_id, kind)
    if out.exists():
        return out
    sample = sample_path(video_id, kind)
    if sample.exists():
        return sample
    return None


# --------------------------------------------------------------------------- #
# Typed accessors used by the API / dashboard
# --------------------------------------------------------------------------- #
def load_artifact(video_id: str, kind: str) -> Optional[Any]:
    path = _resolve(video_id, kind)
    return read_json(path) if path else None


def load_timeline(video_id: str) -> Optional[Dict[str, Any]]:
    return load_artifact(video_id, "timeline")


def load_transcript(video_id: str) -> Optional[List[Dict[str, Any]]]:
    return load_artifact(video_id, "transcript")


def load_visual_events(video_id: str) -> Optional[List[Dict[str, Any]]]:
    return load_artifact(video_id, "visual_events")


def load_bookmarks(video_id: str) -> Optional[List[Dict[str, Any]]]:
    return load_artifact(video_id, "bookmarks")


def load_metrics(video_id: str) -> Optional[Dict[str, Any]]:
    return load_artifact(video_id, "metrics")


def save_analysis(video_id: str, artifacts: Dict[str, Any]) -> Dict[str, str]:
    """Persist a full analysis (dict keyed by artifact kind) to data/outputs."""
    written: Dict[str, str] = {}
    for kind, data in artifacts.items():
        if kind not in ARTIFACTS:
            continue
        path = write_json(output_path(video_id, kind), data)
        written[kind] = str(path)
    return written


def save_as_sample(video_id: str, artifacts: Dict[str, Any]) -> Dict[str, str]:
    """Persist a full analysis directly into ``data/sample_outputs/``.

    Same artifacts as :func:`save_analysis`, but written to the flat sample
    paths the dashboard/API serve from — i.e. "promote" a local analysis into a
    demo sample in one step.
    """
    written: Dict[str, str] = {}
    for kind, data in artifacts.items():
        if kind not in ARTIFACTS:
            continue
        path = write_json(sample_path(video_id, kind), data)
        written[kind] = str(path)
    return written