diff --git a/CHANGELOG.md b/CHANGELOG.md
index fb8dc5a7..7ce3b812 100644
--- a/CHANGELOG.md
+++ b/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)
diff --git a/apps/epg/models.py b/apps/epg/models.py
index b85453af..48c1cc78 100644
--- a/apps/epg/models.py
+++ b/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'':
+ elif header.startswith(b'\xfd7zXZ\x00'):
+ file_ext = '.xz'
+ elif header.startswith(b''):
file_ext = '.xml'
except Exception:
pass
diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py
index 6e742c63..7e145bd1 100644
--- a/apps/epg/tasks.py
+++ b/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'' 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'
diff --git a/apps/epg/tests/test_fetch_xmltv_xz.py b/apps/epg/tests/test_fetch_xmltv_xz.py
new file mode 100644
index 00000000..73f003f7
--- /dev/null
+++ b/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 = (
+ '\n'
+ "\n"
+ ' \n'
+ ' \n'
+ " XZ Title\n"
+ " \n"
+ "\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))
diff --git a/apps/epg/tests/test_xz_support.py b/apps/epg/tests/test_xz_support.py
new file mode 100644
index 00000000..5d09b780
--- /dev/null
+++ b/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 = (
+ '\n'
+ "\n"
+ ' \n'
+ ' \n'
+ " XZ Title\n"
+ " \n"
+ "\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- 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''`: a 5-byte slice
+ of a 6-byte header can never equal the 4-byte literal b'', 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"))
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)
diff --git a/core/tasks.py b/core/tasks.py
index a64469b4..eefd1697 100644
--- a/core/tasks.py
+++ b/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