diff --git a/tvj_epg.py b/tvj_epg.py index 527e0a9..d1aa544 100644 --- a/tvj_epg.py +++ b/tvj_epg.py @@ -1,6 +1,8 @@ #!/usr/bin/env python3 import os +import re +import html import requests import xml.etree.ElementTree as ET from datetime import datetime, timezone @@ -8,15 +10,26 @@ from zoneinfo import ZoneInfo # ---------------- CONFIG ---------------- -EPG_URL = "https://www.1spotmedia.com/index.php/api/epg/get_epg_timeline_by_id" +EPG_URL = "https://insight-api-shared.univtec.com/interface/epg" + +# Stable TVJ object ID returned by the new Univtec API CHANNEL_ID = "66e312890478fd00235244cb" 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" +# Podman/container output path +OUTPUT_FILE = os.environ.get( + "OUTPUT_FILE", + "/app/output/tvj.xml", +) + +REQUEST_HEADERS = { + "accept": "application/json", + "platform": "web", + "x-tenant-id": "rjr", +} # ---------------------------------------- @@ -34,58 +47,104 @@ def fmt_log_time(dt_utc: datetime) -> str: return dt_utc.astimezone(LOG_TZ).strftime("%Y-%m-%d %H:%M:%S %Z") -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") +def parse_iso8601(value: str) -> datetime: + """Parse an ISO-8601 timestamp returned by Univtec.""" + if not value: + raise ValueError("Empty timestamp") + + # Python accepts +00:00 more consistently than trailing Z. + if value.endswith("Z"): + value = value[:-1] + "+00:00" + + dt = datetime.fromisoformat(value) + + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + + return dt.astimezone(timezone.utc) -def clean_text(s): - if s is None: +def datetime_to_xmltv(dt: datetime) -> str: + """Convert datetime to XMLTV UTC timestamp.""" + return dt.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S +0000") + + +def clean_text(value): + if value is None: return None - s = str(s).strip() - return s if s else None + + value = str(value).strip() + return value if value else None -def get_program_obj(item: dict) -> dict: - prog = item.get("program") - return prog if isinstance(prog, dict) else {} +def strip_html(value): + """Convert provider HTML descriptions to normal text.""" + value = clean_text(value) + if not value: + return None + + value = re.sub(r"", "\n", value, flags=re.IGNORECASE) + value = re.sub(r"", "\n", value, flags=re.IGNORECASE) + value = re.sub(r"<[^>]+>", "", value) + value = html.unescape(value) + + lines = [line.strip() for line in value.splitlines()] + value = " ".join(line for line in lines if line) + + return clean_text(value) + + +def get_custom(item: dict) -> dict: + custom = item.get("custom") + return custom if isinstance(custom, dict) else {} def get_description(item: dict) -> str: """ - Extract description from nested program object. - Treat 'Dummy description.' as missing. + Prefer the normal description. + Fall back to custom.longDescription with HTML removed. """ - prog = get_program_obj(item) + custom = get_custom(item) desc = ( - prog.get("longDescription") - or prog.get("description") - or prog.get("shortDescription") + clean_text(item.get("description")) + or strip_html(custom.get("longDescription")) ) - 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_categories(programme_el, item: dict): + """Add each provider genre as an XMLTV category.""" + custom = get_custom(item) + genres = custom.get("genre") + + if isinstance(genres, str): + genres = [genres] + + if not isinstance(genres, list): + return + + for genre in genres: + genre = clean_text(genre) + if genre: + ET.SubElement( + programme_el, + "category", + lang="en" + ).text = genre 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) + Add episode number using xmltv_ns when available. - season = prog.get("tvSeasonNumber") - episode = prog.get("tvSeasonEpisodeNumber") or prog.get("seriesEpisodeNumber") + xmltv_ns is zero-based. + """ + custom = get_custom(item) + + season = custom.get("seasonNumber") + episode = custom.get("episodeNumber") try: season_i = int(season) if season is not None else None @@ -100,11 +159,24 @@ def add_episode_numbers(programme_el, item: dict): 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 + 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" + ) - 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: @@ -115,114 +187,270 @@ def add_episode_numbers(programme_el, item: dict): def add_date(programme_el, item: dict): """ - Add YYYY if we can find a year. - Try: program.year, else pubDate epoch-ms -> year. + Add YYYY using custom.airDate when available. """ - prog = get_program_obj(item) + custom = get_custom(item) + air_date = custom.get("airDate") - year = prog.get("year") - year_out = None + if air_date is None: + return try: - if year is not None: - y = int(year) - if 1800 <= y <= 2100: - year_out = str(y) - except (ValueError, TypeError): - year_out = None + dt = datetime.fromtimestamp( + int(air_date) / 1000, + tz=timezone.utc, + ) - 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 + ET.SubElement(programme_el, "date").text = str(dt.year) - if year_out: - ET.SubElement(programme_el, "date").text = year_out + except (ValueError, TypeError, OSError): + pass def add_length(programme_el, item: dict): """ - Add N from program.runtime if present. + Add programme duration from custom.duration. + + Univtec duration appears to be seconds. """ - prog = get_program_obj(item) - runtime = prog.get("runtime") - if runtime is None: + custom = get_custom(item) + duration = custom.get("duration") + + if duration is None: return try: - seconds = int(str(runtime).strip()) + seconds = int(duration) + if seconds > 0: - ET.SubElement(programme_el, "length", units="seconds").text = str(seconds) + ET.SubElement( + programme_el, + "length", + units="seconds" + ).text = str(seconds) + except (ValueError, TypeError): + pass + + +def add_rating(programme_el, item: dict): + """Add provider content rating.""" + rating = clean_text(get_custom(item).get("rating")) + + if not rating: return + rating_el = ET.SubElement(programme_el, "rating") + ET.SubElement(rating_el, "value").text = rating + + +def add_icon(programme_el, item: dict): + """Add programme artwork if supplied by Univtec.""" + custom = get_custom(item) + + image = custom.get("image") + poster = custom.get("poster") + + image_url = None + + if isinstance(image, dict): + image_url = clean_text(image.get("downloadUrl")) + + if not image_url and isinstance(poster, dict): + image_url = clean_text(poster.get("downloadUrl")) + + if image_url: + ET.SubElement( + programme_el, + "icon", + src=image_url, + ) + + +def fetch_tvj_events(): + """ + Fetch the complete RJR EPG response and extract the TVJ object. + """ + resp = requests.get( + EPG_URL, + headers=REQUEST_HEADERS, + 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)." + ) + + channel = next( + ( + item + for item in data + if item.get("id") == CHANNEL_ID + ), + None, + ) + + if channel is None: + raise RuntimeError( + f"TVJ channel ID {CHANNEL_ID} " + "was not found in EPG response." + ) + + epg_object = channel.get("epgObject") + + if not isinstance(epg_object, dict): + raise RuntimeError( + "TVJ EPG response has no valid epgObject." + ) + + events = epg_object.get("events") + + if not isinstance(events, list): + raise RuntimeError( + "TVJ epgObject.events is missing or invalid." + ) + + return events + 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() + print( + f"[{fmt_log_time(run_time_utc)}] " + f"Script run started (TZ={TZ_NAME})" + ) - if not isinstance(data, list) or not data: - raise RuntimeError("EPG response is empty or invalid (expected a non-empty list).") + events = fetch_tvj_events() - # 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) + # Discard malformed entries and provider filler. + valid_events = [] - tv = ET.Element("tv", attrib={"generator-info-name": "1SpotMedia TVJ EPG (enriched)"}) + for item in events: + if not isinstance(item, dict): + continue + + if not item.get("start") or not item.get("end"): + continue + + title = clean_text(item.get("title")) + + if title and title.casefold() == "no program": + continue + + try: + start_dt = parse_iso8601(item["start"]) + end_dt = parse_iso8601(item["end"]) + except (ValueError, TypeError): + continue + + if end_dt <= start_dt: + continue + + valid_events.append((item, start_dt, end_dt)) + + if not valid_events: + raise RuntimeError( + "No valid TVJ programmes were returned." + ) + + # Sort defensively in case provider changes ordering. + valid_events.sort(key=lambda x: x[1]) + + last_update_utc = max( + end_dt + for _, _, end_dt in valid_events + ) + + tv = ET.Element( + "tv", + attrib={ + "generator-info-name": + "Univtec / 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) + + 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 + for item, start_dt, end_dt in valid_events: programme = ET.SubElement( tv, "programme", - start=epoch_ms_to_xmltv(int(item["startTime"])), - stop=epoch_ms_to_xmltv(int(item["endTime"])), + start=datetime_to_xmltv(start_dt), + stop=datetime_to_xmltv(end_dt), channel=TVG_ID, ) - title = ET.SubElement(programme, "title", lang="en") - title.text = ( + title = ( clean_text(item.get("title")) - or clean_text(get_program_obj(item).get("title")) + or clean_text(get_custom(item).get("seriesTitle")) or "Unknown Program" ) - ET.SubElement(programme, "desc", lang="en").text = get_description(item) + ET.SubElement( + programme, + "title", + lang="en", + ).text = title - cat = get_category(item) - if cat: - ET.SubElement(programme, "category", lang="en").text = cat + ET.SubElement( + programme, + "desc", + lang="en", + ).text = get_description(item) + add_categories(programme, item) add_episode_numbers(programme, item) add_date(programme, item) add_length(programme, item) + add_rating(programme, item) + add_icon(programme, item) - # Write XMLTV file (overwrites) - ET.ElementTree(tv).write(OUTPUT_FILE, encoding="utf-8", xml_declaration=True) + # Ensure output directory exists. + output_dir = os.path.dirname(OUTPUT_FILE) + + if output_dir: + os.makedirs(output_dir, exist_ok=True) + + # Write XMLTV file. + 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"[{fmt_log_time(done_utc)}] " + f"XMLTV written: {OUTPUT_FILE}" + ) + + print( + f"Summary: Programmes={len(valid_events)}, " + f"Last run={fmt_log_time(run_time_utc)}, " f"EPG updated until={fmt_log_time(last_update_utc)}" ) if __name__ == "__main__": - main() + main() \ No newline at end of file