diff --git a/apps/epg/migrations/0014_epgsource_original_file_path.py b/apps/epg/migrations/0014_epgsource_original_file_path.py new file mode 100644 index 00000000..e6c04fec --- /dev/null +++ b/apps/epg/migrations/0014_epgsource_original_file_path.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 on 2025-05-25 23:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0013_alter_epgsource_refresh_interval'), + ] + + operations = [ + migrations.AddField( + model_name='epgsource', + name='original_file_path', + field=models.CharField(blank=True, help_text='Original path to compressed file before extraction', max_length=1024, null=True), + ), + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index 27e85e29..8fd4b2f5 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -32,6 +32,8 @@ class EPGSource(models.Model): api_key = models.CharField(max_length=255, blank=True, null=True) # For Schedules Direct is_active = models.BooleanField(default=True) file_path = models.CharField(max_length=1024, blank=True, null=True) + original_file_path = models.CharField(max_length=1024, blank=True, null=True, + help_text="Original path to compressed file before extraction") refresh_interval = models.IntegerField(default=0) refresh_task = models.ForeignKey( PeriodicTask, on_delete=models.SET_NULL, null=True, blank=True diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 371e6c2d..cdf942e8 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -240,13 +240,6 @@ def fetch_xmltv(source): send_epg_update(source.id, "downloading", 100, status="error", error="No URL provided and no valid local file exists") return False - # Clean up existing cache file - if os.path.exists(source.get_cache_file()): - try: - os.remove(source.get_cache_file()) - except Exception as e: - logger.warning(f"Failed to remove existing cache file: {e}") - logger.info(f"Fetching XMLTV data from source: {source.name}") try: # Get default user agent from settings @@ -335,7 +328,16 @@ def fetch_xmltv(source): response.raise_for_status() logger.debug("XMLTV data fetched successfully.") - cache_file = source.get_cache_file() + # Define base paths for consistent file naming + cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg") + os.makedirs(cache_dir, exist_ok=True) + + # Create temporary download file with .tmp extension + temp_download_path = os.path.join(cache_dir, f"{source.id}.tmp") + + # Define final paths based on content type + compressed_path = os.path.join(cache_dir, f"{source.id}.compressed"); + xml_path = os.path.join(cache_dir, f"{source.id}.xml"); # Check if we have content length for progress tracking total_size = int(response.headers.get('content-length', 0)) @@ -344,7 +346,8 @@ def fetch_xmltv(source): last_update_time = start_time update_interval = 0.5 # Only update every 0.5 seconds - with open(cache_file, 'wb') as f: + # Download to temporary file + with open(temp_download_path, 'wb') as f: for chunk in response.iter_content(chunk_size=16384): # Increased chunk size for better performance if chunk: f.write(chunk) @@ -389,65 +392,79 @@ def fetch_xmltv(source): content_type = response.headers.get('Content-Type', '').lower() original_url = source.url.lower() - # Default extension is xml - file_extension = '.xml' + # Is this file compressed? + is_compressed = False + if 'application/x-gzip' in content_type or 'application/gzip' in content_type or 'application/zip' in content_type: + is_compressed = True + elif original_url.endswith('.gz') or original_url.endswith('.zip'): + is_compressed = True - # Check content type first - if 'application/x-gzip' in content_type or 'application/gzip' in content_type: - file_extension = '.gz' - elif 'application/zip' in content_type: - file_extension = '.zip' - # If content type didn't help, try the URL - elif original_url.endswith('.gz'): - file_extension = '.gz' - elif original_url.endswith('.zip'): - file_extension = '.zip' + # Clean up old files before saving new ones + if os.path.exists(compressed_path): + try: + os.remove(compressed_path) + logger.debug(f"Removed old compressed file: {compressed_path}") + except OSError as e: + logger.warning(f"Failed to remove old compressed file: {e}") - # Update the cache file path with the correct extension - base_cache_path = os.path.splitext(cache_file)[0] - new_cache_file = f"{base_cache_path}{file_extension}" + if os.path.exists(xml_path): + try: + os.remove(xml_path) + logger.debug(f"Removed old XML file: {xml_path}") + except OSError as e: + logger.warning(f"Failed to remove old XML file: {e}") - # Actually rename the file on disk - logger.debug(f"Renaming file from {cache_file} to {new_cache_file}") - try: - os.rename(cache_file, new_cache_file) - cache_file = new_cache_file # Update the variable to point to the renamed file - except OSError as e: - logger.error(f"Failed to rename temporary file: {e}") - # If rename fails, keep using the original file - logger.warning(f"Continuing with temporary file: {cache_file}") - new_cache_file = cache_file # Fall back to the tmp file + # Rename the temp file to appropriate final path + if is_compressed: + try: + os.rename(temp_download_path, compressed_path) + logger.debug(f"Renamed temp file to compressed file: {compressed_path}") + current_file_path = compressed_path + except OSError as e: + logger.error(f"Failed to rename temp file to compressed file: {e}") + current_file_path = temp_download_path # Fall back to using temp file + else: + try: + os.rename(temp_download_path, xml_path) + logger.debug(f"Renamed temp file to XML file: {xml_path}") + current_file_path = xml_path + except OSError as e: + logger.error(f"Failed to rename temp file to XML file: {e}") + current_file_path = temp_download_path # Fall back to using temp file # Now extract the file if it's compressed - extracted_file = None - if file_extension in ('.gz', '.zip'): + if is_compressed: try: - logger.info(f"Extracting compressed file {new_cache_file}") + logger.info(f"Extracting compressed file {current_file_path}") send_epg_update(source.id, "extracting", 0, message="Extracting downloaded file") - extracted_file = extract_compressed_file(new_cache_file) + # Always extract to the standard XML path + extracted = extract_compressed_file(current_file_path, xml_path) - if extracted_file: - logger.info(f"Successfully extracted to {extracted_file}") - send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully: {os.path.basename(extracted_file)}") + if extracted: + logger.info(f"Successfully extracted to {xml_path}") + send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully") # Update the source's file_path to the extracted XML file - source.file_path = extracted_file + source.file_path = xml_path + + # Store the original compressed file path + source.original_file_path = current_file_path else: logger.error("Extraction failed, using compressed file") send_epg_update(source.id, "extracting", 100, status="error", message="Extraction failed, using compressed file") # Use the compressed file - source.file_path = new_cache_file + source.file_path = current_file_path except Exception as e: logger.error(f"Error extracting file: {str(e)}", exc_info=True) send_epg_update(source.id, "extracting", 100, status="error", message=f"Error during extraction: {str(e)}") # Use the compressed file if extraction fails - source.file_path = new_cache_file + source.file_path = current_file_path else: # It's already an XML file - source.file_path = new_cache_file + source.file_path = current_file_path # Update the source's file_path to reflect the correct file - source.save(update_fields=['file_path', 'status']) + source.save(update_fields=['file_path', 'status', 'original_file_path']) # Update status to parsing source.status = 'parsing' @@ -559,29 +576,37 @@ def fetch_xmltv(source): return False -def extract_compressed_file(file_path, delete_original=True): +def extract_compressed_file(file_path, output_path=None, delete_original=False): """ Extracts a compressed file (.gz or .zip) to an XML file. Args: file_path: Path to the compressed file + output_path: Specific path where the file should be extracted (optional) delete_original: Whether to delete the original compressed file after successful extraction Returns: Path to the extracted XML file, or None if extraction failed """ try: - base_path = os.path.splitext(file_path)[0] - extracted_path = f"{base_path}.xml" + if output_path is None: + base_path = os.path.splitext(file_path)[0] + extracted_path = f"{base_path}.xml" + else: + extracted_path = output_path # Make sure the output path doesn't already exist if os.path.exists(extracted_path): try: os.remove(extracted_path) + logger.info(f"Removed existing extracted file: {extracted_path}") except Exception as e: logger.warning(f"Failed to remove existing extracted file: {e}") - # Try with a unique filename instead - extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml" + # If we can't delete the existing file and no specific output was requested, + # create a unique filename instead + if output_path is None: + base_path = os.path.splitext(file_path)[0] + extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml" if file_path.endswith('.gz'): logger.debug(f"Extracting gzip file: {file_path}") @@ -1588,14 +1613,14 @@ def extract_custom_properties(prog): return custom_props -def clear_element(elem): - """Clear an XML element and its parent to free memory.""" - try: - elem.clear() - parent = elem.getparent() - if parent is not None: - while elem.getprevious() is not None: - del parent[0] + + + + + + + + while elem.getprevious() is not None: if parent is not None: parent = elem.getparent() elem.clear() try: """Clear an XML element and its parent to free memory."""def clear_element(elem): del parent[0] parent.remove(elem) except Exception as e: logger.warning(f"Error clearing XML element: {e}", exc_info=True) diff --git a/core/tasks.py b/core/tasks.py index c2297569..062ffaf2 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -7,6 +7,9 @@ import logging import re import time import os +import gzip +import zipfile +import uuid from core.utils import RedisClient, send_websocket_update from apps.proxy.ts_proxy.channel_status import ChannelStatus from apps.m3u.models import M3UAccount @@ -16,11 +19,13 @@ from apps.epg.tasks import refresh_epg_data from .models import CoreSettings from apps.channels.models import Stream, ChannelStream from django.db import transaction +from django.conf import settings logger = logging.getLogger(__name__) EPG_WATCH_DIR = '/data/epgs' M3U_WATCH_DIR = '/data/m3us' +EPG_CACHE_DIR = os.path.join(settings.MEDIA_ROOT, "cached_epg") MIN_AGE_SECONDS = 6 STARTUP_SKIP_AGE = 30 REDIS_PREFIX = "processed_file:" @@ -46,6 +51,58 @@ def throttled_log(logger_method, message, key=None, *args, **kwargs): logger_method(message, *args, **kwargs) _last_log_times[key] = now +def extract_compressed_file(file_path): + """ + Extracts a compressed file (.gz or .zip) to an XML file in the cache directory. + + Args: + file_path: Path to the compressed file + + Returns: + Path to the extracted XML file, or None if extraction failed + """ + try: + # Create cache directory if it doesn't exist + os.makedirs(EPG_CACHE_DIR, exist_ok=True) + + # Generate a unique filename for the extracted file + extracted_filename = f"extracted_{uuid.uuid4().hex[:8]}.xml" + extracted_path = os.path.join(EPG_CACHE_DIR, extracted_filename) + + if file_path.endswith('.gz'): + logger.debug(f"Extracting gzip file: {file_path}") + with gzip.open(file_path, 'rb') as gz_file: + with open(extracted_path, 'wb') as out_file: + out_file.write(gz_file.read()) + logger.info(f"Successfully extracted gzip file to: {extracted_path}") + return extracted_path + + elif file_path.endswith('.zip'): + logger.debug(f"Extracting zip file: {file_path}") + with zipfile.ZipFile(file_path, 'r') as zip_file: + # Find the first XML file in the ZIP archive + xml_files = [f for f in zip_file.namelist() if f.lower().endswith('.xml')] + + if not xml_files: + logger.error("No XML file found in ZIP archive") + return None + + # Extract the first XML file + xml_content = zip_file.read(xml_files[0]) + with open(extracted_path, 'wb') as out_file: + out_file.write(xml_content) + + logger.info(f"Successfully extracted zip file to: {extracted_path}") + return extracted_path + + else: + logger.error(f"Unsupported compressed file format: {file_path}") + return None + + except Exception as e: + logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True) + return None + @shared_task def beat_periodic_task(): fetch_channel_stats() @@ -192,7 +249,7 @@ def scan_and_process_files(): # Instead of assuming old files were processed, check if they exist in the database if not stored_mtime and age > STARTUP_SKIP_AGE: # Check if this file is already in the database - existing_epg = EPGSource.objects.filter(file_path=filepath).exists() + existing_epg = EPGSource.objects.filter(original_file_path=filepath).exists() if existing_epg: # Use trace level if not first scan if _first_scan_completed: @@ -227,11 +284,39 @@ def scan_and_process_files(): continue try: - epg_source, created = EPGSource.objects.get_or_create(file_path=filepath, defaults={ - "name": filename, - "source_type": "xmltv", - "is_active": CoreSettings.get_auto_import_mapped_files() in [True, "true", "True"], - }) + extracted_path = None + is_compressed = filename.endswith('.gz') or filename.endswith('.zip') + + # Extract compressed files + if is_compressed: + logger.info(f"Detected compressed EPG file: {filename}, extracting") + extracted_path = extract_compressed_file(filepath) + + if not extracted_path: + logger.error(f"Failed to extract compressed file: {filename}") + epg_errors += 1 + continue + + logger.info(f"Successfully extracted {filename} to {extracted_path}") + + # Set the file path to use (either extracted or original) + file_to_use = extracted_path if extracted_path else filepath + + epg_source, created = EPGSource.objects.get_or_create( + original_file_path=filepath, + defaults={ + "name": filename, + "source_type": "xmltv", + "is_active": CoreSettings.get_auto_import_mapped_files() in [True, "true", "True"], + "file_path": file_to_use + } + ) + + # If the source already exists but we extracted a new file, update its file_path + if not created and extracted_path: + epg_source.file_path = extracted_path + epg_source.save(update_fields=['file_path']) + logger.info(f"Updated existing EPG source with new extracted file: {extracted_path}") redis_client.set(redis_key, mtime, ex=REDIS_TTL)