Programme guide for 12 stations, rebuilt every 6 hours

BBC (RMS API), LBC News (epgshare01), Antena 1-3 (RTP JSON), RNE, Radio 3,
Radio 5, Radio Clásica (RTVE guide pages) and WBUR (weekly schedule).
IPTV Simple now reloads the list and guide hourly.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

resources/lib/epg.py

@@ -0,0 +1,294 @@
+# -*- coding: utf-8 -*-
+"""Builds an XMLTV programme guide for the stations that publish a schedule.
+
+Plain Python (no Kodi modules), so it can be tested outside Kodi:
+    python3 -I resources/lib/epg.py /tmp/epg.xml
+
+Each source returns programmes as dicts with UTC epoch 'start'/'stop',
+'title' and optional 'subtitle', 'desc', 'category' and 'icon'. Times are
+parsed by hand: datetime.strptime fails at random inside Kodi's Python.
+"""
+import calendar
+import gzip
+import html
+import io
+import json
+import re
+import time
+import urllib.request
+import xml.etree.ElementTree as ET
+
+UA = {'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36'}
+DAY = 86400
+
+# tvg-id in radio.m3u -> (source, source's own id)
+CHANNELS = {
+    'bbc-radio-4': ('bbc', 'bbc_radio_fourfm'),
+    'bbc-world-service': ('bbc', 'bbc_world_service'),
+    'bbc-radio-5-live': ('bbc', 'bbc_radio_five_live'),
+    'lbc-news': ('epgshare_uk', 'LBC.News.uk'),
+    'antena-1': ('rtp', '1'),
+    'antena-2': ('rtp', '2'),
+    'antena-3': ('rtp', '3'),
+    'rne-radio-nacional-galicia': ('rtve', 'rne'),
+    'rne-radio-3': ('rtve', 'radio3'),
+    'rne-radio-5-galicia': ('rtve', 'radio5'),
+    'rne-radio-clasica': ('rtve', 'radioclasica'),
+    'wbur-boston': ('wbur', None),
+}
+
+
+def get(url, timeout=30):
+    data = urllib.request.urlopen(urllib.request.Request(url, headers=UA), timeout=timeout).read()
+    return data
+
+
+# --- time zones -------------------------------------------------------------
+
+def _last_sunday(year, month):
+    last = calendar.monthrange(year, month)[1]
+    return last - (calendar.weekday(year, month, last) + 1) % 7
+
+
+def _nth_sunday(year, month, n):
+    first = (6 - calendar.weekday(year, month, 1)) % 7 + 1
+    return first + 7 * (n - 1)
+
+
+def utc_offset(zone, ts):
+    """Hours east of UTC at UTC epoch ts for 'madrid', 'lisbon' or 'boston'."""
+    year = time.gmtime(ts).tm_year
+    if zone in ('madrid', 'lisbon'):  # EU summer time: last Sunday of March to last Sunday of October, 01:00 UTC
+        start = calendar.timegm((year, 3, _last_sunday(year, 3), 1, 0, 0))
+        end = calendar.timegm((year, 10, _last_sunday(year, 10), 1, 0, 0))
+        base = 1 if zone == 'madrid' else 0
+        return base + (1 if start <= ts < end else 0)
+    # US Eastern: second Sunday of March 02:00 EST to first Sunday of November 02:00 EDT
+    start = calendar.timegm((year, 3, _nth_sunday(year, 3, 2), 7, 0, 0))
+    end = calendar.timegm((year, 11, _nth_sunday(year, 11, 1), 6, 0, 0))
+    return -4 if start <= ts < end else -5
+
+
+def local_to_utc(zone, y, mo, d, h, mi, s=0):
+    naive = calendar.timegm((y, mo, d, h, mi, s))
+    guess = naive - utc_offset(zone, naive) * 3600
+    return naive - utc_offset(zone, guess) * 3600
+
+
+def local_date(zone, ts):
+    """(year, month, day, weekday) of UTC epoch ts in the zone."""
+    t = time.gmtime(ts + utc_offset(zone, ts) * 3600)
+    return t.tm_year, t.tm_mon, t.tm_mday, t.tm_wday
+
+
+def iso_utc(text):
+    """'2026-11-11T23:32:30Z' -> epoch."""
+    n = [int(x) for x in re.findall(r'\d+', text)[:6]]
+    return calendar.timegm(tuple(n))
+
+
+def chain_stops(programmes, last_length=3600):
+    """Sources that give only start times: each programme ends when the next starts."""
+    programmes.sort(key=lambda p: p['start'])
+    for p, nxt in zip(programmes, programmes[1:] + [None]):
+        p.setdefault('stop', nxt['start'] if nxt else p['start'] + last_length)
+    return [p for p in programmes if p['stop'] > p['start']]
+
+
+def text(fragment):
+    return html.unescape(re.sub(r'<[^>]+>', '', fragment or '')).strip()
+
+
+# --- sources ----------------------------------------------------------------
+
+def bbc(service, now):
+    out = []
+    for day in (now, now + DAY):
+        url = 'https://rms.api.bbc.co.uk/v2/experience/inline/schedules/{}/{}'.format(
+            service, time.strftime('%Y-%m-%d', time.gmtime(day)))
+        for module in json.loads(get(url))['data']:
+            for b in module.get('data') or []:
+                if b.get('type') != 'broadcast_summary':
+                    continue
+                titles, synopses = b.get('titles') or {}, b.get('synopses') or {}
+                out.append({'start': iso_utc(b['start']), 'stop': iso_utc(b['end']),
+                            'title': titles.get('primary') or '',
+                            'subtitle': titles.get('secondary') or '',
+                            'desc': synopses.get('medium') or synopses.get('short') or '',
+                            'icon': (b.get('image_url') or '').replace('{recipe}', '480x270') or None})
+    return out
+
+
+def rtp(channel, now):
+    out = []
+    for day in (now, now + DAY):
+        y, m, d, _ = local_date('lisbon', day)
+        url = 'https://www.rtp.pt/EPG/json/rtp-channels-page/list-grid/radio/{}/{}-{}-{}/lis'.format(channel, d, m, y)
+        result = json.loads(get(url))['result']
+        for part in ('morning', 'afternoon', 'evening'):
+            for item in result.get(part) or []:
+                n = [int(x) for x in re.findall(r'\d+', item['date'])[:6]]
+                images = item.get('image') or []
+                episode = item.get('episode') or {}
+                out.append({'start': local_to_utc('lisbon', *n),
+                            'title': text(item.get('name')),
+                            'subtitle': text(episode.get('title')),
+                            'desc': text(episode.get('sinopse')) or text(item.get('description')),
+                            'icon': images[-1]['src'] if images else None})
+    return chain_stops(out)
+
+
+RTVE_ITEM = re.compile(
+    r'data-inicio="(\d\d):(\d\d)".*?<span class="datfin">\s*(?:<!-- -->)?(\d\d):(\d\d)</span>'
+    r'.*?<span class="pretitle">([^<]*)</span>\s*<h2 class="maintitle">([^<]*)</h2>(.*?)</div>'
+    r'.*?<img [^>]*src="([^"]+)"', re.S)
+
+
+def rtve(slug, now):
+    out = []
+    for day, suffix in ((now, ''), (now + DAY, 'manana/')):
+        y, m, d, _ = local_date('madrid', day)
+        page = get('https://www.rtve.es/play/guia-rne/{}/{}'.format(slug, suffix)).decode('utf-8', 'replace')
+        midnight = local_to_utc('madrid', y, m, d, 0, 0)
+        previous = -1
+        for h1, m1, h2, m2, category, title, rest, image in RTVE_ITEM.findall(page):
+            start = local_to_utc('madrid', y, m, d, int(h1), int(m1))
+            if start < previous:  # the page went past midnight
+                start += DAY
+            stop = local_to_utc('madrid', y, m, d, int(h2), int(m2))
+            while stop <= start:
+                stop += DAY
+            previous = start
+            desc = re.search(r'<p>(.*?)</p>', rest, re.S)
+            out.append({'start': start, 'stop': stop, 'title': text(title), 'category': text(category),
+                        'desc': text(desc.group(1)) if desc else '', 'icon': html.unescape(image)})
+        if not out or out[-1]['stop'] < midnight:
+            raise ValueError('no programmes on ' + slug)
+    return out
+
+
+def wbur(_, now):
+    body = json.loads(get('https://api.wbur.org/schedule'))['body']
+    out = []
+    for day in (now - DAY, now, now + DAY, now + 2 * DAY):
+        y, m, d, wday = local_date('boston', day)
+        rows = body['saturday' if wday == 5 else 'sunday' if wday == 6 else 'weekday']
+        for row in rows:
+            hh, mm, ampm = re.match(r'(\d+):(\d+)\s*([AP]M)', row['time']).groups()
+            hour = int(hh) % 12 + (12 if ampm == 'PM' else 0)
+            labels = [text(i.get('label')) for i in row.get('items') or [] if i.get('label')]
+            if labels:
+                out.append({'start': local_to_utc('boston', y, m, d, hour, int(mm)), 'title': ' / '.join(labels)})
+    return chain_stops(out)
+
+
+def epgshare_uk(channel_ids, now):
+    """All wanted channels from one download of epgshare01's UK guide."""
+    wanted = set(channel_ids)
+    data = gzip.GzipFile(fileobj=io.BytesIO(get('https://epgshare01.online/epgshare01/epg_ripper_UK1.xml.gz', 120)))
+    out = dict((c, []) for c in wanted)
+    for _, el in ET.iterparse(data):
+        if el.tag == 'programme':
+            channel = el.get('channel')
+            if channel in wanted:
+                out[channel].append({'start': xmltv_time(el.get('start')), 'stop': xmltv_time(el.get('stop')),
+                                     'title': el.findtext('title') or '', 'subtitle': el.findtext('sub-title') or '',
+                                     'desc': el.findtext('desc') or ''})
+            el.clear()
+        elif el.tag == 'channel':
+            el.clear()
+    return out
+
+
+def xmltv_time(value):
+    n = re.match(r'(\d{4})(\d\d)(\d\d)(\d\d)(\d\d)(\d\d)\s*([+-])(\d\d)(\d\d)', value).groups()
+    ts = calendar.timegm(tuple(int(x) for x in n[:6]))
+    offset = (int(n[7]) * 3600 + int(n[8]) * 60) * (1 if n[6] == '+' else -1)
+    return ts - offset
+
+
+SOURCES = {'bbc': bbc, 'rtp': rtp, 'rtve': rtve, 'wbur': wbur}
+
+
+# --- guide ------------------------------------------------------------------
+
+def fetch_all(log=print, now=None):
+    """{tvg-id: [programmes]} for every channel whose source answered."""
+    now = now or time.time()
+    guide = {}
+    for cid, (source, sid) in CHANNELS.items():
+        if source not in SOURCES:
+            continue
+        try:
+            guide[cid] = SOURCES[source](sid, now)
+        except Exception as e:  # one broken source mustn't stop the others
+            log('EPG {} ({}) failed: {}'.format(cid, source, e))
+    shared = dict((sid, cid) for cid, (source, sid) in CHANNELS.items() if source == 'epgshare_uk')
+    if shared:
+        try:
+            for sid, programmes in epgshare_uk(list(shared), now).items():
+                if programmes:
+                    guide[shared[sid]] = programmes
+        except Exception as e:
+            log('EPG epgshare01 UK failed: {}'.format(e))
+    return guide
+
+
+def xmltv(guide, names):
+    def stamp(ts):
+        return time.strftime('%Y%m%d%H%M%S +0000', time.gmtime(ts))
+
+    def esc(s):
+        return html.escape(s or '', quote=True)
+
+    lines = ['<?xml version="1.0" encoding="UTF-8"?>', '<tv generator-info-name="service.radios.vigo">']
+    for cid in sorted(guide):
+        lines.append('  <channel id="{}"><display-name>{}</display-name></channel>'.format(esc(cid), esc(names.get(cid, cid))))
+    for cid in sorted(guide):
+        for p in sorted(guide[cid], key=lambda p: p['start']):
+            lines.append('  <programme start="{}" stop="{}" channel="{}">'.format(stamp(p['start']), stamp(p['stop']), esc(cid)))
+            lines.append('    <title>{}</title>'.format(esc(p['title'])))
+            if p.get('subtitle'):
+                lines.append('    <sub-title>{}</sub-title>'.format(esc(p['subtitle'])))
+            if p.get('desc'):
+                lines.append('    <desc>{}</desc>'.format(esc(p['desc'])))
+            if p.get('category'):
+                lines.append('    <category>{}</category>'.format(esc(p['category'])))
+            if p.get('icon'):
+                lines.append('    <icon src="{}"/>'.format(esc(p['icon'])))
+            lines.append('  </programme>')
+    lines.append('</tv>')
+    return '\n'.join(lines) + '\n'
+
+
+def merge(old, new, now):
+    """New programmes, plus the old ones of channels whose source failed this
+    time, as long as they still reach into the future."""
+    guide = dict(new)
+    for cid, programmes in (old or {}).items():
+        if cid not in guide and any(p['stop'] > now for p in programmes):
+            guide[cid] = programmes
+    for cid in guide:  # keep a few hours of the past, drop the rest
+        guide[cid] = [p for p in guide[cid] if p['stop'] > now - 6 * 3600]
+    return guide
+
+
+def channel_names(m3u):
+    return dict(re.findall(r'tvg-id="([^"]+)".*,(.*)', m3u))
+
+
+if __name__ == '__main__':
+    import os
+    import sys
+    root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+    with io.open(os.path.join(root, 'radio.m3u'), encoding='utf-8') as f:
+        names = channel_names(f.read())
+    g = fetch_all()
+    for cid in sorted(CHANNELS):
+        ps = g.get(cid, [])
+        first = time.strftime('%d %H:%M', time.gmtime(ps[0]['start'])) if ps else '-'
+        last = time.strftime('%d %H:%M', time.gmtime(max(p['stop'] for p in ps))) if ps else '-'
+        print('{:<28} {:>4} programmes  {} -> {} UTC'.format(cid, len(ps), first, last))
+    if len(sys.argv) > 1:
+        with io.open(sys.argv[1], 'w', encoding='utf-8') as f:
+            f.write(xmltv(g, names))