projects/Dockers/self-hosted-iptv/iptv/schedule.py

"""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 build_playlist(path_entries: list, base_url: str) -> list[tuple[str, str]]:
    """Build playlist as list of (display_name, stream_url)."""
    paths, _ = build_channel_path_list(path_entries)
    playlist = []
    for p in paths:
        try:
            if MOVIES_ROOT in p.parents:
                rel = p.relative_to(MOVIES_ROOT)
                path_str = "movies/" + str(rel).replace("\\", "/")
            else:
                rel = p.relative_to(SERIES_ROOT)
                path_str = "series/" + str(rel).replace("\\", "/")
        except ValueError:
            path_str = "movies/" + p.name
        stream_url = f"{base_url}/stream?path={path_str}"
        playlist.append((p.stem, stream_url))
    return playlist


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

    def set_channel_repo(self, repo) -> None:
        """Set the channel repository for schedule building."""
        self._channel_repo = repo

    def get_or_build(
        self,
        channel_id: str,
    ) -> tuple[list[Path], list[float], list[dict]] | None:
        """Return (paths, durations, path_configs) for the channel. Cached per TTL."""
        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

        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)
        logger.info("Schedule for channel %s built in %.1fs (%s videos)", channel_id, elapsed, len(paths))
        return paths, durations, path_configs