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"))