This commit is contained in:
@@ -0,0 +1,187 @@
|
||||
"""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):
|
||||
self._schedule = schedule_builder
|
||||
self._tmdb = tmdb_client
|
||||
self._channels = channel_repo
|
||||
|
||||
def build_xml(self) -> str:
|
||||
"""
|
||||
Build XMLTV EPG for all channels. Schedule repeats from midnight UTC.
|
||||
Programmes generated for EPG_TIMESPAN_DAYS.
|
||||
"""
|
||||
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
|
||||
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)
|
||||
return (
|
||||
'<?xml version="1.0" encoding="UTF-8"?>\n'
|
||||
+ ET.tostring(root, encoding="unicode", default_namespace=None)
|
||||
)
|
||||
Reference in New Issue
Block a user