From c0d6e951a3bbdf44409b2cbea5ce28826906e951 Mon Sep 17 00:00:00 2001 From: Curt LeCaptain Date: Tue, 7 Jul 2026 19:06:31 -0500 Subject: [PATCH] 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 --- apps/m3u/tasks.py | 10 ++++- apps/m3u/tests/test_xz_playlist.py | 66 ++++++++++++++++++++++++++++++ 2 files changed, 74 insertions(+), 2 deletions(-) create mode 100644 apps/m3u/tests/test_xz_playlist.py diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index 581a390f..662a8749 100644 --- a/apps/m3u/tasks.py +++ b/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 diff --git a/apps/m3u/tests/test_xz_playlist.py b/apps/m3u/tests/test_xz_playlist.py new file mode 100644 index 00000000..8daa0a91 --- /dev/null +++ b/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)