Blame for app.py

import hmac
import hashlib
import importlib.util
import json
import logging
import math
import os
import re
import shutil
import subprocess
import threading
import tempfile
import time
from concurrent.futures import ThreadPoolExecutor
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from functools import lru_cache
from pathlib import Path
from urllib.request import urlretrieve
from uuid import UUID, uuid5
from zipfile import ZipFile

import httpx
import numpy as np
import cv2
from email.message import Message
from email.utils import collapse_rfc2231_value
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import FileResponse, JSONResponse
from huggingface_hub import HfApi, hf_hub_download, snapshot_download
from huggingface_hub.errors import RemoteEntryNotFoundError
from storage_listing import session_files
from playback_timing import action_times, frame_times
from video_stream import video_response
from starlette.concurrency import run_in_threadpool

load_dotenv(Path(__file__).with_name(".env"))
jobs = {}
executor = ThreadPoolExecutor(max_workers=1)
logger = logging.getLogger("sporalize")
POSE_MODEL_FILE = "Weights/rtmpose-x-halpe26.onnx"
POSE_MODEL_REPO_ID = ""
POSE_MODEL_HUB_FILE = "rtmpose-x-halpe26.onnx"
POSE_MODEL_REPO_TYPE = "model"
YOLO_MODEL_FILE = "Weights/yolo11x.pt"
YOLO_MODEL_REPO_ID = "Ultralytics/YOLO11"
YOLO_MODEL_HUB_FILE = "yolo11x.pt"
YOLO_MODEL_REPO_TYPE = "model"
YOLO_IMGSZ = 896
MAX_VIDEO_SIZE_BYTES = int(os.environ.get("MAX_VIDEO_SIZE_MB", "2048")) * 1024 * 1024
MAX_ACTIVE_ANALYSES = 5
MAX_ACTIONS_SIZE_BYTES = 32 * 1024 * 1024
MAX_CALIBRATION_SIZE_BYTES = 256 * 1024 * 1024


def hub():
    return HfApi(token=os.environ["HF_TOKEN"])


def backend(path, user_id, method="GET", body=None, is_admin=False):
    for attempt in range(3):
        try:
            response = httpx.request(method, f'{os.environ["BACKEND_INTERNAL_URL"].rstrip("/")}/internal/analysis{path}',
                json=body, timeout=60, headers={"x-sporalize-proxy-secret": os.environ["ANALYSIS_PROXY_SECRET"], "x-sporalize-user-id": user_id,
                    **({"x-sporalize-admin": "true"} if is_admin else {})})
            response.raise_for_status()
            return response.json() if response.content else None
        except (httpx.TransportError, httpx.HTTPStatusError) as error:
            if attempt == 2 or isinstance(error, httpx.HTTPStatusError) and error.response.status_code < 500:
                raise
            time.sleep(2 ** attempt)


def registered_colab_worker():
    secret = os.environ.get("COLAB_WORKER_SECRET", "")
    if len(secret) < 32:
        return None
    try:
        response = httpx.get(
            f'{os.environ["BACKEND_INTERNAL_URL"].rstrip("/")}/internal/colab-worker/current',
            timeout=3,
            headers={"x-sporalize-proxy-secret": os.environ["ANALYSIS_PROXY_SECRET"]},
        )
        response.raise_for_status()
        registration = response.json()
        if not registration.get("active") or not registration.get("url"):
            return None
        base_url = registration["url"].rstrip("/")
        health = httpx.get(
            f"{base_url}/worker/health", timeout=5,
            headers={"x-sporalize-colab-secret": secret},
        )
        health.raise_for_status()
        status = health.json()
        return base_url if status.get("ready") and status.get("cuda") else None
    except (httpx.HTTPError, ValueError, KeyError) as error:
        logger.info("Colab GPU worker is not available: %s", error)
        return None


def run_pipeline_on_colab(worker_url, client_id, work, cameras, job):
    """Use HF as a temporary handoff so large videos never cross the tunnel."""
    secret = os.environ["COLAB_WORKER_SECRET"]
    headers = {"x-sporalize-colab-secret": secret}
    input_path = f"colab-staging/{client_id}"
    result_path = f"colab-results/{client_id}/reports.json"
    job.update(status="running", progress=5, phase="Preparing analysis", stage="analysis")

    try:
        hub().upload_folder(
            repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset", folder_path=str(work),
            path_in_repo=input_path, allow_patterns=["actions.json", "calibration.npz", "videos/*"],
            commit_message=f"Stage analysis inputs {client_id}", token=os.environ["HF_TOKEN"],
        )
        video_files = [{"order": order, "filename": Path(path).name} for order, path in cameras.items()]
        response = httpx.post(
            f"{worker_url}/worker/jobs",
            json={"jobId": client_id, "sessionId": job["sessionId"], "inputPath": input_path, "videos": video_files},
            headers=headers, timeout=20,
        )
        response.raise_for_status()

        failures = 0
        while True:
            if job.get("cancel"):
                try:
                    httpx.post(f"{worker_url}/worker/jobs/{client_id}/cancel", headers=headers, timeout=5)
                except httpx.HTTPError:
                    pass
                raise InterruptedError("Processing cancelled by user")

            try:
                status_response = httpx.get(f"{worker_url}/worker/jobs/{client_id}", headers=headers, timeout=10)
                status_response.raise_for_status()
                remote = status_response.json()
                failures = 0
            except (httpx.HTTPError, ValueError) as error:
                failures += 1
                if failures >= 4:
                    logger.warning("Colab worker connection was lost; using the primary analysis server: %s", error)
                    return None
                time.sleep(3)
                continue

            status = remote.get("status")
            if status == "completed":
                try:
                    downloaded = hf_hub_download(
                        repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset", filename=result_path,
                        token=os.environ["HF_TOKEN"], force_download=True,
                    )
                    reports = json.loads(Path(downloaded).read_text(encoding="utf-8"))
                    return reports if isinstance(reports, list) else None
                except Exception as error:
                    logger.warning("Colab result could not be read; using the primary analysis server: %s", error)
                    return None
            if status in {"failed", "missing"}:
                logger.warning("Colab analysis failed; using the primary analysis server: %s", remote.get("error", status))
                return None
            if status == "cancelled":
                raise InterruptedError("Processing cancelled by user")

            try:
                percentage = min(100.0, max(0.0, float(remote.get("progress", 0))))
            except (TypeError, ValueError):
                percentage = 0.0
            job.update(
                status="running", stage="analysis", progress=5 + 0.85 * percentage,
                phase=remote.get("phase") or "Analyzing session",
                step=remote.get("step", 0), total=remote.get("total", 0),
                action=remote.get("action", 0), actionsTotal=remote.get("actionsTotal", 0),
                updatedAt=int(time.time() * 1000),
            )
            time.sleep(2)
    except InterruptedError:
        raise
    except Exception as error:
        logger.warning("Colab handoff failed; using the primary analysis server: %s", error)
        return None
    finally:
        for remote_path in (input_path, f"colab-results/{client_id}"):
            try:
                hub().delete_folder(
                    repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset", path_in_repo=remote_path,
                    commit_message=f"Clean up analysis handoff {client_id}", token=os.environ["HF_TOKEN"],
                )
            except Exception as error:
                logger.info("Temporary Colab handoff cleanup skipped for %s: %s", remote_path, error)


@lru_cache(maxsize=1)
def runtime():
    root = Path(__file__).resolve().parent.parent / "Core"
    if not (root / "pipeline.py").is_file():
        if not os.environ.get("CORE_REPO_ID"):
            raise RuntimeError("CORE_REPO_ID is required when Core is not included locally")
        root = Path(snapshot_download(repo_id=os.environ["CORE_REPO_ID"], repo_type="dataset",
            revision=os.environ.get("CORE_REVISION"), token=os.environ["HF_TOKEN"],
            allow_patterns=["pipeline.py", "Weights/*"]))
    spec = importlib.util.spec_from_file_location("sporalize_core", root / "pipeline.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    pose_file = POSE_MODEL_FILE
    yolo_file = YOLO_MODEL_FILE
    pose = (root / pose_file).resolve()
    yolo = (root / yolo_file).resolve()
    if root.resolve() not in pose.parents or root.resolve() not in yolo.parents:
        raise RuntimeError("Model files must be inside the Core repository")
    if not pose.is_file():
        cache = Path(os.environ.get("HF_HOME", str(Path.home() / ".cache/huggingface"))) / "sporalize"
        cache.mkdir(parents=True, exist_ok=True)
        if not (cache / Path(pose_file).name).is_file():
            if POSE_MODEL_REPO_ID:
                downloaded = hf_hub_download(POSE_MODEL_REPO_ID, POSE_MODEL_HUB_FILE,
                    repo_type=POSE_MODEL_REPO_TYPE, token=os.environ["HF_TOKEN"])
                shutil.copyfile(downloaded, cache / Path(pose_file).name)
            elif pose_file == "Weights/rtmpose-x-halpe26.onnx":
                with tempfile.TemporaryDirectory() as temporary:
                    archive = Path(temporary) / "pose.zip"
                    urlretrieve("https://download.openmmlab.com/mmpose/v1/projects/rtmposev1/onnx_sdk/rtmpose-x_simcc-body7_pt-body7-halpe26_700e-384x288-7fb6e239_20230606.zip", archive)
                    with ZipFile(archive) as bundle:
                        member = next(name for name in bundle.namelist() if name.endswith("end2end.onnx"))
                        with bundle.open(member) as source, (cache / Path(pose_file).name).open("wb") as target:
                            shutil.copyfileobj(source, target)
            else:
                raise RuntimeError("Configured pose model is missing from Core and no pose model repository was configured")
        pose = cache / Path(pose_file).name
    if not yolo.is_file():
        yolo = Path(hf_hub_download(YOLO_MODEL_REPO_ID, YOLO_MODEL_HUB_FILE,
            repo_type=YOLO_MODEL_REPO_TYPE, token=os.environ["HF_TOKEN"]))
    return module.run_pipeline, {"POSE_PATH": str(pose), "YOLO_PATH": str(yolo)}


@asynccontextmanager
async def lifespan(app):
    for name in ("STORAGE_REPO_ID", "HF_TOKEN", "BACKEND_INTERNAL_URL"):
        if not os.environ.get(name):
            raise RuntimeError(f"{name} is required")
    if len(os.environ.get("ANALYSIS_PROXY_SECRET", "")) < 32:
        raise RuntimeError("ANALYSIS_PROXY_SECRET must contain at least 32 characters")
    if os.environ["STORAGE_REPO_ID"].lower() == "shoraky/sporalizelabs-storage":
        raise RuntimeError("Use a separate storage dataset for this application")
    if not 32 <= len(os.environ["ANALYSIS_PROXY_SECRET"]) <= 4096:
        raise RuntimeError("ANALYSIS_PROXY_SECRET length is invalid")
    runtime()
    yield


app = FastAPI(title="Sporalize AI", lifespan=lifespan)


@app.exception_handler(httpx.HTTPStatusError)
async def backend_error(request: Request, error: httpx.HTTPStatusError):
    status = error.response.status_code
    logger.error("Analysis backend request failed: path=%s status=%s actor_is_admin=%s response=%s",
        request.url.path, status, getattr(request.state, "is_admin", False), error.response.text[:500])
    try:
        backend_body = error.response.json()
    except ValueError:
        backend_body = {}
    detail = backend_body.get("error") or backend_body.get("detail") or "Backend request failed"
    payload = {"detail": detail}
    if backend_body.get("code"):
        payload["code"] = backend_body["code"]
    return JSONResponse(payload, status_code=status if status < 500 else 502)


@app.middleware("http")
async def authenticate(request: Request, call_next):
    if request.url.path != "/healthz":
        secret = os.environ.get("ANALYSIS_PROXY_SECRET", "")
        if len(secret) < 32 or not hmac.compare_digest(secret, request.headers.get("x-sporalize-proxy-secret", "")):
            return JSONResponse({"detail": "Unauthorized"}, status_code=403)
        try:
            request.state.user_id = str(UUID(request.headers.get("x-sporalize-user-id", "")))
        except ValueError:
            return JSONResponse({"detail": "Invalid user identity"}, status_code=403)
        request.state.is_admin = request.headers.get("x-sporalize-admin") == "true"
    return await call_next(request)


def json_value(value):
    if isinstance(value, np.ndarray):
        return json_value(value.tolist())
    if isinstance(value, np.generic):
        return json_value(value.item())
    if isinstance(value, dict):
        return {str(key): json_value(item) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [json_value(item) for item in value]
    if isinstance(value, float) and not math.isfinite(value):
        return None
    return value


def drive_file_id(link):
    value = link.strip()
    if re.fullmatch(r"[A-Za-z0-9_-]{10,}", value):
        return value
    if not any(host in value for host in ("drive.google.com", "drive.usercontent.google.com")):
        raise ValueError("Not a Google Drive file link")
    match = re.search(r"/file/d/([A-Za-z0-9_-]{10,})|/d/([A-Za-z0-9_-]{10,})|[?&]id=([A-Za-z0-9_-]{10,})", value)
    if not match:
        raise ValueError("Could not find a Google Drive file ID")
    return next(group for group in match.groups() if group)


def drive_file_name(content_disposition, file_id):
    message = Message()
    if content_disposition:
        message["content-disposition"] = content_disposition
    name = message.get_filename()
    if isinstance(name, tuple):
        name = collapse_rfc2231_value(name)
    name = (name or f"{file_id}.mp4").replace("\\", "/")
    name = Path(name).name
    name = "".join(character for character in name if character.isprintable()).strip()
    return name[:255] or f"{file_id}.mp4"


async def download_drive_file(link, destination):
    file_id = drive_file_id(link)
    url = f"https://drive.google.com/uc?export=download&confirm=t&id={file_id}"
    total = 0
    async with httpx.AsyncClient(follow_redirects=True, timeout=300) as client:
        async with client.stream("GET", url) as response:
            response.raise_for_status()
            content_type = response.headers.get("content-type", "").lower()
            if "text/html" in content_type:
                raise ValueError("Google Drive did not return a public video; check link access")
            with destination.open("wb") as output:
                async for chunk in response.aiter_bytes(1024 * 1024):
                    total += len(chunk)
                    if total > MAX_VIDEO_SIZE_BYTES:
                        raise HTTPException(413, "Video exceeds the configured per-file size limit")
                    output.write(chunk)
    if total == 0:
        raise ValueError("Google Drive returned an empty file")
    return file_id, total


def create_video_poster(source, destination):
    capture = cv2.VideoCapture(str(source))
    try:
        capture.set(cv2.CAP_PROP_POS_MSEC, 500)
        ok, frame = capture.read()
        if not ok:
            capture.set(cv2.CAP_PROP_POS_FRAMES, 0)
            ok, frame = capture.read()
        if not ok:
            logger.warning("Could not decode preview frame from %s", source.name)
            return False
        height, width = frame.shape[:2]
        if width > 640:
            frame = cv2.resize(frame, (640, round(height * 640 / width)), interpolation=cv2.INTER_AREA)
        ok, encoded = cv2.imencode(".webp", frame, [cv2.IMWRITE_WEBP_QUALITY, 76])
        if not ok:
            return False
        destination.write_bytes(encoded.tobytes())
        return True
    finally:
        capture.release()


def probe_video(path):
    result = subprocess.run(
        ["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)],
        check=True, capture_output=True, text=True, timeout=60,
    )
    metadata = json.loads(result.stdout)
    video = next((stream for stream in metadata.get("streams", []) if stream.get("codec_type") == "video"), None)
    if not video:
        raise RuntimeError(f"No video stream found in {path.name}")
    def numeric(value):
        try:
            parsed = float(value)
            return parsed if math.isfinite(parsed) else 0
        except (TypeError, ValueError):
            return 0

    duration = numeric(video.get("duration")) or numeric(metadata.get("format", {}).get("duration"))
    fps_text = video.get("avg_frame_rate", "0/1")
    try:
        numerator, denominator = fps_text.split("/", 1)
        denominator_value = float(denominator)
        fps = float(numerator) / denominator_value if denominator_value > 0 else 0
    except (ValueError, ZeroDivisionError):
        fps = 0
    try:
        total_frames = int(video.get("nb_frames"))
    except (TypeError, ValueError):
        total_frames = round(duration * fps)
    compatible = (
        Path(path).suffix.lower() == ".mp4"
        and video.get("codec_name") == "h264"
        and video.get("pix_fmt") == "yuv420p"
        and duration > 0
    )
    return {"duration": duration, "frames": total_frames, "compatible": compatible}


def finalize_videos(work, cameras, job):
    video_dir = work / "videos"
    inputs = []
    for order in cameras:
        matches = [path for path in video_dir.glob(f"camera_{order}.*") if path.is_file()]
        if len(matches) != 1:
            raise RuntimeError(f"Expected one source video for camera {order}")
        metadata = probe_video(matches[0])
        inputs.append((order, matches[0], metadata))

    total_duration = sum(metadata["duration"] for _, _, metadata in inputs)
    if total_duration <= 0:
        raise RuntimeError("Could not determine video duration for finalization")
    completed_duration = 0.0

    def report(order, message, local_progress, frames_done=0, frames_total=0):
        weighted = (completed_duration + inputs_by_order[order]["duration"] * local_progress) / total_duration
        job.update(
            status="running", stage="finalizing", progress=90 + 8 * weighted,
            phase=message, videoProgress=round(local_progress * 100),
            videoFramesProcessed=frames_done, videoFramesTotal=frames_total,
            updatedAt=int(time.time() * 1000),
        )

    inputs_by_order = {order: metadata for order, _, metadata in inputs}
    for index, (order, source, metadata) in enumerate(inputs, start=1):
        if job.get("cancel"):
            raise InterruptedError()
        frames_total = metadata["frames"]
        if metadata["compatible"]:
            report(order, f"Video {index}/{len(inputs)} already browser-ready", 1, 0, 0)
            completed_duration += metadata["duration"]
            continue

        target = video_dir / f"camera_{order}.mp4"
        temporary = video_dir / f"camera_{order}.finalizing.mp4"
        temporary.unlink(missing_ok=True)
        command = [
            "ffmpeg", "-hide_banner", "-loglevel", "error", "-nostats", "-y", "-i", str(source),
            "-map", "0:v:0", "-an", "-sn", "-dn", "-vf", "scale=trunc(iw/2)*2:trunc(ih/2)*2",
            "-c:v", "libx264", "-preset", "veryfast", "-crf", "23", "-pix_fmt", "yuv420p",
            "-fps_mode", "passthrough", "-movflags", "+faststart", "-progress", "pipe:1", str(temporary),
        ]
        process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1)
        stderr_lines = []
        stderr_reader = threading.Thread(target=lambda: stderr_lines.extend(process.stderr), daemon=True)
        stderr_reader.start()
        last_frames = 0
        try:
            for line in process.stdout:
                if job.get("cancel"):
                    process.terminate()
                    raise InterruptedError()
                key, separator, value = line.strip().partition("=")
                if not separator:
                    continue
                if key == "frame":
                    try:
                        last_frames = int(value)
                    except ValueError:
                        pass
                elif key == "out_time_us":
                    try:
                        local_progress = min(0.99, max(0, int(value) / 1_000_000 / metadata["duration"]))
                        report(order, f"Preparing video {index}/{len(inputs)}", local_progress, last_frames, frames_total)
                    except ValueError:
                        pass
            return_code = process.wait()
            stderr_reader.join(timeout=5)
            if return_code != 0:
                raise RuntimeError("Video conversion failed: " + "".join(stderr_lines)[-1000:])
            if not temporary.is_file() or temporary.stat().st_size == 0:
                raise RuntimeError("Video conversion produced an empty file")
            verified = probe_video(temporary)
            if not verified["compatible"]:
                raise RuntimeError("Converted video failed browser compatibility verification")
            os.replace(temporary, target)
            if source != target:
                source.unlink()
        except Exception:
            if process.poll() is None:
                process.kill()
                process.wait()
            temporary.unlink(missing_ok=True)
            raise
        report(order, f"Video {index}/{len(inputs)} verified", 1, frames_total, frames_total)
        completed_duration += metadata["duration"]


def run_job(client_id, work, cameras):
    job = jobs[client_id]
    session_id = job["sessionId"]
    saved = False
    try:
        if job.get("cancel"):
            raise InterruptedError()
        job.update(status="running", progress=5, phase="Analyzing", stage="analysis")
        worker_url = registered_colab_worker()
        reports = run_pipeline_on_colab(worker_url, client_id, work, cameras, job) if worker_url else None
        if reports is None:
            if job.get("cancel"):
                raise InterruptedError()
            job.update(status="running", progress=5, phase="Analyzing session", stage="analysis")
            pipeline, weights = runtime()

            def progress(action, total, frame, frames):
                action_number = action + 1
                frame_total = max(1, frames)
                job.update(progress=5 + 85 * (action + (frame + 1) / frame_total) / max(total, 1),
                           phase=f"Analyzing action {action_number}/{total}", step=frame + 1,
                           total=frame_total, action=action_number, actionsTotal=total,
                           updatedAt=int(time.time() * 1000))
                return not job.get("cancel")

            reports = pipeline(cameras, {**weights, "CALIBRATION_PATH": str(work / "calibration.npz"), "ACTIONS_PATH": str(work / "actions.json")}, {"YOLO_IMGSZ": YOLO_IMGSZ}, progress)
        if job.get("cancel"):
            raise InterruptedError()
        (work / "results.json").write_text(json.dumps(json_value(reports), allow_nan=False), encoding="utf-8")
        finalize_videos(work, cameras, job)
        if job.get("cancel"):
            raise InterruptedError()
        try:
            playback_boundaries = frame_times(work / "videos" / f"camera_{min(cameras)}.mp4")
        except (OSError, ValueError, subprocess.CalledProcessError):
            logger.exception("Unable to index final video timestamps for session %s", session_id)
            playback_boundaries = None
        job.update(progress=98, phase="Saving results", stage="saving", videoProgress=100)
        hub().upload_folder(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
            folder_path=str(work), path_in_repo=f"sessions/{session_id}",
            allow_patterns=["results.json", "videos/*", "posters/*.webp"])
        saved = True
        completion = {"status": "completed", "path": f"sessions/{session_id}", "actions": [
            {"index": index, **{key: json_value(report[key]) for key in ("action", "start_frame", "end_frame", "fps", "error") if key in report},
             **(action_times(playback_boundaries, int(report["start_frame"]), int(report["end_frame"]))
                if playback_boundaries and not report.get("error")
                and int(report["end_frame"]) + 1 < len(playback_boundaries) else {})}
            for index, report in enumerate(reports)]}
        job["completion"] = completion
        backend(f"/{session_id}", job.get("ownerUserId") or job["userId"], "PUT", completion, is_admin=job.get("actorIsAdmin", False))
        job.pop("completion", None)
        job.update(status="completed", phase="Completed", stage="done", progress=100, resultUrl=f"/api/sessions/{session_id}")
    except Exception as error:
        status = "cancelled" if isinstance(error, InterruptedError) else "failed"
        job.update(status=status, phase="Sync failed" if saved else status.capitalize(), stage="done" if status in {"cancelled", "failed"} else "saving", error="Files saved; retry backend sync" if saved else "Analysis cancelled" if status == "cancelled" else "Analysis or saving failed")
        if saved:
            job["status"] = "failed"
        if status == "cancelled":
            logger.info("Session %s cancelled", session_id)
        else:
            logger.exception("Session %s %s", session_id, status)
        try:
            if not saved:
                backend(f"/{session_id}", job.get("ownerUserId") or job["userId"], "PUT", {"status": status}, is_admin=job.get("actorIsAdmin", False))
        except Exception:
            logger.exception("Could not update session %s", session_id)
    finally:
        job["updatedAt"] = int(time.time() * 1000)
        shutil.rmtree(work, ignore_errors=True)
        job["done"].set()


@app.get("/healthz")
def health():
    return {"ok": True}


@app.post("/api/drive/probe")
def probe_drive(payload: dict):
    links = payload.get("links") or ([payload["link"]] if payload.get("link") else [])
    if not isinstance(links, list) or not 1 <= len(links) <= 16:
        raise HTTPException(400, "Provide between 1 and 16 Drive links")
    results = []
    for link in links:
        result = {"link": str(link)}
        try:
            file_id = drive_file_id(str(link))
            with httpx.stream("GET", f"https://drive.google.com/uc?export=download&confirm=t&id={file_id}", follow_redirects=True, timeout=30) as response:
                content_type = response.headers.get("content-type", "")
                content_disposition = response.headers.get("content-disposition", "")
                size = int(response.headers.get("content-length", "0") or 0)
                if not response.is_success or "text/html" in content_type.lower():
                    raise ValueError("File is not publicly downloadable")
            result.update(ok=True, fileId=file_id, fileName=drive_file_name(content_disposition, file_id), sizeBytes=size or None, contentType=content_type)
        except Exception as error:
            result.update(ok=False, code="unavailable", error=str(error))
        results.append(result)
    return {"results": results}


@app.post("/api/analyze", status_code=202)
async def analyze(request: Request, playerId: str = Form(...), clientId: UUID = Form(...),
                  videoOrders: list[int] = Form(...), videos: list[UploadFile] = File(None),
                  videoDriveLinks: list[str] = Form(None),
                  actionsJson: UploadFile = File(...), calibration: UploadFile = File(...),
                  sessionName: str = Form("")):
    videos = videos or []
    videoDriveLinks = videoDriveLinks or []
    if videos and videoDriveLinks or len(videos or videoDriveLinks) < 2 or len(videos or videoDriveLinks) != len(videoOrders) or len(set(videoOrders)) != len(videoOrders):
        raise HTTPException(422, "Provide at least two videos with unique camera orders")
    client_id, user_id = str(clientId), request.state.user_id
    if any(video.size is not None and video.size > MAX_VIDEO_SIZE_BYTES for video in videos):
        raise HTTPException(413, f"Each video must be smaller than {MAX_VIDEO_SIZE_BYTES // (1024 * 1024)} MB")
    if actionsJson.size is not None and actionsJson.size > MAX_ACTIONS_SIZE_BYTES:
        raise HTTPException(413, "The actions file must be 32 MB or smaller")
    if calibration.size is not None and calibration.size > MAX_CALIBRATION_SIZE_BYTES:
        raise HTTPException(413, "The calibration file must be 256 MB or smaller")
    active_jobs = [job for job in jobs.values() if job["status"] in {"uploading", "queued", "running"}]
    if any(job["userId"] == user_id for job in active_jobs):
        raise HTTPException(429, "Finish or cancel your active analysis before starting another")
    if len(active_jobs) >= MAX_ACTIVE_ANALYSES:
        raise HTTPException(429, f"The analysis queue is full (maximum {MAX_ACTIVE_ANALYSES} active jobs). Try again when a slot opens.")
    cutoff = int(time.time() * 1000) - 24 * 60 * 60 * 1000
    for key, job in list(jobs.items()):
        if job["status"] in {"completed", "failed", "cancelled", "sync_failed"} and job["updatedAt"] < cutoff:
            jobs.pop(key, None)
    if client_id in jobs:
        raise HTTPException(409, "Analysis already exists")
    reference = playerId.strip()
    if not reference or len(reference) > 100 or len(sessionName) > 150:
        raise HTTPException(422, "Invalid player reference or session name")
    session = backend("", user_id, "POST", {"clientId": client_id, "playerId": reference, "name": sessionName},
        is_admin=request.state.is_admin)
    session_id, player_id = session["session_id"], session["player_id"]
    owner_user_id = session.get("user_id") if "user_id" in session else user_id
    actor_is_admin = bool(session.get("actor_is_admin", False))
    now = int(time.time() * 1000)
    jobs[client_id] = {"clientId": client_id, "sessionId": session_id, "playerId": player_id,
        "userId": user_id, "status": "uploading", "phase": "Uploading", "stage": "upload", "progress": 0, "createdAt": now, "updatedAt": now,
        "ownerUserId": owner_user_id, "actorIsAdmin": actor_is_admin,
        "done": threading.Event(), "workPath": None, "future": None}
    work = Path(tempfile.mkdtemp(prefix="sporalize-"))
    jobs[client_id]["workPath"] = str(work)
    try:
        (work / "videos").mkdir()
        cameras = {}
        uploads = [(actionsJson, work / "actions.json", MAX_ACTIONS_SIZE_BYTES),
                   (calibration, work / "calibration.npz", MAX_CALIBRATION_SIZE_BYTES)]
        video_names = []
        sources = list(zip(videoOrders, videos)) if videos else list(zip(videoOrders, videoDriveLinks))
        for order, source in sources:
            if videos:
                video = source
                suffix = Path(video.filename or "video.mp4").suffix.lower()
            else:
                video = None
                suffix = ".mp4"
            if suffix not in {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v"}:
                raise HTTPException(422, "Unsupported video format")
            target = work / "videos" / f"camera_{order}{suffix}"
            video_names.append(target.name)
            cameras[order] = str(target)
            if video is not None:
                uploads.append((video, target, MAX_VIDEO_SIZE_BYTES))
            else:
                await download_drive_file(source, target)
        for upload, target, size_limit in uploads:
            with target.open("wb") as output:
                total = 0
                while chunk := await upload.read(1024 * 1024):
                    if jobs[client_id].get("cancel"):
                        raise InterruptedError("Analysis cancelled during upload")
                    total += len(chunk)
                    if total > size_limit:
                        raise HTTPException(413, f"{target.name} exceeds the allowed size")
                    output.write(chunk)
        poster_dir = work / "posters"
        poster_dir.mkdir()
        for order, source in cameras.items():
            create_video_poster(source, poster_dir / f"camera_{order}.webp")
        jobs[client_id].update(status="queued", phase="Queued", stage="analysis")
        jobs[client_id]["videoNames"] = video_names
        jobs[client_id]["future"] = executor.submit(run_job, client_id, work, cameras)
        return {"accepted": True, "status": "queued", "clientId": client_id, "sessionId": session_id}
    except Exception:
        jobs[client_id].update(status="failed", phase="Failed")
        shutil.rmtree(work)
        try:
            backend(f"/{session_id}", owner_user_id or user_id, "PUT", {"status": "failed"}, is_admin=actor_is_admin)
        except Exception:
            logger.exception("Could not report failed upload %s", session_id)
        finally:
            jobs[client_id]["done"].set()
        raise


def owned_job(client_id, user_id, is_admin=False):
    job = jobs.get(client_id)
    if not job or (job["userId"] != user_id and not is_admin):
        raise HTTPException(404, "Analysis not found")
    return job


@app.get("/api/progress/{client_id}")
def progress(client_id: str, request: Request):
    job = owned_job(client_id, request.state.user_id, request.state.is_admin)
    return {key: value for key, value in job.copy().items() if key not in {"userId", "cancel", "completion", "done", "workPath", "future"}}


@app.post("/api/retry-save/{client_id}")
def retry_save(client_id: str, request: Request):
    job = owned_job(client_id, request.state.user_id, request.state.is_admin)
    if job.get("completion"):
        backend(f'/{job["sessionId"]}', job.get("ownerUserId") or job["userId"], "PUT", job["completion"], is_admin=job.get("actorIsAdmin", False))
        job.pop("completion", None)
        job.update(status="completed", phase="Completed", progress=100, resultUrl=f'/api/sessions/{job["sessionId"]}')
    return progress(client_id, request)


@app.get("/api/active-analyses")
def active(request: Request):
    is_admin = request.state.is_admin
    return {"jobs": [
        {key: value for key, value in job.copy().items() if key not in {"userId", "cancel", "completion", "done", "workPath", "future"}}
        for job in list(jobs.values())
        if (is_admin or job["userId"] == request.state.user_id)
        and job["status"] in {"uploading", "queued", "running"}
    ]}


@app.post("/api/cancel/{client_id}")
def cancel(client_id: str, request: Request):
    job = owned_job(client_id, request.state.user_id, request.state.is_admin)
    if job["status"] not in {"completed", "failed", "cancelled", "sync_failed"}:
        job["cancel"] = True
    return {"status": job["status"], "clientId": client_id}


def cancel_session_job(session_id: UUID):
    matching = [(client_id, job) for client_id, job in list(jobs.items()) if job.get("sessionId") == str(session_id)]
    for client_id, job in matching:
        if job.get("status") in {"uploading", "queued", "running"}:
            job["cancel"] = True
            future = job.get("future")
            if future is not None:
                try:
                    future.result(timeout=300)
                except TimeoutError as error:
                    raise HTTPException(504, "The analysis did not stop in time; deletion was not started") from error
            else:
                job["done"].wait(timeout=300)
                if not job["done"].is_set():
                    raise HTTPException(504, "The analysis upload did not stop in time; deletion was not started")
    return [(client_id, job.get("workPath")) for client_id, job in matching]


def cleanup_deleted_session_files(session_id, storage_path, job_paths):
    for _, work_path in job_paths:
        if work_path:
            shutil.rmtree(work_path, ignore_errors=True)
    folders = ([storage_path] if storage_path else []) + [
        path for client_id, _ in job_paths
        for path in (f"colab-staging/{client_id}", f"colab-results/{client_id}")
    ]
    api = hub()
    repo_id = os.environ["STORAGE_REPO_ID"]
    failed_folders = []
    for folder in folders:
        for attempt in range(3):
            try:
                api.delete_folder(repo_id=repo_id, repo_type="dataset", path_in_repo=folder,
                    commit_message=f"Delete session {session_id} files")
                break
            except RemoteEntryNotFoundError:
                break
            except Exception:
                if attempt == 2:
                    logger.exception("Unable to delete session storage folder %s", folder)
                    failed_folders.append(folder)
                else:
                    time.sleep(2 ** attempt)
    if failed_folders:
        logger.error("Background storage cleanup incomplete for session %s: %s", session_id, failed_folders)
    else:
        logger.info("Finished background storage cleanup for session %s", session_id)


def session_row(session_id, user_id, is_admin=False, playback_times=True):
    row = backend(f"/{session_id}" + ("" if playback_times else "?playbackTimes=false"), user_id, is_admin=is_admin)
    if row["path"] != f"sessions/{session_id}":
        raise HTTPException(409, "Session uses a different storage layout")
    return row


def result_action_id(session_id, index):
    digest = hashlib.sha256(f"{session_id}:{index}".encode()).hexdigest()
    return f"{digest[:8]}-{digest[8:12]}-5{digest[13:16]}-8{digest[17:20]}-{digest[20:32]}"


def owned_session_row(session_id, user_id, is_admin=False):
    row = backend(f"/{session_id}", user_id, is_admin=is_admin)
    if row.get("user_id") != user_id and not is_admin:
        raise HTTPException(404, "Session not found")
    return row


def upload_session_results(path_in_repo, payload, commit_message):
    from io import BytesIO

    hub().upload_file(
        path_or_fileobj=BytesIO(payload),
        path_in_repo=path_in_repo,
        repo_id=os.environ["STORAGE_REPO_ID"],
        repo_type="dataset",
        commit_message=commit_message,
    )


@app.delete("/api/sessions/{session_id}/actions/{action_id}")
def delete_action(session_id: UUID, action_id: UUID, request: Request):
    row = owned_session_row(session_id, request.state.user_id, request.state.is_admin)
    if row.get("path") != f"sessions/{session_id}":
        raise HTTPException(409, "Session results are not available")

    result_path = f'{row["path"]}/results.json'
    downloaded_path = hf_hub_download(
        repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
        filename=result_path, token=os.environ["HF_TOKEN"],
    )
    original_payload = Path(downloaded_path).read_bytes()
    reports = json.loads(original_payload.decode("utf-8"))
    matching_index = next((
        index for index, report in enumerate(reports)
        if not report.get("error") and not report.get("_deleted")
        and result_action_id(session_id, index) == str(action_id)
    ), None)
    if matching_index is None:
        raise HTTPException(404, "Action not found")

    # Keep a tombstone so action IDs based on pipeline report indexes remain
    # stable for every action that follows the deleted one.
    reports[matching_index]["_deleted"] = True
    updated_reports = reports
    updated_payload = json.dumps(json_value(updated_reports), allow_nan=False).encode("utf-8")
    upload_session_results(result_path, updated_payload, f"Delete action {action_id}")
    try:
        backend(f"/{session_id}/actions/{action_id}", request.state.user_id, "DELETE", is_admin=request.state.is_admin)
    except Exception:
        try:
            upload_session_results(result_path, original_payload, f"Restore action {action_id} after delete failure")
        except Exception:
            logger.exception("Could not restore session results after action deletion failed: session=%s action=%s", session_id, action_id)
        raise
    return {"ok": True}


@app.patch("/api/sessions/{session_id}/actions/{action_id}")
def update_action(session_id: UUID, action_id: UUID, payload: dict, request: Request):
    if not request.state.is_admin:
        raise HTTPException(403, "Administrator access is required")
    row = owned_session_row(session_id, request.state.user_id, True)
    if row.get("path") != f"sessions/{session_id}":
        raise HTTPException(409, "Session results are not available")

    start_frame = payload.get("startFrame")
    end_frame = payload.get("endFrame")
    action_type = payload.get("type")
    if not isinstance(start_frame, int) or isinstance(start_frame, bool) or start_frame < 0:
        raise HTTPException(400, "Start frame must be a non-negative integer")
    if not isinstance(end_frame, int) or isinstance(end_frame, bool) or end_frame < start_frame:
        raise HTTPException(400, "End frame must be an integer greater than or equal to the start frame")
    if not isinstance(action_type, str) or not action_type.strip() or len(action_type.strip()) > 50:
        raise HTTPException(400, "Action type is invalid")

    result_path = f'{row["path"]}/results.json'
    downloaded_path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
        filename=result_path, token=os.environ["HF_TOKEN"])
    original_payload = Path(downloaded_path).read_bytes()
    reports = json.loads(original_payload.decode("utf-8"))
    matching_index = next((index for index, report in enumerate(reports)
        if not report.get("error") and not report.get("_deleted")
        and result_action_id(session_id, index) == str(action_id)), None)
    if matching_index is None:
        raise HTTPException(404, "Action not found")
    video_dir = f'{row["path"]}/videos/'
    video_files = sorted(file for file in session_files(hub(), os.environ["STORAGE_REPO_ID"], video_dir.rstrip("/"))
                         if re.search(r"/camera_\d+\.mp4$", file))
    if not video_files:
        raise HTTPException(409, "Session video is unavailable")
    video_path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
                                 filename=video_files[0], token=os.environ["HF_TOKEN"])
    playback_times = action_times(frame_times(video_path), start_frame, end_frame)
    reports[matching_index]["start_frame"] = start_frame
    reports[matching_index]["end_frame"] = end_frame
    reports[matching_index]["action"] = action_type.strip()
    updated_payload = json.dumps(json_value(reports), allow_nan=False).encode("utf-8")
    upload_session_results(result_path, updated_payload, f"Update action {action_id}")
    try:
        backend(f"/{session_id}/actions/{action_id}", request.state.user_id, "PATCH", {
            "startFrame": start_frame, "endFrame": end_frame, "type": action_type.strip(),
            "startTimeSeconds": playback_times["start_time_seconds"],
            "endTimeSeconds": playback_times["end_time_seconds"],
        }, is_admin=True)
    except Exception:
        try:
            upload_session_results(result_path, original_payload, f"Restore action {action_id} after update failure")
        except Exception:
            logger.exception("Could not restore session results after action update failed: session=%s action=%s", session_id, action_id)
        raise
    return {"ok": True}


@app.delete("/api/sessions/{session_id}")
def delete_session(session_id: UUID, request: Request, background_tasks: BackgroundTasks):
    row = owned_session_row(session_id, request.state.user_id, request.state.is_admin)
    expected_path = f"sessions/{session_id}"
    storage_path = row.get("path")
    if storage_path is not None and storage_path != expected_path:
        raise HTTPException(409, "Session uses a different storage layout")

    job_paths = cancel_session_job(session_id)
    backend(f"/{session_id}", request.state.user_id, "DELETE", is_admin=request.state.is_admin)
    for client_id, _ in job_paths:
        jobs.pop(client_id, None)
    # The database deletion makes the session disappear before Hugging Face finishes deleting its files.
    background_tasks.add_task(cleanup_deleted_session_files, session_id, storage_path, job_paths)
    return {"ok": True}


@app.get("/api/sessions/{session_id}")
def results(session_id: UUID, request: Request):
    row = session_row(session_id, request.state.user_id, request.state.is_admin)
    path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
        filename=f'{row["path"]}/results.json', token=os.environ["HF_TOKEN"])
    reports = json.loads(Path(path).read_text(encoding="utf-8"))
    video_dir = f'{row["path"]}/videos/'
    prefix = f'{video_dir}camera_'
    video_extensions = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".m4v"}
    camera_files = []
    for repo_file in session_files(hub(), os.environ["STORAGE_REPO_ID"], video_dir.rstrip("/")):
        if not repo_file.startswith(prefix):
            continue
        filename = repo_file[len(video_dir):]
        camera_id_text = Path(filename).stem.removeprefix("camera_")
        if Path(filename).suffix.lower() in video_extensions and camera_id_text.isdigit():
            camera_files.append((int(camera_id_text), filename))
    camera_files.sort()
    fps_values = [float(report.get("fps") or 30) for report in reports if report.get("fps")]
    saved_times = {item["action_id"]: item for item in row.get("action_times", [])}
    actions, failed = [], []
    for index, report in enumerate(reports):
        if report.get("_deleted"):
            continue
        start = int(report.get("start_frame") or 0)
        end = int(report.get("end_frame") or start)
        action_fps = float(report.get("fps") or (fps_values[0] if fps_values else 30))
        label = report.get("action", "Unknown")
        if report.get("error"):
            failed.append({"id": str(uuid5(session_id, str(index))), "label": label, "start": str(start), "end": str(end), "error": report["error"]})
            continue
        analytics = report.get("analytics") or {}
        metrics = [{"name": key.replace("_", " ").title(), "value": value, "unit": ""} for key, value in analytics.items() if isinstance(value, (int, float, str, bool)) or value is None]
        full_interval = report.get("full_interval") or {}
        telemetry_names = {
            "head_angle": ("Head Angle", "°"),
            "trunk_lean": ("Trunk Lean", "°"),
            "left_knee_angle": ("Left Knee Angle", "°"),
            "right_knee_angle": ("Right Knee Angle", "°"),
            "active_ball_dist": ("Active Foot to Ball Distance", "cm"),
            "ball_speed": ("Ball Speed", "cm/s"),
            "active_heel_speed": ("Active Heel Speed", "cm/s"),
            "feet_distance": ("Feet Distance", "cm"),
            "left_heel_height": ("Left Heel Height", "cm"),
            "right_heel_height": ("Right Heel Height", "cm"),
        }
        full_interval_metrics = []
        metric_frames = {}
        for frame_text, frame_data in full_interval.items():
            if not isinstance(frame_data, dict):
                continue
            try:
                frame_number = int(frame_text) - start
            except (TypeError, ValueError):
                continue
            if frame_number < 0 or frame_number >= max(0, end - start + 1):
                continue
            for key, (metric_name, unit) in telemetry_names.items():
                value = frame_data.get(key)
                if isinstance(value, (int, float)) and math.isfinite(value):
                    metric_frames.setdefault(key, []).append({"frame": frame_number, "value": value})
        for key, values in metric_frames.items():
            if values:
                full_interval_metrics.append({"name": telemetry_names[key][0], "unit": telemetry_names[key][1], "values": values})
        clips = [{"cameraId": camera_id,
            "url": f"/storage/{session_id}/videos/{filename}",
            "poster": f"/storage/{session_id}/posters/camera_{camera_id}.webp"}
            for camera_id, filename in camera_files]
        action_id = result_action_id(session_id, index)
        playback = saved_times.get(action_id) or {}
        start_seconds = playback.get("start_time_seconds")
        end_seconds = playback.get("end_time_seconds")
        has_playback_times = (isinstance(start_seconds, (int, float)) and isinstance(end_seconds, (int, float))
                              and math.isfinite(start_seconds) and math.isfinite(end_seconds)
                              and start_seconds >= 0 and end_seconds > start_seconds)
        actions.append({"id": action_id, "label": label,
            "start": str(start), "end": str(end), "fps": action_fps,
            "startFrame": start, "endFrame": end, "startSeconds": start_seconds if has_playback_times else start / action_fps,
            "endSeconds": end_seconds if has_playback_times else (end + 1) / action_fps,
            "hasPlaybackTimes": has_playback_times, "totalFrames": max(0, end - start + 1),
            "preFrames": 0, "inFrame": max(0, (end - start) // 2), "postFrames": 0,
            "cameraClips": clips, "sourceCameraClips": [], "preMetrics": [],
            "preActionMetrics": [], "inActionMetrics": metrics, "postMetrics": [],
            "postActionMetrics": [], "fullIntervalMetrics": full_interval_metrics,
            "skeleton": report.get("skeleton_history", []), "rawAnalytics": analytics})
    created = row.get("created_at")
    if created:
        created_iso = created.replace("Z", "+00:00")
        # Python 3.10's fromisoformat requires a six-digit fractional second,
        # while the API may return a shorter fraction such as `.89`.
        created_iso = re.sub(
            r"(?<=\.)\d+(?=(?:[+-]\d{2}:\d{2})?$)",
            lambda match: match.group(0)[:6].ljust(6, "0"),
            created_iso,
        )
        created_at = datetime.fromisoformat(created_iso)
    else:
        created_at = datetime.now(timezone.utc)
    if created_at.tzinfo is None:
        created_at = created_at.replace(tzinfo=timezone.utc)
    return {"id": str(session_id), "playerId": row["player_id"], "playerName": "",
        "name": row.get("name", "Analysis session"),
        "byAdmin": bool(row.get("by_admin", False)),
        "isOwner": row.get("user_id") == request.state.user_id,
        "createdAt": int(created_at.timestamp() * 1000),
        "targetSize": [0, 0], "cameraCount": len(camera_files), "actions": actions, "failedActions": failed}


@app.api_route("/storage/{session_id}/videos/{filename}", methods=["GET", "HEAD"])
async def video(session_id: UUID, filename: str, request: Request):
    if Path(filename).name != filename or "\\" in filename:
        raise HTTPException(404, "Video not found")
    row = await run_in_threadpool(session_row, session_id, request.state.user_id, request.state.is_admin, False)
    return await video_response(request, os.environ["STORAGE_REPO_ID"],
                                f'{row["path"]}/videos/{filename}', os.environ["HF_TOKEN"])


@app.api_route("/storage/{session_id}/posters/{filename}", methods=["GET", "HEAD"])
def video_poster(session_id: UUID, filename: str, request: Request):
    row = session_row(session_id, request.state.user_id, request.state.is_admin, playback_times=False)
    if Path(filename).name != filename or "\\" in filename or not filename.endswith(".webp"):
        raise HTTPException(404, "Preview not found")
    try:
        path = hf_hub_download(repo_id=os.environ["STORAGE_REPO_ID"], repo_type="dataset",
            filename=f'{row["path"]}/posters/{filename}', token=os.environ["HF_TOKEN"])
    except RemoteEntryNotFoundError as error:
        raise HTTPException(404, "Preview not found") from error
    return FileResponse(path, media_type="image/webp", headers={"Cache-Control": "private, max-age=86400"})