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()