feat(m3u): add native xz decompression support for uploaded M3U playlists

Extends _open_m3u_text_source() and fetch_m3u_lines() to treat an
uploaded .xz playlist the same way as the existing .m3u.gz path: it is
streamed lazily via lzma.open() rather than loaded fully into memory
(unlike the .zip path, which must read archive members). Uses stdlib
lzma, no new dependency.

Companion to the EPG xz support added for Dispatcharr/Dispatcharr#1414 -
the M3U upload path has the same gzip/zip dispatch structure and would
otherwise hit the same gap for an xz-compressed playlist.

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

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)