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)