Merge pull request add-ons#30 from dagwieers/iptv-epg-fix

Fix IPTV EPG processing

.github/workflows/status.yml

@@ -26,6 +26,6 @@ jobs:
  run: |
  python -m pip install --upgrade pip
  pip install -r requirements.txt
-# - name: TEST IPTV Manager EPG
-# run: python -m unittest -v test_iptvmanager.TestIPTVManager.test_get_iptv_epg
-# if: always()
+ - name: TEST IPTV Manager EPG
+ run: python -m unittest -v test_iptvmanager.TestIPTVManager.test_get_iptv_epg
+ if: always()

resources/lib/iptvmanager.py

@@ -4,7 +4,6 @@
 
 from __future__ import absolute_import, division, unicode_literals
 
-from xbmc import log
 
 class IPTVManager:
  """ Interface to IPTV Manager """
@@ -32,51 +31,11 @@ def send(self):
  @via_socket
  def send_channels(self): # pylint: disable=no-method-argument,no-self-use
  """ Return JSON-STREAMS formatted information to IPTV Manager. """
- from kodiutils import addon_icon
-
- streams = []
-
- streams.append(dict(
- name="Red Bull TV",
- stream="plugin:https://plugin.video.redbull.tv/iptv/play",
- id="redbulltv",
- logo=addon_icon(),
- preset=88,
- ))
-
- return dict(version=1, streams=streams)
+ from redbull import RedBullTV
+ return dict(version=1, streams=RedBullTV().get_iptv_channels())
 
  @via_socket
  def send_epg(self): # pylint: disable=no-method-argument,no-self-use
  """ Return JSON-EPG formatted information to IPTV Manager. """
- import re
- from collections import defaultdict
- from datetime import datetime
- from dateutil.tz import UTC
- from kodiutils import url_for
  from redbull import RedBullTV
-
- redbull = RedBullTV()
-
- epg = defaultdict(list)
-
- for item in redbull.get_epg().get('items'):
- if (
- item.get('start_time')
- and item.get('end_time')
- and re.match('\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}Z', item.get('start_time')) # pylint: disable=anomalous-backslash-in-string
- and re.match('\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}Z', item.get('end_time'))): # pylint: disable=anomalous-backslash-in-string
- epg['redbulltv'].append(dict(
- start=datetime.strptime(item.get('start_time'), "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=UTC).isoformat(),
- stop=datetime.strptime(item.get('end_time'), "%Y-%m-%dT%H:%M:%S.%fZ").replace(tzinfo=UTC).isoformat(),
- title=item.get('title'),
- description=item.get('long_description'),
- subtitle=item.get('subheading'),
- genre='Sport',
- image=redbull.get_image_url(item.get('id'), item.get('resources'), 'landscape'),
- stream=url_for('play_uid', uid=item.get('id'))
- ))
- else:
- log("Invalid start or end time for Red Bull item ID {uid}".format(uid=item.get('id')))
-
- return dict(version=1, epg=epg)
+ return dict(version=1, epg=RedBullTV().get_iptv_epg())

resources/lib/redbull.py

@@ -4,7 +4,7 @@
 
 from __future__ import absolute_import, division, unicode_literals
 import xbmc
-from kodiutils import to_unicode
+from kodiutils import addon_icon, to_unicode, url_for
 
 
 class RedBullTV():
@@ -48,9 +48,51 @@ def get_json(self, url, use_token=False):
  # NOTE: With Python 3.5 and older json.loads() does not support bytes or bytearray, so we convert to unicode
  return loads(to_unicode(response.read()))
 
+ @staticmethod
+ def get_iptv_channels():
+ # We only have a single live stream for Red Bul TV
+ return [
+ dict(
+ name='Red Bull TV',
+ stream='plugin:https://plugin.video.redbull.tv/iptv/play',
+ id='redbulltv',
+ logo=addon_icon(),
+ preset=88,
+ ),
+ ]
+
  def get_epg(self):
  return self.get_json(self.REDBULL_API + 'epg?complete=true', use_token=True)
 
+ def get_iptv_epg(self):
+ from collections import defaultdict
+ import re
+
+ epg = defaultdict(list)
+ regexp = re.compile(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}.\d{3}(Z|\+00:00)')
+
+ # Process the individual EPG items
+ for item in self.get_epg().get('items'):
+ if not regexp.match(item.get('start_time', '')):
+ xbmc.log("Invalid start_time '{start_time}' for Red Bull item ID '{id}'".format(**item))
+ continue
+
+ if not regexp.match(item.get('end_time', '')):
+ xbmc.log("Invalid start_time '{end_time}' for Red Bull item ID '{id}'".format(**item))
+ continue
+
+ epg['redbulltv'].append(dict(
+ start=item.get('start_time'),
+ stop=item.get('end_time'),
+ title=item.get('title'),
+ description=item.get('long_description'),
+ subtitle=item.get('subheading'),
+ genre='Sport',
+ image=self.get_image_url(item.get('id'), item.get('resources'), 'landscape'),
+ stream=url_for('play_uid', uid=item.get('id'))
+ ))
+ return epg
+
  def get_image_url(self, element_id, resources, element_type, width=1024, quality=70):
  if element_type == 'fanart':
  if 'rbtv_background_landscape' in resources:

tests/test_iptvmanager.py

@@ -0,0 +1,44 @@
+# -*- coding: utf-8 -*-
+# Copyright: (c) 2021, Dag Wieers (@dagwieers) <[email protected]>
+# GNU General Public License v3.0 (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+"""Integration tests for IPTV Manager functionality"""
+
+# pylint: disable=invalid-name,line-too-long
+
+from __future__ import absolute_import, division, print_function, unicode_literals
+from datetime import datetime, timedelta
+import unittest
+import dateutil.tz
+import addon
+
+xbmc = __import__('xbmc')
+xbmcaddon = __import__('xbmcaddon')
+xbmcgui = __import__('xbmcgui')
+xbmcplugin = __import__('xbmcplugin')
+xbmcvfs = __import__('xbmcvfs')
+
+plugin = addon.plugin
+now = datetime.now(dateutil.tz.tzlocal())
+lastweek = now + timedelta(days=-7)
+
+
+class TestIPTVManager(unittest.TestCase):
+ """TestCase class"""
+
+ def test_get_iptv_channels(self):
+ """Get IPTV channels"""
+ from redbull import RedBullTV
+ iptv_channels = RedBullTV().get_iptv_channels()
+ # print(iptv_channels)
+ self.assertTrue(iptv_channels)
+
+ def test_get_iptv_epg(self):
+ """Get IPTV EPG"""
+ from redbull import RedBullTV
+ iptv_epg = RedBullTV().get_iptv_epg()
+ # print(iptv_epg)
+ self.assertTrue(iptv_epg)
+
+
+if __name__ == '__main__':
+ unittest.main()