From 408c3c6beaf13cb52b5d6cb849eaaba039e9e923 Mon Sep 17 00:00:00 2001 From: Curt LeCaptain Date: Tue, 7 Jul 2026 19:06:24 -0500 Subject: [PATCH] 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"))