Files
oscars-pool/video_server.py
T

632 lines
20 KiB
Python

import mimetypes
import os
import re
import shutil
import subprocess
import time
import hashlib
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
from pathlib import Path
from typing import Any, Dict, Generator, Iterable, List, Optional, Tuple
import threading
import jwt
from fastapi import FastAPI, HTTPException, Query, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, StreamingResponse
APP_NAME = "video-file-server"
BASE_DIR = Path(__file__).resolve().parent
MEDIA_DIR = BASE_DIR / "media"
HLS_CACHE_DIR = Path(os.getenv("HLS_CACHE_DIR", str(BASE_DIR / ".hls_cache"))).resolve()
# Load repo .env (shared with the Node app) if present.
try:
from dotenv import load_dotenv # type: ignore
load_dotenv(BASE_DIR / ".env")
except Exception:
pass
# Allow common video types (what browsers can play is separate from what we serve).
ALLOWED_EXTS = {".mp4", ".mkv"}
RANGE_RE = re.compile(r"^bytes=(\d*)-(\d*)$")
AUTH_REQUIRED = os.getenv("VIDEO_AUTH_REQUIRED", "1").strip() not in {"0", "false", "False", "no", "NO"}
JWT_SECRET = os.getenv("JWT_SECRET")
# Limit concurrent ffmpeg transcodes (HLS generation) to avoid CPU overload.
HLS_MAX_JOBS = max(1, int(os.getenv("HLS_MAX_JOBS", "2").strip() or "2"))
_HLS_JOB_SEM = threading.Semaphore(HLS_MAX_JOBS)
def _safe_resolve_media_path(rel_path: str) -> Path:
"""
Resolve a user-provided path against MEDIA_DIR, preventing path traversal.
"""
# Normalize separators and strip leading slashes.
rel_path = rel_path.replace("\\", "/").lstrip("/")
if not rel_path:
raise HTTPException(status_code=404, detail="Missing file path")
candidate = (MEDIA_DIR / rel_path).resolve()
media_root = MEDIA_DIR.resolve()
try:
candidate.relative_to(media_root)
except ValueError as exc:
raise HTTPException(status_code=400, detail="Invalid path") from exc
if not candidate.is_file():
raise HTTPException(status_code=404, detail="File not found")
if candidate.suffix.lower() not in ALLOWED_EXTS:
raise HTTPException(status_code=415, detail="Unsupported file type")
return candidate
def _extract_token(request: Request) -> Optional[str]:
# Query-param token (works for <video> and Safari HLS)
q = request.query_params.get("token")
if q:
return q
# Authorization: Bearer <token> (works for fetch/hls.js)
auth = request.headers.get("authorization") or request.headers.get("Authorization")
if auth and auth.lower().startswith("bearer "):
return auth.split(" ", 1)[1].strip() or None
return None
def _require_auth(request: Request) -> dict:
if not AUTH_REQUIRED:
return {}
token = _extract_token(request)
if not token:
raise HTTPException(status_code=401, detail="Authentication required")
if not JWT_SECRET:
raise HTTPException(status_code=500, detail="Server misconfigured: JWT_SECRET is not set")
try:
return jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
except jwt.ExpiredSignatureError as exc:
raise HTTPException(status_code=401, detail="Invalid or expired token") from exc
except jwt.InvalidTokenError as exc:
raise HTTPException(status_code=401, detail="Invalid or expired token") from exc
def _guess_content_type(path: Path) -> str:
ext = path.suffix.lower()
if ext == ".mkv":
return "video/x-matroska"
# Let mimetypes handle mp4 and others.
ctype, _ = mimetypes.guess_type(str(path))
return ctype or "application/octet-stream"
def _parse_range(range_header: str, size: int) -> Optional[Tuple[int, int]]:
"""
Parse a single HTTP Range header of form: bytes=start-end
Returns (start, end) inclusive, or None if header is missing/empty.
Raises HTTPException for invalid ranges.
"""
if not range_header:
return None
m = RANGE_RE.match(range_header.strip())
if not m:
raise HTTPException(status_code=416, detail="Invalid Range header")
start_s, end_s = m.group(1), m.group(2)
if start_s == "" and end_s == "":
raise HTTPException(status_code=416, detail="Invalid Range header")
if start_s == "":
# suffix range: last N bytes
suffix_len = int(end_s)
if suffix_len <= 0:
raise HTTPException(status_code=416, detail="Invalid Range header")
if suffix_len > size:
return (0, size - 1)
return (size - suffix_len, size - 1)
start = int(start_s)
end = int(end_s) if end_s != "" else size - 1
if start >= size or start < 0:
raise HTTPException(status_code=416, detail="Range not satisfiable")
if end < start:
raise HTTPException(status_code=416, detail="Range not satisfiable")
end = min(end, size - 1)
return (start, end)
def _iter_file(path: Path, start: int, end: int, chunk_size: int = 1024 * 1024) -> Generator[bytes, None, None]:
with path.open("rb") as f:
f.seek(start)
remaining = end - start + 1
while remaining > 0:
chunk = f.read(min(chunk_size, remaining))
if not chunk:
break
remaining -= len(chunk)
yield chunk
def _hls_content_type(path: Path) -> str:
ext = path.suffix.lower()
if ext == ".m3u8":
return "application/vnd.apple.mpegurl"
if ext == ".ts":
return "video/mp2t"
ctype, _ = mimetypes.guess_type(str(path))
return ctype or "application/octet-stream"
def _hls_id_for_media(path: Path) -> str:
rel = path.relative_to(MEDIA_DIR).as_posix()
st = path.stat()
key = f"{rel}|{st.st_size}|{st.st_mtime_ns}".encode("utf-8")
return hashlib.sha1(key).hexdigest()
def _probe_streams(path: Path) -> Dict[str, Any]:
"""
Returns basic info about the input using ffprobe: {width,height,has_audio}.
"""
cmd = [
"ffprobe",
"-v",
"error",
"-select_streams",
"v:0",
"-show_entries",
"stream=width,height",
"-of",
"csv=p=0:s=x",
str(path),
]
try:
p = subprocess.run(cmd, check=True, capture_output=True, text=True)
wh = (p.stdout or "").strip()
except FileNotFoundError as exc:
raise HTTPException(status_code=500, detail="ffprobe not installed") from exc
except subprocess.CalledProcessError as exc:
stderr = (exc.stderr or "").strip()
raise HTTPException(status_code=500, detail=f"ffprobe failed: {stderr or 'ffprobe error'}") from exc
width = height = None
if "x" in wh:
try:
w_s, h_s = wh.split("x", 1)
width, height = int(w_s), int(h_s)
except Exception:
width = height = None
cmd_a = [
"ffprobe",
"-v",
"error",
"-select_streams",
"a:0",
"-show_entries",
"stream=index",
"-of",
"csv=p=0",
str(path),
]
try:
pa = subprocess.run(cmd_a, check=False, capture_output=True, text=True)
has_audio = bool((pa.stdout or "").strip())
except FileNotFoundError as exc:
raise HTTPException(status_code=500, detail="ffprobe not installed") from exc
if not width or not height:
raise HTTPException(status_code=415, detail="Unsupported media: missing video stream")
return {"width": width, "height": height, "has_audio": has_audio}
def _acquire_lock(lock_path: Path, timeout_s: int = 600) -> None:
"""
Simple file lock to avoid concurrent transcodes for the same id.
"""
started = time.time()
while True:
try:
fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
try:
os.write(fd, str(os.getpid()).encode("utf-8"))
finally:
os.close(fd)
return
except FileExistsError:
if time.time() - started > timeout_s:
raise HTTPException(status_code=409, detail="HLS generation already running")
time.sleep(0.25)
def _release_lock(lock_path: Path) -> None:
try:
lock_path.unlink(missing_ok=True)
except Exception:
# Best-effort; lock will be overwritten on next run if needed.
pass
def _ensure_hls_for_media(media_path: Path) -> str:
"""
Generates an HLS VOD playlist for a given media file in HLS_CACHE_DIR.
Returns the HLS id.
"""
hls_id = _hls_id_for_media(media_path)
out_dir = HLS_CACHE_DIR / hls_id
index_path = out_dir / "index.m3u8"
if index_path.exists():
return hls_id
HLS_CACHE_DIR.mkdir(parents=True, exist_ok=True)
lock_path = HLS_CACHE_DIR / f"{hls_id}.lock"
_acquire_lock(lock_path)
try:
# Another process might have finished while we waited.
if index_path.exists():
return hls_id
tmp_dir = HLS_CACHE_DIR / f"{hls_id}.tmp"
if tmp_dir.exists():
shutil.rmtree(tmp_dir, ignore_errors=True)
tmp_dir.mkdir(parents=True, exist_ok=True)
# Multi-rendition adaptive HLS (master + variants).
hls_time = os.getenv("HLS_TIME", "6").strip()
crf = os.getenv("HLS_CRF", "23").strip()
preset = os.getenv("HLS_PRESET", "veryfast").strip()
info = _probe_streams(media_path)
src_h = int(info["height"])
has_audio = bool(info["has_audio"])
# Variant syntax: "height:video_kbps,height:video_kbps,..."
default_variants = "1080:5500,720:3000,480:1500,360:900"
variants_env = os.getenv("HLS_VARIANTS", default_variants).strip()
variants: List[Tuple[int, int]] = []
for part in [p.strip() for p in variants_env.split(",") if p.strip()]:
if ":" not in part:
continue
h_s, b_s = part.split(":", 1)
try:
h = int(h_s)
b = int(b_s)
except ValueError:
continue
if h <= 0 or b <= 0:
continue
if h <= src_h:
variants.append((h, b))
if not variants:
variants = [(min(720, src_h), 3000)]
variants = sorted(list({v for v in variants}), key=lambda x: x[0], reverse=True)
n = len(variants)
# Build filter_complex split+scale for N renditions
split_tags = "".join([f"[v{i}]" for i in range(n)])
scale_filters = [f"[v{i}]scale=-2:{variants[i][0]}[v{i}o]" for i in range(n)]
filter_complex = f"[0:v]split={n}{split_tags};" + ";".join(scale_filters)
cmd: List[str] = [
"ffmpeg",
"-hide_banner",
"-loglevel",
"error",
"-y",
"-i",
str(media_path),
"-filter_complex",
filter_complex,
]
for i in range(n):
cmd += ["-map", f"[v{i}o]"]
if has_audio:
cmd += ["-map", "0:a?"]
cmd += ["-pix_fmt", "yuv420p"]
audio_bitrate = os.getenv("HLS_AUDIO_BITRATE", "160k").strip()
for i, (_h, b_kbps) in enumerate(variants):
maxrate = int(b_kbps * 1.07)
bufsize = int(b_kbps * 1.5)
cmd += [
f"-c:v:{i}",
"libx264",
"-preset",
preset,
"-crf",
crf,
f"-b:v:{i}",
f"{b_kbps}k",
f"-maxrate:v:{i}",
f"{maxrate}k",
f"-bufsize:v:{i}",
f"{bufsize}k",
"-sc_threshold",
"0",
]
if has_audio:
cmd += [f"-c:a:{i}", "aac", f"-b:a:{i}", audio_bitrate]
cmd += ["-force_key_frames", f"expr:gte(t,n_forced*{hls_time})"]
cmd += [
"-f",
"hls",
"-hls_time",
hls_time,
"-hls_playlist_type",
"vod",
"-hls_flags",
"independent_segments",
"-hls_segment_filename",
str(tmp_dir / "seg_%v_%05d.ts"),
"-master_pl_name",
"index.m3u8",
]
if has_audio:
var_map = " ".join([f"v:{i},a:{i}" for i in range(n)])
else:
var_map = " ".join([f"v:{i}" for i in range(n)])
cmd += ["-var_stream_map", var_map, str(tmp_dir / "v%v.m3u8")]
try:
with _HLS_JOB_SEM:
subprocess.run(cmd, check=True, capture_output=True)
except FileNotFoundError as exc:
raise HTTPException(status_code=500, detail="ffmpeg not installed") from exc
except subprocess.CalledProcessError as exc:
stderr = (exc.stderr or b"").decode("utf-8", errors="replace").strip()
raise HTTPException(status_code=500, detail=f"HLS generation failed: {stderr or 'ffmpeg error'}") from exc
# Move into place.
if out_dir.exists():
# Race: someone else created it; keep theirs.
shutil.rmtree(tmp_dir, ignore_errors=True)
return hls_id
os.replace(str(tmp_dir), str(out_dir))
return hls_id
finally:
_release_lock(lock_path)
def _safe_resolve_hls_asset(hls_id: str, asset_path: str) -> Path:
if not re.fullmatch(r"[a-f0-9]{40}", hls_id):
raise HTTPException(status_code=400, detail="Invalid HLS id")
asset_path = asset_path.replace("\\", "/").lstrip("/")
if not asset_path:
raise HTTPException(status_code=404, detail="Missing asset path")
root = (HLS_CACHE_DIR / hls_id).resolve()
candidate = (root / asset_path).resolve()
try:
candidate.relative_to(root)
except ValueError as exc:
raise HTTPException(status_code=400, detail="Invalid asset path") from exc
if not candidate.is_file():
raise HTTPException(status_code=404, detail="Asset not found")
# Restrict to typical HLS outputs.
if candidate.suffix.lower() not in {".m3u8", ".ts"}:
raise HTTPException(status_code=415, detail="Unsupported asset type")
return candidate
def _with_token_query(uri: str, token: str) -> str:
parts = urlsplit(uri)
q = dict(parse_qsl(parts.query, keep_blank_values=True))
if "token" not in q:
q["token"] = token
return urlunsplit((parts.scheme, parts.netloc, parts.path, urlencode(q), parts.fragment))
def _rewrite_m3u8_with_token(content: str, token: str) -> str:
"""
Adds ?token=... to each URI line (segments, key URIs, etc.) in a playlist.
Only touches non-comment, non-empty lines.
"""
out_lines: list[str] = []
for raw_line in content.splitlines(True):
line = raw_line.strip()
if not line or line.startswith("#"):
out_lines.append(raw_line)
continue
# Preserve original newline by rewriting only the URI portion.
uri = line
# If the line already includes leading/trailing whitespace, normalize to original.
rewritten = _with_token_query(uri, token)
out_lines.append(raw_line.replace(uri, rewritten, 1))
return "".join(out_lines)
app = FastAPI(title=APP_NAME)
# Default to permissive CORS because the player may be on another port/host.
# Configure with env: CORS_ORIGINS="http://localhost:5000,http://127.0.0.1:5000"
cors_origins_env = os.getenv("CORS_ORIGINS", "*").strip()
if cors_origins_env == "*":
allow_origins: Iterable[str] = ["*"]
else:
allow_origins = [o.strip() for o in cors_origins_env.split(",") if o.strip()]
app.add_middleware(
CORSMiddleware,
allow_origins=list(allow_origins),
allow_credentials=False,
allow_methods=["GET", "HEAD", "OPTIONS"],
allow_headers=["*"],
)
@app.get("/health")
def health() -> dict:
return {"ok": True, "app": APP_NAME}
@app.get("/api/media")
def list_media(request: Request) -> JSONResponse:
_require_auth(request)
MEDIA_DIR.mkdir(parents=True, exist_ok=True)
items = []
for p in sorted(MEDIA_DIR.rglob("*")):
if not p.is_file():
continue
if p.suffix.lower() not in ALLOWED_EXTS:
continue
rel = p.relative_to(MEDIA_DIR).as_posix()
items.append(
{
"name": rel,
"size": p.stat().st_size,
"url": f"/media/{rel}",
}
)
return JSONResponse({"items": items})
@app.get("/api/hls")
def prepare_hls(
request: Request,
source: str = Query(..., description="Path relative to ./media (e.g. MyMovie.mkv)"),
) -> JSONResponse:
"""
Generates (and caches) an HLS VOD playlist for a local media file.
Returns an id and a playlist URL like: /hls/<id>/index.m3u8
"""
_require_auth(request)
media_path = _safe_resolve_media_path(source)
hls_id = _ensure_hls_for_media(media_path)
base = str(request.base_url).rstrip("/")
playlist_path = f"/hls/{hls_id}/index.m3u8"
return JSONResponse(
{
"hls_id": hls_id,
"playlist_url": playlist_path,
"playlist_abs_url": f"{base}{playlist_path}",
}
)
@app.api_route("/hls/{hls_id}/{asset_path:path}", methods=["GET", "HEAD"])
async def serve_hls_asset(hls_id: str, asset_path: str, request: Request) -> Response:
_require_auth(request)
path = _safe_resolve_hls_asset(hls_id, asset_path)
ctype = _hls_content_type(path)
if request.method == "HEAD":
size = path.stat().st_size
headers = {
"Content-Type": ctype,
"Content-Length": str(size),
"Cache-Control": "no-store",
}
return Response(status_code=200, headers=headers)
# For playlists, inject token into segment URIs so the browser can fetch them without headers.
if path.suffix.lower() == ".m3u8":
token = _extract_token(request) or ""
text = path.read_text("utf-8", errors="replace")
if token:
text = _rewrite_m3u8_with_token(text, token)
data = text.encode("utf-8")
return Response(
content=data,
status_code=200,
headers={"Content-Type": ctype, "Content-Length": str(len(data)), "Cache-Control": "no-store"},
)
size = path.stat().st_size
headers = {"Content-Type": ctype, "Content-Length": str(size), "Cache-Control": "no-store"}
return StreamingResponse(
_iter_file(path, 0, size - 1),
status_code=200,
headers=headers,
media_type=ctype,
)
@app.api_route("/media/{file_path:path}", methods=["GET", "HEAD"])
async def serve_media(file_path: str, request: Request) -> Response:
_require_auth(request)
path = _safe_resolve_media_path(file_path)
size = path.stat().st_size
ctype = _guess_content_type(path)
# Support byte ranges for HTML5 video seeking.
range_header = request.headers.get("range", "")
try:
parsed = _parse_range(range_header, size) if range_header else None
except HTTPException as exc:
# Required by RFC: include */size
headers = {"Content-Range": f"bytes */{size}", "Accept-Ranges": "bytes"}
return Response(status_code=exc.status_code, content=exc.detail, headers=headers)
common_headers = {
"Accept-Ranges": "bytes",
"Content-Type": ctype,
# Helps some players / proxies behave better with range requests.
"Cache-Control": "no-store",
}
if parsed is None:
if request.method == "HEAD":
return Response(status_code=200, headers={**common_headers, "Content-Length": str(size)})
return StreamingResponse(
_iter_file(path, 0, size - 1),
status_code=200,
headers={**common_headers, "Content-Length": str(size)},
media_type=ctype,
)
start, end = parsed
length = end - start + 1
headers = {
**common_headers,
"Content-Length": str(length),
"Content-Range": f"bytes {start}-{end}/{size}",
}
if request.method == "HEAD":
return Response(status_code=206, headers=headers)
return StreamingResponse(
_iter_file(path, start, end),
status_code=206,
headers=headers,
media_type=ctype,
)
if __name__ == "__main__":
import uvicorn
host = os.getenv("VIDEO_SERVER_HOST", "0.0.0.0")
port = int(os.getenv("VIDEO_SERVER_PORT", "8001"))
reload = os.getenv("VIDEO_SERVER_RELOAD", "1").strip() not in {"0", "false", "False", "no", "NO"}
workers = int(os.getenv("VIDEO_SERVER_WORKERS", "1").strip() or "1")
# Uvicorn ignores/limits workers when reload=True.
if reload:
workers = 1
uvicorn.run("video_server:app", host=host, port=port, reload=reload, workers=workers)