feat: setup dataset enrichment app codebase and scripts

backend/library.py

@@ -0,0 +1,117 @@
+"""The video archive, read as `<video_root>/<date>/<batch>.<ext>` (REQ-010…012).
+
+Read-only by construction: this module only ever lists and stats files, never
+writes into the user's recordings (REQ-074).
+"""
+
+import os
+import re
+from typing import List, Optional
+
+from backend import config, db, video
+
+
+class LibraryError(Exception):
+    pass
+
+
+def _safe_join(video_root: str, *parts: str) -> str:
+    """Join under the archive root, refusing anything that escapes it."""
+    root = os.path.realpath(video_root)
+    target = os.path.realpath(os.path.join(root, *parts))
+    if target != root and not target.startswith(root + os.sep):
+        raise LibraryError("Path is outside the video archive")
+    return target
+
+
+def batch_label(filename: str) -> str:
+    """`batch-4.mp4` -> `batch-4`. The filename is the batch's identity."""
+    return os.path.splitext(filename)[0]
+
+
+def _batch_sort_key(filename: str):
+    """Sort batch-2 before batch-10, and keep unnumbered names after them."""
+    numbers = re.findall(r"\d+", batch_label(filename))
+    return (0, int(numbers[0])) if numbers else (1, 0), filename.lower()
+
+
+def _effective_root(video_root: str) -> str:
+    """Return video_root if it exists, otherwise fall back to config.VIDEO_ROOT."""
+    if os.path.isdir(video_root):
+        return video_root
+    if os.path.isdir(config.VIDEO_ROOT):
+        return config.VIDEO_ROOT
+    return video_root
+
+
+def list_dates(video_root: str) -> List[dict]:
+    """Date folders, newest name first, with how many videos each holds."""
+    video_root = _effective_root(video_root)
+    if not os.path.isdir(video_root):
+        raise LibraryError(f"Video archive folder not found: {video_root}")
+
+    dates = []
+    for name in sorted(os.listdir(video_root), reverse=True):
+        path = os.path.join(video_root, name)
+        if not os.path.isdir(path) or name.startswith("."):
+            continue
+        try:
+            count = sum(1 for f in os.listdir(path) if f.lower().endswith(video.VIDEO_EXTS))
+        except OSError:
+            continue
+        dates.append({"date": name, "video_count": count})
+    return dates
+
+
+def list_videos(video_root: str, date: str, project_id: Optional[int] = None) -> List[dict]:
+    """Videos in one date folder, with metadata and how often each was used."""
+    video_root = _effective_root(video_root)
+    folder = _safe_join(video_root, date)
+    if not os.path.isdir(folder):
+        raise LibraryError(f"No such date in the archive: {date}")
+
+    used = _usage(project_id)
+    videos = []
+    for filename in sorted(
+        (f for f in os.listdir(folder) if f.lower().endswith(video.VIDEO_EXTS)),
+        key=_batch_sort_key,
+    ):
+        path = os.path.join(folder, filename)
+        entry = {
+            "rel": f"{date}/{filename}",
+            "filename": filename,
+            "batch_label": batch_label(filename),
+            "used_count": used.get(os.path.realpath(path), 0),
+        }
+        try:
+            entry.update(video.probe(path))
+        except video.VideoError as exc:
+            # A file ffprobe cannot read still belongs in the list, flagged —
+            # hiding it would look like the archive is missing recordings.
+            entry.update({"duration": 0.0, "width": 0, "height": 0, "fps": 0.0,
+                          "error": str(exc)})
+        videos.append(entry)
+    return videos
+
+
+def _usage(project_id: Optional[int]) -> dict:
+    """How many batches already came out of each video path (REQ-012)."""
+    if project_id is None:
+        return {}
+    with db.cursor() as cur:
+        cur.execute(
+            "SELECT video_path, COUNT(*) FROM batches WHERE project_id = ? GROUP BY video_path",
+            (project_id,),
+        )
+        return {os.path.realpath(row[0]): row[1] for row in cur.fetchall()}
+
+
+def resolve(video_root: str, rel: str) -> str:
+    """Turn a `<date>/<file>` reference into an absolute path inside the archive."""
+    video_root = _effective_root(video_root)
+    path = _safe_join(video_root, rel)
+    if not os.path.isfile(path):
+        raise LibraryError(f"No such video: {rel}")
+    if not path.lower().endswith(video.VIDEO_EXTS):
+        raise LibraryError("That file is not a video")
+    return path