186 lines
6.5 KiB
Python
186 lines
6.5 KiB
Python
"""Schedule building: video collection, shuffle, durations, and caching."""
|
|
|
|
import logging
|
|
import random
|
|
import subprocess
|
|
import time
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
from .channel import normalize_path_config, resolve_path
|
|
from .config import (
|
|
VIDEO_EXTENSIONS,
|
|
DEFAULT_DURATION_SEC,
|
|
SCHEDULE_CACHE_TTL_SEC,
|
|
MOVIES_ROOT,
|
|
SERIES_ROOT,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def collect_videos(root: Path) -> list[Path]:
|
|
"""Collect all video files under root."""
|
|
if not root.exists() or not root.is_dir():
|
|
return []
|
|
videos = []
|
|
try:
|
|
for p in root.rglob("*"):
|
|
if p.is_file() and p.suffix.lower() in VIDEO_EXTENSIONS:
|
|
videos.append(p)
|
|
except OSError:
|
|
pass
|
|
return videos
|
|
|
|
|
|
def collect_channel_videos(path_configs: list[dict]) -> list[tuple[dict, Path]]:
|
|
"""For each path config, collect videos. Returns [(path_config, absolute_path)]."""
|
|
out = []
|
|
for cfg in path_configs:
|
|
path_key = cfg.get("path_key") or ""
|
|
root = resolve_path(path_key)
|
|
if root is None:
|
|
continue
|
|
for vid in collect_videos(root):
|
|
out.append((cfg, vid))
|
|
return out
|
|
|
|
|
|
def shuffle_no_adjacent_same_show(
|
|
items: list[tuple[dict, Path]],
|
|
) -> list[tuple[Path, dict]]:
|
|
"""Shuffle so same path_key never appears twice in a row. Returns [(path, path_config)]."""
|
|
if not items:
|
|
return []
|
|
by_key: dict[str, list[tuple[Path, dict]]] = defaultdict(list)
|
|
for cfg, path in items:
|
|
by_key[cfg.get("path_key") or ""].append((path, cfg))
|
|
for key in by_key:
|
|
random.shuffle(by_key[key])
|
|
groups = list(by_key.values())
|
|
result = []
|
|
n = max(len(g) for g in groups)
|
|
for i in range(n):
|
|
for g in groups:
|
|
if i < len(g):
|
|
result.append(g[i])
|
|
return result
|
|
|
|
|
|
def build_channel_path_list(path_entries: list) -> tuple[list[Path], list[dict]]:
|
|
"""Build randomized list of (path, path_config) for a channel."""
|
|
configs = [normalize_path_config(e) for e in path_entries]
|
|
configs = [c for c in configs if c.get("path_key")]
|
|
items = collect_channel_videos(configs)
|
|
pairs = shuffle_no_adjacent_same_show(items)
|
|
return [p for p, _ in pairs], [c for _, c in pairs]
|
|
|
|
|
|
def _path_to_stream_path(p: Path) -> str:
|
|
"""Convert absolute path to stream path (for stream?path=)."""
|
|
try:
|
|
if MOVIES_ROOT in p.parents:
|
|
rel = p.relative_to(MOVIES_ROOT)
|
|
return "movies/" + str(rel).replace("\\", "/")
|
|
rel = p.relative_to(SERIES_ROOT)
|
|
return "series/" + str(rel).replace("\\", "/")
|
|
except ValueError:
|
|
return "movies/" + p.name
|
|
|
|
|
|
def build_playlist(path_entries: list, base_url: str) -> list[tuple[str, str]]:
|
|
"""Build playlist as list of (display_name, stream_url)."""
|
|
data = build_playlist_data(path_entries)
|
|
return [(name, f"{base_url}/stream?path={path_str}") for name, path_str in data]
|
|
|
|
|
|
def build_playlist_data(path_entries: list) -> list[tuple[str, str]]:
|
|
"""Build playlist data as list of (display_name, path_str) for caching."""
|
|
paths, _ = build_channel_path_list(path_entries)
|
|
return [(p.stem, _path_to_stream_path(p)) for p in paths]
|
|
|
|
|
|
def get_duration(path: Path) -> float:
|
|
"""Return duration in seconds via ffprobe."""
|
|
t0 = time.monotonic()
|
|
try:
|
|
out = subprocess.run(
|
|
[
|
|
"ffprobe",
|
|
"-v", "error",
|
|
"-show_entries", "format=duration",
|
|
"-of", "default=noprint_wrappers=1:nokey=1",
|
|
str(path),
|
|
],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
if out.returncode == 0 and out.stdout.strip():
|
|
d = float(out.stdout.strip())
|
|
return d
|
|
except (subprocess.TimeoutExpired, ValueError, FileNotFoundError):
|
|
pass
|
|
return DEFAULT_DURATION_SEC
|
|
|
|
|
|
class ScheduleBuilder:
|
|
"""Builds and caches video schedules per channel."""
|
|
|
|
def __init__(self):
|
|
self._cache: dict[str, tuple[list[Path], list[float], list[dict], float]] = {}
|
|
self._ttl_sec = SCHEDULE_CACHE_TTL_SEC
|
|
self._channel_repo = None # injected
|
|
self._cache_manager = None # injected for disk persistence
|
|
|
|
def invalidate_cache(self) -> None:
|
|
"""Clear in-memory schedule cache (e.g. when channels.json changes)."""
|
|
self._cache.clear()
|
|
logger.debug("Schedule cache invalidated")
|
|
|
|
def set_channel_repo(self, repo) -> None:
|
|
"""Set the channel repository for schedule building."""
|
|
self._channel_repo = repo
|
|
|
|
def set_cache_manager(self, cache_manager) -> None:
|
|
"""Set the cache manager for disk persistence across restarts."""
|
|
self._cache_manager = cache_manager
|
|
|
|
def get_or_build(
|
|
self,
|
|
channel_id: str,
|
|
) -> tuple[list[Path], list[float], list[dict]] | None:
|
|
"""Return (paths, durations, path_configs) for the channel. Uses memory and disk cache."""
|
|
now = time.time()
|
|
if channel_id in self._cache:
|
|
paths, durations, path_configs, ts = self._cache[channel_id]
|
|
if now - ts < self._ttl_sec:
|
|
logger.debug("schedule cache hit for channel %s (%s videos)", channel_id, len(paths))
|
|
return paths, durations, path_configs
|
|
|
|
if self._cache_manager:
|
|
cached = self._cache_manager.get_schedule(channel_id)
|
|
if cached is not None:
|
|
paths, durations, path_configs = cached
|
|
self._cache[channel_id] = (paths, durations, path_configs, now)
|
|
logger.debug("schedule disk cache hit for channel %s (%s videos)", channel_id, len(paths))
|
|
return paths, durations, path_configs
|
|
|
|
logger.info("Building schedule for channel %s ...", channel_id)
|
|
t0 = time.monotonic()
|
|
if not self._channel_repo:
|
|
return None
|
|
channel = self._channel_repo.get_by_id(channel_id)
|
|
if not channel:
|
|
return None
|
|
paths, path_configs = build_channel_path_list(channel.get("paths", []))
|
|
if not paths:
|
|
return None
|
|
durations = [get_duration(p) for p in paths]
|
|
elapsed = time.monotonic() - t0
|
|
self._cache[channel_id] = (paths, durations, path_configs, now)
|
|
if self._cache_manager:
|
|
self._cache_manager.set_schedule(channel_id, paths, durations, path_configs)
|
|
logger.info("Schedule for channel %s built in %.1fs (%s videos)", channel_id, elapsed, len(paths))
|
|
return paths, durations, path_configs
|