From 8d8c318478182b2241897b5f29eea987a8fe6aa3 Mon Sep 17 00:00:00 2001 From: Lyncoln Date: Sun, 1 Feb 2026 11:38:53 -0500 Subject: [PATCH] added m3u to HTTP server and extracted additional information from api --- Dockerfile | 4 +- tvj.m3u | 3 + tvj_epg.py | 261 +++++++++++++++++++++++++++++++++++++++++++---------- 3 files changed, 217 insertions(+), 51 deletions(-) create mode 100644 tvj.m3u diff --git a/Dockerfile b/Dockerfile index 02aebc7..41b2275 100644 --- a/Dockerfile +++ b/Dockerfile @@ -5,6 +5,9 @@ WORKDIR /app # Copy the EPG script COPY tvj_epg.py . +# Copy static M3U playlist +COPY tvj.m3u /app/output/tvj.m3u + # Install requests for API fetching RUN pip install --no-cache-dir requests @@ -26,4 +29,3 @@ while true; do \ done & \ # Start HTTP server to serve XML file on port 8787 \ cd /app/output && python3 -m http.server 8787' - diff --git a/tvj.m3u b/tvj.m3u new file mode 100644 index 0000000..5849f3a --- /dev/null +++ b/tvj.m3u @@ -0,0 +1,3 @@ +#EXTM3U +#EXTINF:-1 tvg-id="TVJ.jm@SD" tvg-logo="https://i.imgur.com/R4PoC3L.png" group-title="General",Television Jamaica (TVJ) +https://vod2live.univtec.com/manifest/a99a1804-dc83-411f-8c1c-b62f08cdfa59.m3u8 diff --git a/tvj_epg.py b/tvj_epg.py index 853099f..527e0a9 100644 --- a/tvj_epg.py +++ b/tvj_epg.py @@ -1,67 +1,228 @@ +#!/usr/bin/env python3 + +import os import requests import xml.etree.ElementTree as ET from datetime import datetime, timezone +from zoneinfo import ZoneInfo -# 1SpotMedia API config -API_URL = "https://www.1spotmedia.com/index.php/api/epg/get_epg_timeline_by_id" +# ---------------- CONFIG ---------------- + +EPG_URL = "https://www.1spotmedia.com/index.php/api/epg/get_epg_timeline_by_id" CHANNEL_ID = "66e312890478fd00235244cb" -TVG_ID = "TVJ.jm@SD" -# Output file +TVG_ID = "TVJ.jm@SD" +CHANNEL_NAME = "Television Jamaica" +CHANNEL_ICON = "https://www.televisionjamaica.com/Portals/0/tvj_logo.png" + +# Podman/container output path (mount this directory) OUTPUT_FILE = "/app/output/tvj.xml" -# Record script start time -script_run_time = datetime.now(tz=timezone.utc) -print(f"[{script_run_time.strftime('%Y-%m-%d %H:%M:%S UTC')}] Script run started.") +# ---------------------------------------- -# Fetch EPG data -resp = requests.get(API_URL, params={"id": CHANNEL_ID}, headers={"User-Agent": "Mozilla/5.0"}) -data = resp.json() +# Resolve timezone for LOGGING ONLY +TZ_NAME = os.environ.get("TZ", "UTC") +try: + LOG_TZ = ZoneInfo(TZ_NAME) +except Exception: + print(f"Invalid TZ '{TZ_NAME}', falling back to UTC") + LOG_TZ = timezone.utc -# Determine last update time from API if available -if isinstance(data, dict) and "lastUpdate" in data: - # API may provide a timestamp in milliseconds - api_last_update_ts = data["lastUpdate"] - last_update_time = datetime.fromtimestamp(api_last_update_ts / 1000, tz=timezone.utc) -elif isinstance(data, list) and len(data) > 0: - # Fallback: use the latest program endTime - last_update_time = datetime.fromtimestamp(max(p["endTime"] for p in data) / 1000, tz=timezone.utc) -else: - last_update_time = script_run_time -# Ensure we have a list of programs -if isinstance(data, dict): - programs = data.get("data", []) -elif isinstance(data, list): - programs = data -else: - programs = [] +def fmt_log_time(dt_utc: datetime) -> str: + """Format a UTC datetime into the configured log timezone.""" + return dt_utc.astimezone(LOG_TZ).strftime("%Y-%m-%d %H:%M:%S %Z") -# Create XMLTV root -tv = ET.Element("tv", attrib={"generator-info-name": "custom-1spotmedia"}) -# Add channel -channel = ET.SubElement(tv, "channel", id=TVG_ID) -ET.SubElement(channel, "display-name").text = "TVJ SD" +def epoch_ms_to_xmltv(ts_ms: int) -> str: + """Convert epoch milliseconds to XMLTV timestamp (UTC).""" + dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc) + return dt.strftime("%Y%m%d%H%M%S +0000") -# Add programs -for p in programs: - start = datetime.fromtimestamp(p["startTime"] / 1000, tz=timezone.utc) - stop = datetime.fromtimestamp(p["endTime"] / 1000, tz=timezone.utc) - programme = ET.SubElement(tv, "programme", - start=start.strftime("%Y%m%d%H%M%S +0000"), - stop=stop.strftime("%Y%m%d%H%M%S +0000"), - channel=TVG_ID) - ET.SubElement(programme, "title").text = p["title"] - if "description" in p: - ET.SubElement(programme, "desc").text = p["description"] +def clean_text(s): + if s is None: + return None + s = str(s).strip() + return s if s else None -# Write XMLTV file -tree = ET.ElementTree(tv) -tree.write(OUTPUT_FILE, encoding="utf-8", xml_declaration=True) -# Log completion with timestamps -completion_time = datetime.now(tz=timezone.utc) -print(f"[{completion_time.strftime('%Y-%m-%d %H:%M:%S UTC')}] XMLTV guide updated successfully: {OUTPUT_FILE}") -print(f"Summary: Script run time = {script_run_time.strftime('%Y-%m-%d %H:%M:%S UTC')}, Last guide update from API = {last_update_time.strftime('%Y-%m-%d %H:%M:%S UTC')}") +def get_program_obj(item: dict) -> dict: + prog = item.get("program") + return prog if isinstance(prog, dict) else {} + + +def get_description(item: dict) -> str: + """ + Extract description from nested program object. + Treat 'Dummy description.' as missing. + """ + prog = get_program_obj(item) + + desc = ( + prog.get("longDescription") + or prog.get("description") + or prog.get("shortDescription") + ) + desc = clean_text(desc) + + if desc and desc.lower() == "dummy description.": + desc = None + + return desc or "No description available." + + +def get_category(item: dict): + """Use displayGenre if present.""" + return clean_text(get_program_obj(item).get("displayGenre")) + + +def add_episode_numbers(programme_el, item: dict): + """ + Add episode number in xmltv_ns format if possible. + xmltv_ns: season is 0-based, episode is 0-based. + """ + prog = get_program_obj(item) + + season = prog.get("tvSeasonNumber") + episode = prog.get("tvSeasonEpisodeNumber") or prog.get("seriesEpisodeNumber") + + try: + season_i = int(season) if season is not None else None + except (ValueError, TypeError): + season_i = None + + try: + episode_i = int(episode) if episode is not None else None + except (ValueError, TypeError): + episode_i = None + + if season_i is None and episode_i is None: + return + + # xmltv_ns is 0-based; assume provider values are 1-based if >= 1 + s0 = (season_i - 1) if (season_i is not None and season_i >= 1) else 0 + e0 = (episode_i - 1) if (episode_i is not None and episode_i >= 1) else 0 + + ep = ET.SubElement(programme_el, "episode-num", system="xmltv_ns") + if season_i is not None and episode_i is not None: + ep.text = f"{s0}.{e0}." + elif season_i is not None: + ep.text = f"{s0}.." + else: + ep.text = f".{e0}." + + +def add_date(programme_el, item: dict): + """ + Add YYYY if we can find a year. + Try: program.year, else pubDate epoch-ms -> year. + """ + prog = get_program_obj(item) + + year = prog.get("year") + year_out = None + + try: + if year is not None: + y = int(year) + if 1800 <= y <= 2100: + year_out = str(y) + except (ValueError, TypeError): + year_out = None + + if year_out is None: + pub = prog.get("pubDate") + try: + if pub is not None: + dt = datetime.fromtimestamp(int(pub) / 1000, tz=timezone.utc) + year_out = str(dt.year) + except (ValueError, TypeError, OSError): + year_out = None + + if year_out: + ET.SubElement(programme_el, "date").text = year_out + + +def add_length(programme_el, item: dict): + """ + Add N from program.runtime if present. + """ + prog = get_program_obj(item) + runtime = prog.get("runtime") + if runtime is None: + return + + try: + seconds = int(str(runtime).strip()) + if seconds > 0: + ET.SubElement(programme_el, "length", units="seconds").text = str(seconds) + except (ValueError, TypeError): + return + + +def main(): + run_time_utc = datetime.now(timezone.utc) + print(f"[{fmt_log_time(run_time_utc)}] Script run started (TZ={TZ_NAME})") + + resp = requests.get(EPG_URL, params={"id": CHANNEL_ID}, timeout=30) + resp.raise_for_status() + data = resp.json() + + if not isinstance(data, list) or not data: + raise RuntimeError("EPG response is empty or invalid (expected a non-empty list).") + + # EPG "updated until" = furthest programme end time in returned schedule + last_update_epoch = max(int(p["endTime"]) for p in data if "endTime" in p) + last_update_utc = datetime.fromtimestamp(last_update_epoch / 1000, tz=timezone.utc) + + tv = ET.Element("tv", attrib={"generator-info-name": "1SpotMedia TVJ EPG (enriched)"}) + + # Channel block + ch = ET.SubElement(tv, "channel", id=TVG_ID) + ET.SubElement(ch, "display-name").text = CHANNEL_NAME + ET.SubElement(ch, "icon", src=CHANNEL_ICON) + + # Programmes + for item in data: + if "startTime" not in item or "endTime" not in item: + continue + + programme = ET.SubElement( + tv, + "programme", + start=epoch_ms_to_xmltv(int(item["startTime"])), + stop=epoch_ms_to_xmltv(int(item["endTime"])), + channel=TVG_ID, + ) + + title = ET.SubElement(programme, "title", lang="en") + title.text = ( + clean_text(item.get("title")) + or clean_text(get_program_obj(item).get("title")) + or "Unknown Program" + ) + + ET.SubElement(programme, "desc", lang="en").text = get_description(item) + + cat = get_category(item) + if cat: + ET.SubElement(programme, "category", lang="en").text = cat + + add_episode_numbers(programme, item) + add_date(programme, item) + add_length(programme, item) + + # Write XMLTV file (overwrites) + ET.ElementTree(tv).write(OUTPUT_FILE, encoding="utf-8", xml_declaration=True) + + done_utc = datetime.now(timezone.utc) + print(f"[{fmt_log_time(done_utc)}] XMLTV written: {OUTPUT_FILE}") + print( + f"Summary: Last run={fmt_log_time(run_time_utc)}, " + f"EPG updated until={fmt_log_time(last_update_utc)}" + ) + + +if __name__ == "__main__": + main()