"""The video archive, read as `//.` (REQ-010…012). Reads the archive and appends user-initiated uploads and date folders (REQ-178); nothing that exists is ever moved, renamed or deleted (REQ-163/074 as revised). """ import os import re from typing import List, Optional from backend import config, db, projects, 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}") project = projects.get(project_id) if project_id is not None else None 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) rel = f"{date}/{filename}" entry = { "rel": rel, "filename": filename, "batch_label": batch_label(filename), "used_count": used.get(os.path.realpath(path), 0), } try: entry.update(video.probe(path)) if project is not None: ensure_video_preview(project, rel) 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 class LibraryConflict(LibraryError): pass def create_date(video_root: str, date: str) -> str: """Make a YYYY-MM-DD folder; never touch an existing one (REQ-178).""" if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date): raise LibraryError("Folder name must be YYYY-MM-DD") folder = _safe_join(_effective_root(video_root), date) if os.path.exists(folder): raise LibraryConflict(f"Folder already exists: {date}") try: os.makedirs(folder) except FileExistsError: # Lost the race against a concurrent mkdir; same 409 as above. raise LibraryConflict(f"Folder already exists: {date}") return folder def upload_video(video_root: str, date: str, filename: str, chunks) -> str: """Stream `chunks` into /; atomic via .part + rename.""" name = os.path.basename(filename or "") if not name or name != filename or not name.lower().endswith(video.VIDEO_EXTS): raise LibraryError(f"Unsupported file: {filename}") if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", date): raise LibraryError("Folder name must be YYYY-MM-DD") root = _effective_root(video_root) if not os.path.isdir(_safe_join(root, date)): raise LibraryError(f"No such date in the archive: {date}") dest = _safe_join(root, date, name) if os.path.exists(dest): raise LibraryConflict(f"Already exists: {date}/{name}") part = dest + ".part" try: fd = os.open(part, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644) except FileExistsError: raise LibraryConflict(f"Already exists: {date}/{name}") try: with os.fdopen(fd, "wb") as handle: for chunk in chunks: handle.write(chunk) if os.path.exists(dest): raise LibraryConflict(f"Already exists: {date}/{name}") os.replace(part, dest) except BaseException: if os.path.exists(part): try: os.unlink(part) except OSError: pass raise return f"{date}/{name}" import threading _conversion_queue = set() _conversion_lock = threading.Lock() #: Serialises preview transcodes. Each one saturates several cores on its own. _conversion_slot = threading.Semaphore(1) def ensure_video_preview(project: dict, rel: str) -> None: """Asynchronously convert video to H.264 if it's not natively web-supported.""" rel_key = rel.replace("/", "_") base, _ = os.path.splitext(rel_key) preview_filename = f"{base}.mp4" preview_dir = os.path.join(config.project_dir(project["slug"]), "previews") preview_path = os.path.join(preview_dir, preview_filename) if os.path.isfile(preview_path): return try: full_path = resolve(project["video_root"], rel) info = video.probe(full_path) if info.get("codec_name") == "h264" and full_path.lower().endswith(".mp4"): return except Exception: return with _conversion_lock: if preview_path in _conversion_queue: return _conversion_queue.add(preview_path) def _worker(): # One conversion at a time. The queue above only stops the *same* file # being converted twice; it never bounded how many ran at once, so # opening a date folder with 28 videos started 28 simultaneous x264 # encodes. That pinned every core, drove load average past 250, and # starved everything else in the process — inference included. with _conversion_slot: try: if os.path.isfile(preview_path): return os.makedirs(preview_dir, exist_ok=True) # A preview only has to be watchable in a browser, so it is not # worth `-preset medium -crf 18`: veryfast/23 encodes several # times faster for a difference nobody scrubbing footage sees. video.convert_to_h264(full_path, output_path=preview_path, crf=23, preset="veryfast") except Exception as exc: print(f"[PREVIEW CONVERSION ERROR] {rel}: {exc}") finally: with _conversion_lock: _conversion_queue.discard(preview_path) threading.Thread(target=_worker, daemon=True).start() 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 `/` 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