fix(epg): handle attr-spacing and named entities in index lookup, unblock boot rebuild

apps/epg/tasks.py

@@ -52,6 +52,12 @@ def _build_html_entity_doctype() -> bytes:
 _HTML_ENTITY_DOCTYPE = _build_html_entity_doctype()
 
 
+def _parse_programme_element(element_bytes):
+    """Parse a single <programme> element, prepending the HTML-entity DOCTYPE so references like &eacute; in the text resolve instead of failing."""
+    parser = etree.XMLParser(resolve_entities=True, load_dtd=True, no_network=True)
+    return etree.fromstring(_HTML_ENTITY_DOCTYPE + element_bytes, parser)
+
+
 class _PrependStream:
     """Wraps an open binary file and prepends a bytes prefix to its content.
 
@@ -2360,7 +2366,7 @@ def _resolve_source_file(epg_source):
     return file_path
 
 
-_CHANNEL_ATTR_RE = re.compile(rb"""channel=(?:"([^"]+)"|'([^']+)')""")
+_CHANNEL_ATTR_RE = re.compile(rb"""channel\s*=\s*(?:"([^"]+)"|'([^']+)')""")
 _PROGRAMME_TAG = b'<programme'
 _PROGRAMME_TAG_LEN = len(_PROGRAMME_TAG)
 _TAG_FOLLOW = b' \t\n\r>/'
@@ -2654,7 +2660,7 @@ def _read_programs_at_offsets(file_path, tvg_id, offsets, now):
                     search_from = close_end
 
                     try:
-                        prog = etree.fromstring(element_bytes)
+                        prog = _parse_programme_element(element_bytes)
                     except etree.XMLSyntaxError:
                         continue
 
@@ -2747,7 +2753,7 @@ def _scan_from_offset_for_tvg_id(file_path, tvg_id, start_offset, now, timeout_s
                 search_from = close_end
 
                 try:
-                    prog = etree.fromstring(element_bytes)
+                    prog = _parse_programme_element(element_bytes)
                 except etree.XMLSyntaxError:
                     continue
 

apps/epg/tests/test_programme_index.py

@@ -202,6 +202,39 @@ class FindCurrentProgramTests(TestCase):
         finally:
             os.unlink(tmp_path)
 
+    def test_offset_lookup_resolves_named_html_entities_in_programme_text(self):
+        xml = (
+            '<?xml version="1.0" encoding="UTF-8"?>\n'
+            "<tv>\n"
+            '  <channel id="entity.channel"/>\n'
+            '  <programme start="20000101000000 +0000" '
+            'stop="20991231235959 +0000" channel="entity.channel">\n'
+            "    <title>Caf&eacute; Live</title>\n"
+            "  </programme>\n"
+            "</tv>\n"
+        )
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".xml", delete=False
+        ) as f:
+            f.write(xml)
+            tmp_path = f.name
+
+        try:
+            src = EPGSource.objects.create(
+                name="Named Entities", source_type="xmltv", file_path=tmp_path
+            )
+            build_programme_index(src.id)
+
+            epg = EPGData.objects.create(
+                tvg_id="entity.channel", name="Entity Channel", epg_source=src
+            )
+            result = find_current_program_for_tvg_id(epg)
+
+            self.assertIsNotNone(result)
+            self.assertEqual(result["title"], "Caf\u00e9 Live")
+        finally:
+            os.unlink(tmp_path)
+
     @patch("apps.epg.tasks.build_programme_index_task")
     def test_no_index_dispatches_build_and_returns_timeout(self, mock_build_task):
         # Source with no index and file on disk
@@ -243,6 +276,36 @@ class BuildProgrammeIndexTests(TestCase):
         # Small fixture has no interleaved channels
         self.assertEqual(index["interleaved_channels"], [])
 
+    def test_builds_index_when_channel_attribute_has_valid_xml_spacing(self):
+        xml = (
+            '<?xml version="1.0" encoding="UTF-8"?>\n'
+            "<tv>\n"
+            '  <channel id="spaced.channel"/>\n'
+            '  <programme start="20000101000000 +0000" '
+            'stop="20991231235959 +0000" channel = "spaced.channel">\n'
+            "    <title>Spaced Attribute Current</title>\n"
+            "  </programme>\n"
+            "</tv>\n"
+        )
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".xml", delete=False
+        ) as f:
+            f.write(xml)
+            tmp_path = f.name
+
+        try:
+            src = EPGSource.objects.create(
+                name="Spaced Attribute", source_type="xmltv", file_path=tmp_path
+            )
+            build_programme_index(src.id)
+            src.refresh_from_db()
+
+            self.assertIn(
+                "spaced.channel", src.programme_index["channels"]
+            )
+        finally:
+            os.unlink(tmp_path)
+
     def test_nonexistent_source_does_not_raise(self):
         # Should log error but not raise
         build_programme_index(99999)