feat(epg): add native xz decompression support for EPG sources

Adds .xz to the set of compressed formats detect_file_format(),
extract_compressed_file(), fetch_xmltv(), and EPGSource.get_cache_file()
recognize, mirroring the existing gzip handling. Uses stdlib lzma, no new
dependency. Detection works via LZMA magic bytes (fd 37 7a 58 5a 00),
the .xz extension (including compound extensions like .xml.xz), and the
application/x-xz mimetype fallback.

Widened the magic-byte read in get_cache_file() from 4 to 6 bytes since
the xz signature is 6 bytes (the previous 4-byte read was sufficient
only for the 2-byte gzip/zip signatures it originally supported). That
widening silently broke the bare-<tv> raw-XML detection: the
fixed-length slice comparison header[:5] == b'<tv>' can never match a
5-byte slice of a 6-byte header against a 4-byte literal. Replaced the
fixed-length slice comparisons with header.startswith(...), which is
correct regardless of how many bytes are read - this also fixes the
adjacent <?xml declaration check, which was silently dead before this
change (a 4-byte read could never equal the 5-byte b'<?xml' literal
either).

Fixes Dispatcharr/Dispatcharr#1414

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/epg/models.py

@@ -90,17 +90,21 @@ class EPGSource(models.Model):
                         file_ext = '.gz'
                     elif mime_type == 'application/zip':
                         file_ext = '.zip'
+                    elif mime_type == 'application/x-xz':
+                        file_ext = '.xz'
                     elif mime_type == 'application/xml' or mime_type == 'text/xml':
                         file_ext = '.xml'
                 else:
                     try:
                         with open(self.file_path, 'rb') as f:
-                            header = f.read(4)
-                            if header[:2] == b'\x1f\x8b':
+                            header = f.read(6)
+                            if header.startswith(b'\x1f\x8b'):
                                 file_ext = '.gz'
-                            elif header[:2] == b'PK':
+                            elif header.startswith(b'PK'):
                                 file_ext = '.zip'
-                            elif header[:5] == b'<?xml' or header[:5] == b'<tv>':
+                            elif header.startswith(b'\xfd7zXZ\x00'):
+                                file_ext = '.xz'
+                            elif header.startswith(b'<?xml') or header.startswith(b'<tv>'):
                                 file_ext = '.xml'
                     except Exception:
                         pass

apps/epg/tasks.py

@@ -3,6 +3,7 @@
 import logging
 import gzip
 import html.entities
+import lzma
 import os
 import uuid
 import requests
@@ -766,7 +767,7 @@ def fetch_xmltv(source):
         logger.info(f"Using existing local file for EPG source: {source.name} at {source.file_path}")
 
         # Check if the existing file is compressed and we need to extract it
-        if source.file_path.endswith(('.gz', '.zip')) and not source.file_path.endswith('.xml'):
+        if source.file_path.endswith(('.gz', '.zip', '.xz')) and not source.file_path.endswith('.xml'):
             try:
                 # Define the path for the extracted file in the cache directory
                 cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg")
@@ -1147,7 +1148,7 @@ def fetch_xmltv(source):
 
 def extract_compressed_file(file_path, output_path=None, delete_original=False):
     """
-    Extracts a compressed file (.gz or .zip) to an XML file.
+    Extracts a compressed file (.gz, .zip, or .xz) to an XML file.
 
     Args:
         file_path: Path to the compressed file
@@ -1219,6 +1220,42 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False):
 
             return extracted_path
 
+        elif format_type == 'xz':
+            logger.debug(f"Extracting xz file: {file_path}")
+            try:
+                # First check if the content is XML by reading a sample
+                with lzma.open(file_path, 'rb') as xz_file:
+                    content_sample = xz_file.read(4096)  # Read first 4KB for detection
+                    detected_format, _, _ = detect_file_format(content=content_sample)
+
+                    if detected_format != 'xml':
+                        logger.warning(f"XZ file does not appear to contain XML content: {file_path} (detected as: {detected_format})")
+                        # Continue anyway since XZ only contains one file
+
+                    # Reset file pointer and extract the content
+                    xz_file.seek(0)
+                    with open(extracted_path, 'wb') as out_file:
+                        while True:
+                            chunk = xz_file.read(MAX_EXTRACT_CHUNK_SIZE)
+                            if not chunk or len(chunk) == 0:
+                                break
+                            out_file.write(chunk)
+            except Exception as e:
+                logger.error(f"Error extracting XZ file: {e}", exc_info=True)
+                return None
+
+            logger.info(f"Successfully extracted xz file to: {extracted_path}")
+
+            # Delete original compressed file if requested
+            if delete_original:
+                try:
+                    os.remove(file_path)
+                    logger.info(f"Deleted original compressed file: {file_path}")
+                except Exception as e:
+                    logger.warning(f"Failed to delete original compressed file {file_path}: {e}")
+
+            return extracted_path
+
         elif format_type == 'zip':
             logger.debug(f"Extracting zip file: {file_path}")
             with zipfile.ZipFile(file_path, 'r') as zip_file:
@@ -4543,9 +4580,9 @@ def detect_file_format(file_path=None, content=None):
 
     Returns:
         tuple: (format_type, is_compressed, file_extension)
-        format_type: 'gzip', 'zip', 'xml', or 'unknown'
+        format_type: 'gzip', 'zip', 'xz', 'xml', or 'unknown'
         is_compressed: Boolean indicating if the file is compressed
-        file_extension: Appropriate file extension including dot (.gz, .zip, .xml)
+        file_extension: Appropriate file extension including dot (.gz, .zip, .xz, .xml)
     """
     # Default return values
     format_type = 'unknown'
@@ -4565,6 +4602,10 @@ def detect_file_format(file_path=None, content=None):
         if len(header) >= 2 and header[:2] == b'PK':
             return 'zip', True, '.zip'
 
+        # Check for xz magic number (fd 37 7a 58 5a 00)
+        if len(header) >= 6 and header[:6] == b'\xfd7zXZ\x00':
+            return 'xz', True, '.xz'
+
         # Check for XML - either standard XML header or XMLTV-specific tag
         if len(header) >= 5 and (b'<?xml' in header or b'<tv>' in header):
             return 'xml', False, '.xml'
@@ -4581,6 +4622,8 @@ def detect_file_format(file_path=None, content=None):
             return 'gzip', True, '.gz'
         elif lower_path.endswith('.zip'):
             return 'zip', True, '.zip'
+        elif lower_path.endswith('.xz'):
+            return 'xz', True, '.xz'
         elif lower_path.endswith('.xml'):
             return 'xml', False, '.xml'
 
@@ -4593,6 +4636,8 @@ def detect_file_format(file_path=None, content=None):
                 return 'gzip', True, '.gz'
             elif mime_type == 'application/zip':
                 return 'zip', True, '.zip'
+            elif mime_type == 'application/x-xz':
+                return 'xz', True, '.xz'
             elif mime_type == 'application/xml' or mime_type == 'text/xml':
                 return 'xml', False, '.xml'
 

apps/epg/tests/test_fetch_xmltv_xz.py

@@ -0,0 +1,73 @@
+"""End-to-end coverage for fetch_xmltv() against a local xz-compressed
+XMLTV file (no URL), mirroring FetchM3uLinesXzUploadTests in
+apps/m3u/tests/test_xz_playlist.py.
+
+fetch_xmltv is mocked everywhere else in the suite, so this exercises the
+local-file-with-no-url branch (apps/epg/tasks.py fetch_xmltv) end to end:
+detection, extraction via extract_compressed_file(), and the
+extracted_file_path bookkeeping - with no mocking beyond what's needed to
+keep the test hermetic (Celery Beat periodic task scheduling touches the
+django_celery_beat tables in ways unrelated to this behavior).
+"""
+import lzma
+import os
+import tempfile
+
+from django.test import TestCase
+
+from apps.epg.models import EPGSource
+from apps.epg.tasks import fetch_xmltv
+
+SAMPLE_XML = (
+    '<?xml version="1.0" encoding="UTF-8"?>\n'
+    "<tv>\n"
+    '  <channel id="xz.channel"/>\n'
+    '  <programme channel="xz.channel" '
+    'start="20000101000000 +0000" stop="20991231235959 +0000">\n'
+    "    <title>XZ Title</title>\n"
+    "  </programme>\n"
+    "</tv>\n"
+)
+
+
+class FetchXmltvXzLocalFileTests(TestCase):
+    def setUp(self):
+        self.xz_path = None
+        self.source = None
+
+    def tearDown(self):
+        if self.xz_path and os.path.exists(self.xz_path):
+            os.unlink(self.xz_path)
+        if (
+            self.source
+            and self.source.extracted_file_path
+            and os.path.exists(self.source.extracted_file_path)
+        ):
+            os.unlink(self.source.extracted_file_path)
+
+    def test_fetch_xmltv_extracts_local_xz_file(self):
+        with tempfile.NamedTemporaryFile(suffix=".xz", delete=False) as xz_file:
+            self.xz_path = xz_file.name
+            xz_file.write(lzma.compress(SAMPLE_XML.encode("utf-8")))
+
+        self.source = EPGSource.objects.create(
+            name="XZ local file source",
+            source_type="xmltv",
+            file_path=self.xz_path,
+        )
+
+        result = fetch_xmltv(self.source)
+
+        self.assertTrue(result)
+
+        self.source.refresh_from_db()
+        self.assertEqual(self.source.status, EPGSource.STATUS_SUCCESS)
+        self.assertTrue(self.source.extracted_file_path)
+        self.assertTrue(os.path.exists(self.source.extracted_file_path))
+
+        with open(self.source.extracted_file_path, "r", encoding="utf-8") as f:
+            self.assertEqual(f.read(), SAMPLE_XML)
+
+        # The original compressed file is kept - fetch_xmltv extracts
+        # without deleting the source upload.
+        self.assertTrue(os.path.exists(self.xz_path))

apps/epg/tests/test_xz_support.py

@@ -0,0 +1,291 @@
+"""Tests for native xz (.xz) support in EPG source ingestion.
+
+Covers the three dispatch points that previously only recognized gzip/zip:
+- detect_file_format() format sniffing (magic bytes, extension, mimetype)
+- extract_compressed_file() decompression
+- EPGSource.get_cache_file() extension inference for uploaded files
+"""
+import lzma
+import os
+import tempfile
+from unittest.mock import patch
+
+from django.test import SimpleTestCase
+
+from apps.epg.models import EPGSource
+from apps.epg.tasks import detect_file_format, extract_compressed_file
+
+SAMPLE_XML = (
+    '<?xml version="1.0" encoding="UTF-8"?>\n'
+    "<tv>\n"
+    '  <channel id="xz.channel"/>\n'
+    '  <programme channel="xz.channel" '
+    'start="20000101000000 +0000" stop="20991231235959 +0000">\n'
+    "    <title>XZ Title</title>\n"
+    "  </programme>\n"
+    "</tv>\n"
+)
+
+
+class DetectFileFormatXzTests(SimpleTestCase):
+    def test_detects_xz_by_magic_bytes(self):
+        compressed = lzma.compress(SAMPLE_XML.encode("utf-8"))
+
+        format_type, is_compressed, file_extension = detect_file_format(
+            content=compressed[:64]
+        )
+
+        self.assertEqual(format_type, "xz")
+        self.assertTrue(is_compressed)
+        self.assertEqual(file_extension, ".xz")
+
+    def test_detects_xz_by_extension(self):
+        format_type, is_compressed, file_extension = detect_file_format(
+            file_path="/tmp/epg_source.xz"
+        )
+
+        self.assertEqual(format_type, "xz")
+        self.assertTrue(is_compressed)
+        self.assertEqual(file_extension, ".xz")
+
+    def test_detects_xz_with_compound_extension(self):
+        # Compound extensions like .xml.xz must prioritize the compression
+        # extension, matching the existing .xml.gz behavior.
+        format_type, is_compressed, file_extension = detect_file_format(
+            file_path="/tmp/epg_source.xml.xz"
+        )
+
+        self.assertEqual(format_type, "xz")
+        self.assertTrue(is_compressed)
+        self.assertEqual(file_extension, ".xz")
+
+    def test_detects_xz_by_mimetype_fallback(self):
+        # Python's stdlib mimetypes module treats .xz as an *encoding*
+        # suffix rather than a full MIME type (mirroring .gz), so the
+        # mimetype branch is only reachable when guess_type is coerced to
+        # return the MIME type directly (e.g. a customized mimetypes
+        # configuration). Exercise that fallback branch directly, matching
+        # how the existing gzip/zip mimetype fallback branches are tested.
+        with patch("mimetypes.guess_type", return_value=("application/x-xz", None)):
+            format_type, is_compressed, file_extension = detect_file_format(
+                file_path="/tmp/epg_source_with_no_known_suffix"
+            )
+
+        self.assertEqual(format_type, "xz")
+        self.assertTrue(is_compressed)
+        self.assertEqual(file_extension, ".xz")
+
+    def test_content_magic_bytes_take_priority_over_extension(self):
+        compressed = lzma.compress(SAMPLE_XML.encode("utf-8"))
+
+        # Mismatched extension should not override a confident content match.
+        format_type, is_compressed, file_extension = detect_file_format(
+            file_path="/tmp/misleading_name.gz", content=compressed[:64]
+        )
+
+        self.assertEqual(format_type, "xz")
+        self.assertTrue(is_compressed)
+        self.assertEqual(file_extension, ".xz")
+
+
+class ExtractCompressedFileXzTests(SimpleTestCase):
+    def test_round_trips_lzma_compressed_xmltv_to_identical_xml(self):
+        xz_path = None
+        xml_path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                suffix=".xml.xz", delete=False
+            ) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(lzma.compress(SAMPLE_XML.encode("utf-8")))
+
+            xml_path = f"{os.path.splitext(os.path.splitext(xz_path)[0])[0]}.xml"
+
+            extracted_path = extract_compressed_file(xz_path, xml_path)
+
+            self.assertEqual(extracted_path, xml_path)
+            with open(extracted_path, "r", encoding="utf-8") as f:
+                self.assertEqual(f.read(), SAMPLE_XML)
+        finally:
+            if xz_path and os.path.exists(xz_path):
+                os.unlink(xz_path)
+            if xml_path and os.path.exists(xml_path):
+                os.unlink(xml_path)
+
+    def test_extracts_xz_file_detected_purely_by_magic_bytes(self):
+        # No .xz suffix at all - detection must rely on the LZMA magic
+        # number, not the filename.
+        xz_path = None
+        xml_path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                suffix=".bin", delete=False
+            ) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(lzma.compress(SAMPLE_XML.encode("utf-8")))
+
+            xml_path = f"{os.path.splitext(xz_path)[0]}.xml"
+
+            extracted_path = extract_compressed_file(xz_path, xml_path)
+
+            self.assertEqual(extracted_path, xml_path)
+            with open(extracted_path, "r", encoding="utf-8") as f:
+                self.assertEqual(f.read(), SAMPLE_XML)
+        finally:
+            if xz_path and os.path.exists(xz_path):
+                os.unlink(xz_path)
+            if xml_path and os.path.exists(xml_path):
+                os.unlink(xml_path)
+
+    def test_deletes_original_when_requested(self):
+        xz_path = None
+        xml_path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                suffix=".xml.xz", delete=False
+            ) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(lzma.compress(SAMPLE_XML.encode("utf-8")))
+
+            xml_path = f"{os.path.splitext(os.path.splitext(xz_path)[0])[0]}.xml"
+
+            extracted_path = extract_compressed_file(
+                xz_path, xml_path, delete_original=True
+            )
+
+            self.assertEqual(extracted_path, xml_path)
+            self.assertFalse(os.path.exists(xz_path))
+        finally:
+            xz_path = None  # already deleted by extract_compressed_file
+            if xml_path and os.path.exists(xml_path):
+                os.unlink(xml_path)
+
+
+class ExtractCompressedFileCorruptXzTests(SimpleTestCase):
+    def test_corrupt_xz_fails_closed_with_no_partial_output(self):
+        # Valid LZMA magic bytes (fd 37 7a 58 5a 00) followed by garbage: the
+        # format sniff reports 'xz' (magic bytes match), but decompression
+        # itself must fail. extract_compressed_file() wraps this in a
+        # blanket except Exception so a corrupt/truncated upload can never
+        # raise out of the Celery worker - it must return None instead.
+        xz_path = None
+        xml_path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                suffix=".xz", delete=False
+            ) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(b"\xfd7zXZ\x00" + b"not actually lzma data" * 20)
+
+            xml_path = f"{os.path.splitext(xz_path)[0]}.xml"
+
+            extracted_path = extract_compressed_file(xz_path, xml_path)
+
+            self.assertIsNone(extracted_path)
+            self.assertFalse(
+                os.path.exists(xml_path),
+                "extract_compressed_file must not leave a partial output file "
+                "behind when decompression fails",
+            )
+        finally:
+            if xz_path and os.path.exists(xz_path):
+                os.unlink(xz_path)
+            if xml_path and os.path.exists(xml_path):
+                os.unlink(xml_path)
+
+
+class EPGSourceGetCacheFileXzTests(SimpleTestCase):
+    def test_infers_xz_extension_from_mimetype(self):
+        xz_path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                suffix="", delete=False
+            ) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(lzma.compress(SAMPLE_XML.encode("utf-8")))
+
+            source = EPGSource(
+                name="XZ mimetype source",
+                source_type="xmltv",
+                file_path=xz_path,
+            )
+
+            with patch(
+                "mimetypes.guess_type", return_value=("application/x-xz", None)
+            ):
+                cache_file = source.get_cache_file()
+
+            self.assertTrue(cache_file.endswith(".xz"))
+        finally:
+            if xz_path and os.path.exists(xz_path):
+                os.unlink(xz_path)
+
+    def test_infers_xz_extension_from_magic_bytes(self):
+        xz_path = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                suffix="", delete=False
+            ) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(lzma.compress(SAMPLE_XML.encode("utf-8")))
+
+            source = EPGSource(
+                name="XZ magic bytes source",
+                source_type="xmltv",
+                file_path=xz_path,
+            )
+
+            # No mimetype guess available and no extension on disk - must
+            # fall back to sniffing the LZMA magic number.
+            with patch("mimetypes.guess_type", return_value=(None, None)):
+                cache_file = source.get_cache_file()
+
+            self.assertTrue(cache_file.endswith(".xz"))
+        finally:
+            if xz_path and os.path.exists(xz_path):
+                os.unlink(xz_path)
+
+
+class EPGSourceGetCacheFileRawXmlTests(SimpleTestCase):
+    """Regression coverage for the bare-<tv> raw-XML magic-byte detection.
+
+    get_cache_file()'s magic-byte read was widened from f.read(4) to
+    f.read(6) to fit the 6-byte xz signature. That widening broke the
+    fixed-length slice comparison `header[:5] == b'<tv>'`: a 5-byte slice
+    of a 6-byte header can never equal the 4-byte literal b'<tv>', so a
+    raw (uncompressed) XMLTV file with no filename extension and no
+    resolvable mimetype was silently misdetected as '.tmp' instead of
+    '.xml'. Using header.startswith(...) instead of a fixed-length slice
+    comparison is correct regardless of how many bytes are read.
+    """
+
+    def _cache_file_for_content(self, content):
+        raw_path = None
+        try:
+            with tempfile.NamedTemporaryFile(suffix="", delete=False) as raw_file:
+                raw_path = raw_file.name
+                raw_file.write(content)
+
+            source = EPGSource(
+                name="Raw XML source",
+                source_type="xmltv",
+                file_path=raw_path,
+            )
+
+            # No extension on disk and no resolvable mimetype - detection
+            # must fall back to sniffing the magic bytes.
+            with patch("mimetypes.guess_type", return_value=(None, None)):
+                return source.get_cache_file()
+        finally:
+            if raw_path and os.path.exists(raw_path):
+                os.unlink(raw_path)
+
+    def test_infers_xml_extension_from_bare_tv_tag(self):
+        cache_file = self._cache_file_for_content(SAMPLE_XML.split("\n", 1)[1].encode("utf-8"))
+
+        self.assertTrue(cache_file.endswith(".xml"))
+
+    def test_infers_xml_extension_from_xml_declaration(self):
+        cache_file = self._cache_file_for_content(SAMPLE_XML.encode("utf-8"))
+
+        self.assertTrue(cache_file.endswith(".xml"))