118 lines
4.1 KiB
Python
118 lines
4.1 KiB
Python
"""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
|