209 lines
8.8 KiB
Python
209 lines
8.8 KiB
Python
"""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, prefetcher=None):
|
|
self._schedule = schedule_builder
|
|
self._tmdb = tmdb_client
|
|
self._channels = channel_repo
|
|
self._cache = cache_manager
|
|
self._prefetcher = prefetcher
|
|
|
|
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 '<?xml version="1.0" encoding="UTF-8"?>\n<tv></tv>'
|
|
|
|
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
|
|
scheduled: set[tuple[Path, str | None, str | None, int | None]] = set()
|
|
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
|
|
cfg = path_configs[i] if i < len(path_configs) else {}
|
|
if SERIES_ROOT in path.parents:
|
|
scheduled.add((
|
|
path,
|
|
cfg.get("display_name"),
|
|
cfg.get("tmdb_search"),
|
|
cfg.get("tmdb_id"),
|
|
))
|
|
prog = ET.SubElement(
|
|
root,
|
|
"programme",
|
|
start=format_xmltv_time(start_dt),
|
|
stop=format_xmltv_time(stop_dt),
|
|
channel=cid,
|
|
)
|
|
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 = (
|
|
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
|
+ ET.tostring(root, encoding="unicode", default_namespace=None)
|
|
)
|
|
if self._cache:
|
|
self._cache.set_epg(xml)
|
|
if self._prefetcher:
|
|
self._prefetcher.run_once(scheduled)
|
|
return xml
|