Store programme index in DB and add interleaved XMLTV file support

apps/epg/migrations/0022_epgsource_programme_index.py

@@ -0,0 +1,21 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('epg', '0021_epgsource_priority'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='epgsource',
+            name='programme_index',
+            field=models.JSONField(
+                blank=True,
+                default=None,
+                help_text='Byte-offset index mapping tvg_id to file positions, built after each EPG refresh',
+                null=True,
+            ),
+        ),
+    ]

apps/epg/models.py

@@ -59,6 +59,12 @@ class EPGSource(models.Model):
         blank=True,
         help_text="Last status message, including success results or error information"
     )
+    programme_index = models.JSONField(
+        null=True,
+        blank=True,
+        default=None,
+        help_text="Byte-offset index mapping tvg_id to file positions, built after each EPG refresh"
+    )
     created_at = models.DateTimeField(
         auto_now_add=True,
         help_text="Time when this source was created"

apps/epg/tasks.py

@@ -298,11 +298,7 @@ def refresh_epg_data(source_id):
         if source.source_type == 'xmltv':
             # Invalidate the byte-offset index before downloading the new file
             # so stale offsets are never used during the refresh window.
-            from core.utils import RedisClient
-            try:
-                RedisClient.get_client().delete(f'epg_programme_index_{source.id}')
-            except Exception:
-                pass
+            EPGSource.objects.filter(id=source.id).update(programme_index=None)
 
             fetch_success = fetch_xmltv(source)
             if not fetch_success:
@@ -2369,6 +2365,7 @@ _PROGRAMME_TAG = b'<programme'
 _PROGRAMME_TAG_LEN = len(_PROGRAMME_TAG)
 _TAG_FOLLOW = b' \t\n\r>/'
 _MAX_START_TAG = 4096  # generous upper bound for a start tag with namespaces/extra attrs
+_OFFSET_CAP = 10  # max block-starts recorded per channel; exceeding this flags the file as interleaved
 
 
 def _find_programme_tag(buf, start):
@@ -2417,12 +2414,10 @@ def _programme_to_dict(elem, start_time, end_time):
 def build_programme_index(source_id):
     """
     Scan the XML file with raw binary I/O to build a {tvg_id: [byte_offset, ...]} map.
-    Stores the result in Redis as JSON. Most XMLTV files group programmes by channel,
-    but some split a channel across multiple non-contiguous blocks, so we record the
-    start offset of every block.
+    Persists the result to EPGSource.programme_index. Most XMLTV files group programmes
+    by channel, but some split a channel across multiple non-contiguous blocks, so we
+    record the start offset of every block.
     """
-    from core.utils import RedisClient
-
     try:
         source = EPGSource.objects.get(id=source_id)
     except EPGSource.DoesNotExist:
@@ -2442,6 +2437,7 @@ def build_programme_index(source_id):
     start = time.monotonic()
     index = {}
     prev_channel = None
+    interleaved = False
 
     CHUNK = 8 * 1024 * 1024  # 8MB
 
@@ -2474,7 +2470,10 @@ def build_programme_index(source_id):
                     if channel_id not in index:
                         index[channel_id] = [abs_pos]
                     elif channel_id != prev_channel:
-                        index[channel_id].append(abs_pos)
+                        if len(index[channel_id]) < _OFFSET_CAP:
+                            index[channel_id].append(abs_pos)
+                        else:
+                            interleaved = True
                     prev_channel = channel_id
 
                 search_from = (
@@ -2494,11 +2493,11 @@ def build_programme_index(source_id):
     elapsed = time.monotonic() - start
     logger.info(
         f'[build_programme_index] Indexed {len(index)} channels in {elapsed:.1f}s for source {source_id}'
+        + (' (interleaved)' if interleaved else '')
     )
 
-    redis_client = RedisClient.get_client()
-    redis_key = f'epg_programme_index_{source_id}'
-    redis_client.set(redis_key, json.dumps(index))
+    result = {'channels': index, 'interleaved': interleaved}
+    EPGSource.objects.filter(id=source_id).update(programme_index=result)
 
 
 @shared_task
@@ -2525,7 +2524,7 @@ def find_current_program_for_tvg_id(epg_or_id):
             return None
 
     source = epg.epg_source
-    if not source or source.source_type == 'dummy':
+    if not source or source.source_type in ('dummy', 'schedules_direct'):
         return None
 
     tvg_id = epg.tvg_id
@@ -2536,29 +2535,40 @@ def find_current_program_for_tvg_id(epg_or_id):
     if not file_path or not os.path.exists(file_path):
         return None
 
-    redis_client = RedisClient.get_client()
-    redis_key = f'epg_programme_index_{source.id}'
-    index_raw = redis_client.get(redis_key)
-
     now = timezone.now()
+    # Force a fresh read of the DB-backed index to avoid using stale related-object
+    # state when an EPG refresh invalidates/rebuilds the index concurrently.
+    source.refresh_from_db(fields=['programme_index'])
+    index = source.programme_index
 
-    if index_raw:
-        # Indexed lookup
-        try:
-            index = json.loads(index_raw)
-        except (json.JSONDecodeError, TypeError):
-            index = None
-
-        if index and tvg_id in index:
-            offsets = index[tvg_id]
-            return _read_programs_at_offsets(file_path, tvg_id, offsets, now)
-        # tvg_id not in index - channel has no programmes in the file
-        return None
+    if index is not None:
+        channels = index.get('channels', {})
+        if tvg_id not in channels:
+            # Channel has no programmes in the file
+            return None
+        offsets = channels[tvg_id]
+        if index.get('interleaved'):
+            # Check all stored offsets first (cheap: one seek + one element parse each)
+            result = _read_programs_at_offsets(file_path, tvg_id, offsets, now)
+            if result is not None:
+                return result
+            # Current programme is beyond the stored offsets; scan forward from the
+            # last known position to avoid re-reading the already-checked portion
+            result = _scan_from_offset_for_tvg_id(file_path, tvg_id, offsets[-1], now)
+            if result == 'timeout':
+                logger.warning(
+                    f'[find_current_program_for_tvg_id] Interleaved scan timed out for '
+                    f'tvg_id={tvg_id} source={source.id}; index has {len(offsets)} offsets'
+                )
+                return None
+            return result
+        return _read_programs_at_offsets(file_path, tvg_id, offsets, now)
 
     # No index yet, fall back to sequential scan
     result = _sequential_scan_for_tvg_id(file_path, tvg_id, now, timeout_sec=10)
     if result == 'timeout':
         # Trigger background index build (Redis lock prevents duplicates)
+        redis_client = RedisClient.get_client()
         lock_key = f'building_programme_index_{source.id}'
         if redis_client.set(lock_key, '1', nx=True, ex=300):
             build_programme_index_task.delay(source.id)
@@ -2653,6 +2663,97 @@ def _read_programs_at_offsets(file_path, tvg_id, offsets, now):
     return None
 
 
+def _scan_from_offset_for_tvg_id(file_path, tvg_id, start_offset, now, timeout_sec=10):
+    """
+    Scan forward from start_offset for tvg_id, skipping other channels rather than
+    stopping at a channel boundary. Used for interleaved/time-sorted XMLTV files where
+    the index holds only the first known offset per channel.
+    Returns dict, None, or 'timeout'.
+    """
+    PROG_CLOSE = b'</programme>'
+    CLOSE_LEN = len(PROG_CLOSE)
+    READ_SIZE = 2 * 1024 * 1024
+    deadline = time.monotonic() + timeout_sec
+
+    with open(file_path, 'rb') as f:
+        f.seek(start_offset)
+        buf = bytearray()
+
+        while True:
+            if time.monotonic() > deadline:
+                return 'timeout'
+
+            chunk = f.read(READ_SIZE)
+            if not chunk and not buf:
+                break
+            buf.extend(chunk)
+            search_from = 0
+
+            trim_to = 0
+
+            while True:
+                tag_start, tag_end = _find_programme_tag(buf, search_from)
+                if tag_start == -1:
+                    trim_to = search_from
+                    break
+                if tag_end == -1 and chunk:
+                    trim_to = tag_start  # keep incomplete tag for next read
+                    break
+
+                m = _CHANNEL_ATTR_RE.search(
+                    buf,
+                    tag_start,
+                    tag_end + 1 if tag_end != -1 else tag_start + _MAX_START_TAG,
+                )
+                if not m:
+                    search_from = (
+                        tag_end + 1 if tag_end != -1 else tag_start + _PROGRAMME_TAG_LEN
+                    )
+                    continue
+
+                ch = (m.group(1) or m.group(2)).decode('utf-8', errors='replace')
+                if ch != tvg_id:
+                    search_from = (
+                        tag_end + 1 if tag_end != -1 else tag_start + _PROGRAMME_TAG_LEN
+                    )
+                    continue
+
+                close_pos = buf.find(
+                    PROG_CLOSE, tag_end + 1 if tag_end != -1 else m.end()
+                )
+                if close_pos == -1:
+                    trim_to = tag_start  # keep incomplete element for next read
+                    break
+                close_end = close_pos + CLOSE_LEN
+
+                element_bytes = bytes(buf[tag_start:close_end])
+                search_from = close_end
+
+                try:
+                    prog = etree.fromstring(element_bytes)
+                except etree.XMLSyntaxError:
+                    continue
+
+                start_str = prog.get('start')
+                stop_str = prog.get('stop')
+                if not start_str or not stop_str:
+                    continue
+                start_time = parse_xmltv_time(start_str)
+                end_time = parse_xmltv_time(stop_str)
+                if start_time is None or end_time is None:
+                    continue
+                if start_time <= now < end_time:
+                    return _programme_to_dict(prog, start_time, end_time)
+
+            if trim_to > 0:
+                del buf[:trim_to]
+
+            if not chunk:
+                break
+
+    return None
+
+
 def _sequential_scan_for_tvg_id(file_path, tvg_id, now, timeout_sec=10):
     """Stream through the file with iterparse for *tvg_id*. Returns dict, None, or "timeout"."""
     deadline = time.monotonic() + timeout_sec

core/tasks.py

@@ -388,7 +388,7 @@ def scan_and_process_files():
             }
         )
 
-    # Rebuild EPG programme indices on first run (Redis is ephemeral)
+    # Rebuild EPG programme indices on first run if missing (e.g. after migration or container restart)
     if not _first_scan_completed:
         _rebuild_programme_indices()
 
@@ -396,20 +396,18 @@ def scan_and_process_files():
     _first_scan_completed = True
 
 def _rebuild_programme_indices():
-    """Queue index builds for active EPG sources that are missing their Redis index."""
+    """Queue index builds for active EPG sources that are missing their DB index."""
     try:
         from apps.epg.tasks import build_programme_index_task
         redis_client = RedisClient.get_client()
 
         sources = EPGSource.objects.filter(
             is_active=True,
-        ).exclude(source_type='dummy')
+            programme_index__isnull=True,
+        ).exclude(source_type__in=('dummy', 'schedules_direct'))
 
         count = 0
         for source in sources:
-            redis_key = f"epg_programme_index_{source.id}"
-            if redis_client.exists(redis_key):
-                continue
             lock_key = f"building_programme_index_{source.id}"
             if redis_client.set(lock_key, "1", nx=True, ex=300):
                 build_programme_index_task.delay(source.id)