Merge pull request #1421 from MotWakorb/feature/xz-epg-support

Add native xz decompression support for EPG sources and uploaded M3U playlists

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Native XZ decompression for EPG sources and uploaded M3U playlists.** `.xz`-compressed XMLTV and M3U files are now decompressed using Python's stdlib `lzma` module. EPG ingestion detects XZ via magic bytes, file extension (including compound names like `.xml.xz`), or MIME type; uploaded M3U `.xz` playlists stream line-by-line like `.gz`. The `/data/epgs` auto-import scanner now includes `.xz` files alongside `.xml`, `.gz`, and `.zip`. (Closes #1414) — Thanks [@MotWakorb](https://github.com/MotWakorb)
 - **XC catch-up (timeshift) support**. Adds a native `/timeshift/{user}/{pass}/{stream_id}/{timestamp}/{duration}.ts` endpoint that proxies replay sessions from an Xtream Codes provider to clients such as iPlayTV (Apple TV) and TiviMate. Auth reuses the same `xc_password` custom-property the live `/live/.../{id}.ts` endpoint already uses. Catch-up sessions appear on `/stats` with a violet `TIMESHIFT` badge alongside live sessions and respect per-channel access rules. (Closes #133) — Thanks [@cedric-marcoux](https://github.com/cedric-marcoux)
   - **Catch-up flags** (`tv_archive`, `tv_archive_duration`) denormalized at import time for zero-cost output queries; end-of-refresh SQL rollup updates channels linked to the refreshed account and self-heals stale flags on those channels only (e.g. after catch-up streams are removed or an account is deactivated).
   - **Strictly-UTC API surface, automatic per-provider timezone.** `server_info.timezone` is always `UTC` and the XC EPG `start`/`end` strings are emitted in UTC; the proxy converts the requested instant to the serving provider's own reported timezone (`server_info.timezone` captured on account refresh) at request time. No timezone to configure.
@@ -117,7 +118,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `sentence-transformers` 5.4.1 → 5.6.0
   - `lxml` 6.1.0 → 6.1.1
 
-
 ### Fixed
 
 - **Channel Initialization Grace Period is honoured during live stream startup.** Preview and playback no longer abort after a hardcoded 10s while the channel is still connecting with an empty buffer; the TS generator init-wait stall check and upstream health monitor now use the configured `channel_init_grace_period` (same as the server cleanup watchdog) instead of `CONNECTION_TIMEOUT`. (Fixes #1380)

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"))

apps/m3u/tasks.py

@@ -6,6 +6,7 @@ import requests
 import os
 import gc
 import gzip, zipfile
+import lzma
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from celery import shared_task
 from django.conf import settings
@@ -135,9 +136,11 @@ _EXTINF_ATTR_RE = re.compile(r'([^\s=]+)\s*=\s*(["\'])(.*?)\2')
 
 
 def _open_m3u_text_source(source_path):
-    """Open an on-disk M3U (or .m3u.gz) file for line-by-line parsing."""
+    """Open an on-disk M3U (or .m3u.gz / .m3u.xz) file for line-by-line parsing."""
     if source_path.endswith(".gz"):
         return gzip.open(source_path, "rt", encoding="utf-8")
+    if source_path.endswith(".xz"):
+        return lzma.open(source_path, "rt", encoding="utf-8")
     return open(source_path, "r", encoding="utf-8")
 
 
@@ -502,6 +505,9 @@ def fetch_m3u_lines(account, use_cache=False):
             if account.file_path.endswith(".gz"):
                 return account.file_path, True
 
+            elif account.file_path.endswith(".xz"):
+                return account.file_path, True
+
             elif account.file_path.endswith(".zip"):
                 with zipfile.ZipFile(account.file_path, "r") as zip_file:
                     for name in zip_file.namelist():
@@ -526,7 +532,7 @@ def fetch_m3u_lines(account, use_cache=False):
             else:
                 return account.file_path, True
 
-        except (IOError, OSError, zipfile.BadZipFile, gzip.BadGzipFile) as e:
+        except (IOError, OSError, zipfile.BadZipFile, gzip.BadGzipFile, lzma.LZMAError) as e:
             error_msg = f"Error opening file {account.file_path}: {e}"
             logger.error(error_msg)
             account.status = M3UAccount.Status.ERROR

apps/m3u/tests/test_xz_playlist.py

@@ -0,0 +1,66 @@
+"""Tests for native xz (.xz) support for uploaded M3U playlists.
+
+Mirrors the existing .gz handling: an uploaded .xz playlist is treated as a
+streamable text source (opened lazily via _open_m3u_text_source), never
+loaded fully into memory like the .zip path.
+"""
+import lzma
+import os
+import tempfile
+
+from django.test import SimpleTestCase, TestCase
+
+from apps.m3u.models import M3UAccount
+from apps.m3u.tasks import _open_m3u_text_source, fetch_m3u_lines
+
+SAMPLE_M3U = (
+    "#EXTM3U\n"
+    '#EXTINF:-1 tvg-id="channel.one",Channel One\n'
+    "http://example.com/stream1\n"
+)
+
+
+class OpenM3uTextSourceXzTests(SimpleTestCase):
+    def test_opens_xz_playlist_for_line_by_line_reading(self):
+        xz_path = None
+        try:
+            with tempfile.NamedTemporaryFile(suffix=".xz", delete=False) as xz_file:
+                xz_path = xz_file.name
+                xz_file.write(lzma.compress(SAMPLE_M3U.encode("utf-8")))
+
+            with _open_m3u_text_source(xz_path) as f:
+                content = f.read()
+
+            self.assertEqual(content, SAMPLE_M3U)
+        finally:
+            if xz_path and os.path.exists(xz_path):
+                os.unlink(xz_path)
+
+
+class FetchM3uLinesXzUploadTests(TestCase):
+    def setUp(self):
+        self.xz_path = None
+
+    def tearDown(self):
+        if self.xz_path and os.path.exists(self.xz_path):
+            os.unlink(self.xz_path)
+
+    def test_fetch_m3u_lines_returns_path_for_xz_upload(self):
+        with tempfile.NamedTemporaryFile(suffix=".xz", delete=False) as xz_file:
+            self.xz_path = xz_file.name
+            xz_file.write(lzma.compress(SAMPLE_M3U.encode("utf-8")))
+
+        account = M3UAccount.objects.create(
+            name="XZ upload account",
+            file_path=self.xz_path,
+        )
+
+        source, success = fetch_m3u_lines(account)
+
+        self.assertTrue(success)
+        # Like the .gz path, .xz playlists are streamed rather than loaded
+        # into memory, so fetch_m3u_lines hands back the path itself.
+        self.assertEqual(source, self.xz_path)
+
+        with _open_m3u_text_source(source) as f:
+            self.assertEqual(f.read(), SAMPLE_M3U)

core/tasks.py

@@ -177,12 +177,12 @@ def scan_and_process_files():
             epg_skipped += 1
             continue
 
-        if not filename.endswith('.xml') and not filename.endswith('.gz') and not filename.endswith('.zip'):
+        if not filename.endswith(('.xml', '.gz', '.zip', '.xz')):
             # Use trace level if not first scan
             if _first_scan_completed:
-                logger.trace(f"Skipping {filename}: Not an XML, GZ or zip file")
+                logger.trace(f"Skipping {filename}: Not an XML, GZ, ZIP, or XZ file")
             else:
-                logger.debug(f"Skipping {filename}: Not an XML, GZ or zip file")
+                logger.debug(f"Skipping {filename}: Not an XML, GZ, ZIP, or XZ file")
             epg_skipped += 1
             continue