diff --git a/Containerfile b/Containerfile index cdf2243..e4acbce 100644 --- a/Containerfile +++ b/Containerfile @@ -7,6 +7,7 @@ RUN pip install --no-cache-dir requests ENV UPDATE_INTERVAL=6 PORT=1876 EXPOSE 1876 +HEALTHCHECK --interval=30s --timeout=10s --start-period=60s --retries=3 CMD ["python", "-m", "epg.healthcheck"] COPY entrypoint.sh /entrypoint.sh RUN chmod +x /entrypoint.sh ENTRYPOINT ["/entrypoint.sh"] diff --git a/README.md b/README.md index 7f2b922..034d938 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # TVJ IPTV + XMLTV Container A configurable EPG engine that fetches channel schedules and serves a combined -XMLTV guide and M3U playlist over HTTP. TVJ and TVJ Sports Network share a Univtec +XMLTV guide and M3U playlist over HTTP. TVJ, TVJ Sports Network, and Jamaica News Network share a Univtec provider plugin; the engine fetches their common feed once per update. ## Run @@ -19,6 +19,9 @@ podman run -d --name tvj-epg -p 1876:1876 \ With no `CHANNELS` setting, only TVJ is included, preserving the default behavior. Use `CHANNELS=tvjsn` for TVJ Sports Network alone. +Use `CHANNELS=tvj,tvjsn,jnn` to include Jamaica News Network as well, or +`CHANNELS=jnn` for JNN alone. JNN uses the same Univtec EPG feed and the supplied +stream URL `https://rjrjnn-ioriver-cdn.encoders.immergo.tv/2/streamPlaylist.m3u8`. - Playlist: `http://localhost:1876/tvj.m3u` - Guide: `http://localhost:1876/tvj.xml` @@ -49,6 +52,7 @@ through its HTTP server. | --- | --- | --- | | `tvj` | Television Jamaica | `TVJ.jm@SD` | | `tvjsn` | TVJ Sports Network | `TVJSN.jm@SD` | +| `jnn` | Jamaica News Network | `JNN.jm@SD` | TVJ Sports Network uses the supplied video URL: `https://vod2live.univtec.com/manifest/65cf631b-d422-49e0-ab25-07d5aa08ca8f/7660000.m3u8`. @@ -74,8 +78,8 @@ A failed fetch or invalid schedule leaves previous files intact, and the contain retries at the next interval. Each file is replaced atomically; the two files are not a single transaction. Old data can expire during prolonged outages. -Custom feed/channel environment settings, per-channel cached fallback, and health -reporting are planned for subsequent milestones; they are not implemented yet. +Custom providers are out of scope for now. Per-channel cached fallback is not +implemented; a failed channel update preserves the previous combined outputs. The root `tvj.m3u` is a legacy static example; container playlists are now generated. ## Local verification @@ -86,3 +90,38 @@ With Python 3.11+ and `requests` installed: python3 -m unittest discover -s tests -v CHANNELS=tvj,tvjsn OUTPUT_FILE=output/tvj.xml python3 -m epg.engine ``` + +## Health reporting + +`http://localhost:1876/health` returns JSON describing the latest scheduled +provider fetch, per-channel parsing results, current guide coverage, programme +counts, guide expiry, and XMLTV/M3U output validity. It uses the configured `PORT`. + +| Status | HTTP code | Meaning | +| --- | --- | --- | +| `healthy` | 200 | Latest update succeeded, every selected channel covers now and has data extending into the future, and both outputs are usable | +| `degraded` | 200 | Outputs remain usable, but the latest update failed, no update record is available, or an update is overdue | +| `unhealthy` | 503 | At least one selected channel lacks current/future guide data, or a required output is missing or invalid | + +An update is overdue after `UPDATE_INTERVAL` hours plus a five-minute grace period. +Startup without usable output reports unhealthy. Guide freshness is recalculated +on every health request, including during upstream outages. A current programme +whose end is in the future satisfies the future-data check; no minimum forecast +window is required. Schedule gaps report unhealthy while they are occurring. + +The latest update result is stored atomically in `tvj.health.json` beside the +XMLTV output (the filename follows `OUTPUT_FILE`). This file is a stored update +record; `/health` computes live freshness and output checks. Provider status means +the last request and top-level response parsing succeeded; per-channel parsing +results are reported separately. Errors are identified by exception type. + +The container checks `/health` every 30 seconds with a 60-second startup grace +period and three retries. Degraded status passes the container probe because the +existing outputs are still usable. These checks make no additional requests to +Univtec and do not check video-stream playback. Container health status alone does +not configure automatic restart behavior. + +```sh +curl http://localhost:1876/health +podman inspect --format '{{.State.Health.Status}}' tvj-epg +``` diff --git a/entrypoint.sh b/entrypoint.sh index 4148895..bfad9d0 100644 --- a/entrypoint.sh +++ b/entrypoint.sh @@ -21,5 +21,4 @@ python -c 'from epg.config import load; load()' done ) & -cd /app/output -exec python3 -m http.server "$PORT" +exec python3 -m epg.server diff --git a/epg/config.py b/epg/config.py index ad2b4c2..4e0e60a 100644 --- a/epg/config.py +++ b/epg/config.py @@ -17,6 +17,13 @@ BUILTINS = { icon="https://rjr.vod.immergo.tv/rjr/uploaded/85c62d19-8e57-49b5-820a-acca5adcc1c9.png", group="Sports", ), + "jnn": Channel( + id="JNN.jm@SD", name="Jamaica News Network", + channel_id="66e0a79714ec780025bae4fd", + stream_url="https://rjrjnn-ioriver-cdn.encoders.immergo.tv/2/streamPlaylist.m3u8", + icon="https://rjr.vod.immergo.tv/rjr/uploaded/3c3487a8-5e77-4c3a-abb0-8a5eca477e86.png", + group="News", + ), } diff --git a/epg/engine.py b/epg/engine.py index 0790a41..f185a8c 100644 --- a/epg/engine.py +++ b/epg/engine.py @@ -1,7 +1,8 @@ """Fetch each provider feed once and publish selected channels.""" -from datetime import datetime +from datetime import datetime, timezone import sys from epg.config import load +from epg.health import record from epg.providers import PROVIDERS from epg.writers import atomic_write, m3u, xmltv @@ -10,17 +11,44 @@ def run(environ=None): channels, xml_path, m3u_path = load(environ) feeds = {} schedules = {} + report = {'checked_at': datetime.now(timezone.utc).isoformat(), + 'success': False, 'providers': {}, 'channels': {}} + errors = [] for channel in channels: provider = PROVIDERS[channel.provider] key = (channel.provider, channel.feed_url, channel.tenant) if key not in feeds: - feeds[key] = provider.fetch(channel.feed_url, channel.tenant) - schedules[channel.id] = provider.programmes(feeds[key], channel.channel_id) + try: + feeds[key] = provider.fetch(channel.feed_url, channel.tenant) + report['providers'][channel.provider] = {'status': 'healthy'} + except Exception as error: + feeds[key] = error + report['providers'][channel.provider] = { + 'status': 'unhealthy', 'error': type(error).__name__} + try: + if isinstance(feeds[key], Exception): + raise feeds[key] + schedules[channel.id] = provider.programmes(feeds[key], channel.channel_id) + report['channels'][channel.id] = {'status': 'healthy'} + except Exception as error: + errors.append(error) + report['channels'][channel.id] = { + 'status': 'unhealthy', 'error': type(error).__name__} + if errors: + record(xml_path, report) + raise errors[0] # Fetch and render everything before replacing any previous output. - xml_content = xmltv(channels, schedules) - playlist = m3u(channels) - atomic_write(xml_path, xml_content) - atomic_write(m3u_path, playlist) + try: + xml_content = xmltv(channels, schedules) + playlist = m3u(channels) + atomic_write(xml_path, xml_content) + atomic_write(m3u_path, playlist) + report['success'] = True + except Exception as error: + report['output_error'] = type(error).__name__ + raise + finally: + record(xml_path, report) print(f"[{datetime.now().astimezone().isoformat()}] Wrote {xml_path} and {m3u_path}", flush=True) for channel in channels: programmes = schedules[channel.id] diff --git a/epg/health.py b/epg/health.py new file mode 100644 index 0000000..309e4f8 --- /dev/null +++ b/epg/health.py @@ -0,0 +1,85 @@ +"""Health from local update results and the files actually served to clients.""" +import json +import os +from datetime import datetime, timedelta, timezone +import xml.etree.ElementTree as ET +from epg.config import load +from epg.writers import atomic_write + + +def state_path(xml_path): + return xml_path.with_suffix('.health.json') + + +def record(xml_path, report): + atomic_write(state_path(xml_path), json.dumps(report).encode()) + + +def evaluate(environ=None, now=None): + channels, xml_path, m3u_path = load(environ) + now = now or datetime.now(timezone.utc) + try: + report = json.loads(state_path(xml_path).read_text()) + if not isinstance(report, dict): + raise ValueError('Invalid health state') + except (OSError, ValueError): + report = {} + outputs = {'xmltv': False, 'm3u': False} + root = None + playlist = '' + try: + root = ET.parse(xml_path).getroot() + outputs['xmltv'] = root.tag == 'tv' + except (OSError, ET.ParseError): + pass + try: + playlist = m3u_path.read_text() + outputs['m3u'] = playlist.startswith('#EXTM3U\n') + except (OSError, UnicodeError): + pass + details = {} + for channel in channels: + intervals = [] + declared = False + if outputs['xmltv']: + declared = any(c.get('id') == channel.id for c in root.findall('channel')) + for p in root.findall('programme'): + if p.get('channel') != channel.id: + continue + try: + start, stop = (datetime.strptime(p.get(k, ''), '%Y%m%d%H%M%S %z') for k in ('start', 'stop')) + if stop > start and p.findtext('title'): + intervals.append((start, stop)) + except ValueError: + pass + current = any(start <= now < stop + timedelta(seconds=1) for start, stop in intervals) + future = any(stop > now for _, stop in intervals) + lines = playlist.splitlines() + in_playlist = any( + line.startswith('#EXTINF:') and f'tvg-id="{channel.id}"' in line + and i + 1 < len(lines) and lines[i + 1] == channel.stream_url + for i, line in enumerate(lines) + ) + usable = declared and current and future and outputs['m3u'] and in_playlist + details[channel.id] = { + 'name': channel.name, 'status': 'healthy' if usable else 'unhealthy', + 'programmes': len(intervals), 'covers_now': current, 'has_future_data': future, + 'in_playlist': in_playlist, + 'guide_until': max((stop for _, stop in intervals), default=None).isoformat() if intervals else None, + 'last_update': report.get('channels', {}).get(channel.id), + } + env = os.environ if environ is None else environ + try: + last_attempt = datetime.fromisoformat(report['checked_at']) + overdue = now - last_attempt > timedelta(hours=int(env.get('UPDATE_INTERVAL', '6')), minutes=5) + except (KeyError, ValueError, TypeError): + overdue = True + usable = all(c['status'] == 'healthy' for c in details.values()) and all(outputs.values()) + status = 'unhealthy' if not usable else ('degraded' if overdue or not report.get('success') else 'healthy') + return { + 'status': status, 'checked_at': now.isoformat(), + 'last_attempt': report.get('checked_at'), 'update_overdue': overdue, + 'last_update_success': report.get('success', False), + 'providers': report.get('providers', {}), 'channels': details, 'outputs': outputs, + 'output_error': report.get('output_error'), + } diff --git a/epg/healthcheck.py b/epg/healthcheck.py new file mode 100644 index 0000000..8623bcb --- /dev/null +++ b/epg/healthcheck.py @@ -0,0 +1,16 @@ +"""Container probe: degraded service remains usable and passes the probe.""" +import os +import sys +import urllib.request + + +def main(): + try: + with urllib.request.urlopen(f'http://127.0.0.1:{os.environ.get("PORT", "1876")}/health', timeout=5) as response: + return 0 if response.status == 200 else 1 + except Exception: + return 1 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/epg/server.py b/epg/server.py new file mode 100644 index 0000000..81da0a0 --- /dev/null +++ b/epg/server.py @@ -0,0 +1,41 @@ +"""Serve generated files and a live, locally computed health endpoint.""" +import json +import os +from functools import partial +from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import urlsplit +from epg.health import evaluate + + +class Handler(SimpleHTTPRequestHandler): + def do_GET(self): + if urlsplit(self.path).path.rstrip('/') == '/health': + self.health() + else: + super().do_GET() + + def do_HEAD(self): + if urlsplit(self.path).path.rstrip('/') == '/health': + self.health(head=True) + else: + super().do_HEAD() + + def health(self, head=False): + try: + report = evaluate() + except Exception: + report = {'status': 'unhealthy', 'error': 'Health evaluation failed'} + body = json.dumps(report).encode() + self.send_response(503 if report['status'] == 'unhealthy' else 200) + self.send_header('Content-Type', 'application/json') + self.send_header('Cache-Control', 'no-store') + self.send_header('Content-Length', str(len(body))) + self.end_headers() + if not head: + self.wfile.write(body) + + +if __name__ == '__main__': + server = ThreadingHTTPServer(('', int(os.environ.get('PORT', '1876'))), + partial(Handler, directory='/app/output')) + server.serve_forever() diff --git a/tests/test_engine.py b/tests/test_engine.py index bf5a9bf..7936a1b 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -29,7 +29,7 @@ class EngineTests(unittest.TestCase): load({'CHANNELS': selection}) def test_outputs_and_shared_fetch(self): - for selection in ('tvj', 'tvjsn', 'tvj,tvjsn'): + for selection in ('tvj', 'tvjsn', 'jnn', 'tvj,tvjsn', 'tvj,tvjsn,jnn'): with self.subTest(selection=selection), tempfile.TemporaryDirectory() as directory: output = Path(directory) / 'tvj.xml' with patch.object(univtec, 'fetch', return_value=feed()) as fetch: diff --git a/tests/test_health.py b/tests/test_health.py new file mode 100644 index 0000000..3b7caec --- /dev/null +++ b/tests/test_health.py @@ -0,0 +1,82 @@ +import json +import tempfile +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch + +from epg.config import BUILTINS +from epg.engine import run +from epg.health import evaluate, record, state_path +from epg.providers import univtec + +NOW = datetime(2026, 9, 10, 12, 30, tzinfo=timezone.utc) + + +def feed(): + return [{'id': c.channel_id, 'epgObject': {'events': [ + {'title': 'Sport', 'start': '2026-09-10T12:00:00Z', 'end': '2026-09-10T14:00:00Z'} + ]}} for c in BUILTINS.values()] + + +class HealthTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.xml = Path(self.directory.name) / 'tvj.xml' + self.env = {'CHANNELS': 'tvj,tvjsn', 'OUTPUT_FILE': str(self.xml)} + + def update(self): + with patch.object(univtec, 'fetch', return_value=feed()): + run(self.env) + report = json.loads(state_path(self.xml).read_text()) + report['checked_at'] = NOW.isoformat() + record(self.xml, report) + + def test_startup_and_success(self): + self.assertEqual(evaluate(self.env, NOW)['status'], 'unhealthy') + self.update() + health = evaluate(self.env, NOW) + self.assertEqual(health['status'], 'healthy') + self.assertTrue(all(health['outputs'].values())) + self.assertEqual(len(health['channels']), 2) + + def test_failure_retains_usable_guide(self): + self.update() + before = self.xml.read_bytes() + with patch.object(univtec, 'fetch', side_effect=TimeoutError): + with self.assertRaises(TimeoutError): + run(self.env) + self.assertEqual(self.xml.read_bytes(), before) + health = evaluate(self.env, NOW) + self.assertEqual(health['status'], 'degraded') + self.assertEqual(health['providers']['univtec']['status'], 'unhealthy') + self.assertEqual(evaluate(self.env, NOW + timedelta(hours=2))['status'], 'unhealthy') + + def test_missing_channel_and_corrupt_outputs(self): + self.update() + with patch.object(univtec, 'fetch', return_value=feed()[:1]): + with self.assertRaises(ValueError): + run(self.env) + health = evaluate(self.env, NOW) + self.assertEqual(health['status'], 'degraded') + self.assertEqual(health['channels']['TVJSN.jm@SD']['last_update']['status'], 'unhealthy') + self.xml.with_suffix('.m3u').write_text('#EXTM3U\n') + self.assertEqual(evaluate(self.env, NOW)['status'], 'unhealthy') + self.xml.write_text('broken xml') + self.assertFalse(evaluate(self.env, NOW)['outputs']['xmltv']) + + def test_future_only_and_overdue(self): + self.update() + self.assertEqual(evaluate(self.env, NOW - timedelta(hours=1))['status'], 'unhealthy') + report = json.loads(state_path(self.xml).read_text()) + report['checked_at'] = (NOW - timedelta(hours=7)).isoformat() + record(self.xml, report) + self.assertEqual(evaluate(self.env, NOW)['status'], 'degraded') + + def test_health_requests_do_not_fetch(self): + self.update() + with patch.object(univtec, 'fetch') as fetch: + evaluate(self.env, NOW) + evaluate(self.env, NOW) + fetch.assert_not_called() diff --git a/tests/test_server.py b/tests/test_server.py new file mode 100644 index 0000000..7e588a7 --- /dev/null +++ b/tests/test_server.py @@ -0,0 +1,38 @@ +import json +import tempfile +import threading +import unittest +from functools import partial +from http.client import HTTPConnection +from http.server import ThreadingHTTPServer +from pathlib import Path +from unittest.mock import patch +from epg.server import Handler + + +class ServerTests(unittest.TestCase): + def test_health_codes_and_static_files(self): + with tempfile.TemporaryDirectory() as directory: + Path(directory, 'tvj.m3u').write_text('#EXTM3U\n') + server = ThreadingHTTPServer(('127.0.0.1', 0), partial(Handler, directory=directory)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + for status, expected in [('healthy', 200), ('degraded', 200), ('unhealthy', 503)]: + with patch('epg.server.evaluate', return_value={'status': status}): + connection = HTTPConnection(*server.server_address, timeout=5) + connection.request('GET', '/health') + response = connection.getresponse() + self.assertEqual(response.status, expected) + self.assertEqual(json.loads(response.read())['status'], status) + connection.close() + connection = HTTPConnection(*server.server_address, timeout=5) + connection.request('GET', '/tvj.m3u') + response = connection.getresponse() + self.assertEqual(response.status, 200) + self.assertEqual(response.read(), b'#EXTM3U\n') + connection.close() + finally: + server.shutdown() + server.server_close() + thread.join()