From 408c3c6beaf13cb52b5d6cb849eaaba039e9e923 Mon Sep 17 00:00:00 2001 From: Curt LeCaptain Date: Tue, 7 Jul 2026 19:06:24 -0500 Subject: [PATCH 1/4] feat(epg): add native xz decompression support for EPG sources Adds .xz to the set of compressed formats detect_file_format(), extract_compressed_file(), fetch_xmltv(), and EPGSource.get_cache_file() recognize, mirroring the existing gzip handling. Uses stdlib lzma, no new dependency. Detection works via LZMA magic bytes (fd 37 7a 58 5a 00), the .xz extension (including compound extensions like .xml.xz), and the application/x-xz mimetype fallback. Widened the magic-byte read in get_cache_file() from 4 to 6 bytes since the xz signature is 6 bytes (the previous 4-byte read was sufficient only for the 2-byte gzip/zip signatures it originally supported). That widening silently broke the bare- raw-XML detection: the fixed-length slice comparison header[:5] == b'' can never match a 5-byte slice of a 6-byte header against a 4-byte literal. Replaced the fixed-length slice comparisons with header.startswith(...), which is correct regardless of how many bytes are read - this also fixes the adjacent --- apps/epg/models.py | 12 +- apps/epg/tasks.py | 53 ++++- apps/epg/tests/test_fetch_xmltv_xz.py | 73 +++++++ apps/epg/tests/test_xz_support.py | 291 ++++++++++++++++++++++++++ 4 files changed, 421 insertions(+), 8 deletions(-) create mode 100644 apps/epg/tests/test_fetch_xmltv_xz.py create mode 100644 apps/epg/tests/test_xz_support.py 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")) From c0d6e951a3bbdf44409b2cbea5ce28826906e951 Mon Sep 17 00:00:00 2001 From: Curt LeCaptain Date: Tue, 7 Jul 2026 19:06:31 -0500 Subject: [PATCH 2/4] 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) From 15d0b8f8243cf4e12d999d0fae42f21bf279c397 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Wed, 8 Jul 2026 01:21:05 +0000 Subject: [PATCH 3/4] fix(tasks): update file type checks to include .xz extension for skipping logic Modified the file type validation in scan_and_process_files() to recognize .xz files alongside .xml, .gz, and .zip. Updated logging messages to reflect the inclusion of .xz in the skipped file notifications. --- core/tasks.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 From 0aff1e1d9c6f726eca0c390a875c7ffb67b54d7e Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Wed, 8 Jul 2026 01:24:29 +0000 Subject: [PATCH 4/4] changelog: Update for pr. --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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)