"""XMLTV EPG generation.""" import logging import time import xml.etree.ElementTree as ET from datetime import datetime, timezone, timedelta from pathlib import Path from .config import EPG_TIMESPAN_DAYS, MOVIES_ROOT, SERIES_ROOT from .tmdb_client import parse_season_episode logger = logging.getLogger(__name__) def format_xmltv_time(dt: datetime) -> str: """Format as XMLTV: YYYYMMDDHHmmss +0000""" return dt.strftime("%Y%m%d%H%M%S ") + (dt.strftime("%z") if dt.tzinfo else "+0000") def get_programme_name_from_path(path: Path) -> str: """ Return the series/movie name from the folder (e.g. series/W817/... -> W817, movies/MyMovie/... -> MyMovie). """ try: if SERIES_ROOT in path.parents: rel = path.relative_to(SERIES_ROOT) parts = rel.parts if len(parts) > 1: return parts[0] return path.stem if MOVIES_ROOT in path.parents: rel = path.relative_to(MOVIES_ROOT) parts = rel.parts if len(parts) > 1: return parts[0] return path.stem except ValueError: pass return path.stem def schedule_epoch_utc() -> datetime: """Midnight UTC today — same reference used by EPG and live stream.""" return datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0) class EPGBuilder: """Builds XMLTV EPG for all channels.""" def __init__(self, schedule_builder, tmdb_client, channel_repo, cache_manager=None): self._schedule = schedule_builder self._tmdb = tmdb_client self._channels = channel_repo self._cache = cache_manager def build_xml(self) -> str: """ Build XMLTV EPG for all channels. Schedule repeats from midnight UTC. Programmes generated for EPG_TIMESPAN_DAYS. """ if self._cache: cached = self._cache.get_epg() if cached is not None: logger.debug("EPG served from cache") return cached epg_start = time.monotonic() channels = self._channels.get_all() if not channels: return '\n' logger.info("EPG build started (%s channels)", len(channels)) now = datetime.now(timezone.utc) epoch = schedule_epoch_utc() end_epoch = epoch + timedelta(days=EPG_TIMESPAN_DAYS) root = ET.Element("tv") root.set("source-info-name", "self-hosted-iptv") root.set("generator-info-name", "self-hosted-iptv") for ch in channels: cid = ch.get("id", ch.get("name", "")) name = ch.get("name", cid) chan_el = ET.SubElement(root, "channel", id=cid) ET.SubElement(chan_el, "display-name").text = name total_programmes = 0 for ch in channels: cid = ch.get("id", ch.get("name", "")) ch_start = time.monotonic() schedule = self._schedule.get_or_build(cid) if not schedule: logger.debug("EPG channel %s: no schedule", cid) continue paths, durations, path_configs = schedule cycle_duration_sec = sum(durations) if cycle_duration_sec <= 0: continue cycle_delta = timedelta(seconds=cycle_duration_sec) offsets_sec = [0.0] for d in durations[:-1]: offsets_sec.append(offsets_sec[-1] + d) channel_programmes = 0 cycle_start = epoch while cycle_start < end_epoch: for i, (path, dur) in enumerate(zip(paths, durations)): start_dt = cycle_start + timedelta(seconds=offsets_sec[i]) stop_dt = start_dt + timedelta(seconds=dur) if stop_dt <= now: continue if start_dt >= end_epoch: break channel_programmes += 1 prog = ET.SubElement( root, "programme", start=format_xmltv_time(start_dt), stop=format_xmltv_time(stop_dt), channel=cid, ) cfg = path_configs[i] if i < len(path_configs) else {} programme_name = get_programme_name_from_path(path) meta = self._tmdb.get_episode_metadata( programme_name, path, display_name=cfg.get("display_name"), tmdb_search=cfg.get("tmdb_search"), tmdb_id=cfg.get("tmdb_id"), series_root=SERIES_ROOT, ) lang = (meta.get("lang") or "en") if meta else "en" if meta: ET.SubElement(prog, "title", lang=lang).text = meta["title"] ET.SubElement(prog, "sub-title", lang=lang).text = meta["sub_title"] if meta.get("desc"): ET.SubElement(prog, "desc", lang=lang).text = meta["desc"] if meta.get("icon"): ET.SubElement(prog, "icon", attrib={"src": meta["icon"]}) directors = meta.get("directors") or [] actors = meta.get("actors") or [] producers = meta.get("producers") or [] if directors or actors or producers: credits_el = ET.SubElement(prog, "credits") for d in directors: ET.SubElement(credits_el, "director").text = d for a in actors: ET.SubElement(credits_el, "actor").text = a for p in producers: ET.SubElement(credits_el, "producer").text = p if meta.get("date"): ET.SubElement(prog, "date").text = meta["date"] for cat in meta.get("categories") or []: ET.SubElement(prog, "category", lang=lang).text = cat if meta.get("country"): ET.SubElement(prog, "country", lang=lang).text = meta["country"] nums = meta.get("episode_nums") or {} if nums.get("xmltv_ns"): ET.SubElement( prog, "episode-num", system="xmltv_ns" ).text = nums["xmltv_ns"] if nums.get("onscreen"): ET.SubElement( prog, "episode-num", system="onscreen" ).text = nums["onscreen"] else: fallback_title = ( (cfg.get("display_name") or programme_name).strip() or programme_name ) ET.SubElement(prog, "title", lang="en").text = fallback_title ET.SubElement(prog, "sub-title", lang="en").text = path.stem se = parse_season_episode(path.stem) if se: s, e = se ET.SubElement( prog, "episode-num", system="xmltv_ns" ).text = f"{s}.{e}.0/1" ET.SubElement( prog, "episode-num", system="onscreen" ).text = f"S{s:02d}E{e:02d}" cycle_start += cycle_delta ch_elapsed = time.monotonic() - ch_start total_programmes += channel_programmes logger.info("EPG channel %s: %s programmes in %.1fs", cid, channel_programmes, ch_elapsed) ET.indent(root, space=" ") epg_elapsed = time.monotonic() - epg_start logger.info("EPG build finished in %.1fs (%s programmes total)", epg_elapsed, total_programmes) xml = ( '\n' + ET.tostring(root, encoding="unicode", default_namespace=None) ) if self._cache: self._cache.set_epg(xml) return xml