"""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()) logger.debug("ffprobe %s -> %.0fs in %.2fs", path.name, d, time.monotonic() - t0) return d except (subprocess.TimeoutExpired, ValueError, FileNotFoundError) as e: logger.debug("ffprobe %s failed: %s", path.name, e) 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