diff --git a/apps/epg/migrations/0014_epgsource_extracted_file_path.py b/apps/epg/migrations/0014_epgsource_extracted_file_path.py
new file mode 100644
index 00000000..9ee1170b
--- /dev/null
+++ b/apps/epg/migrations/0014_epgsource_extracted_file_path.py
@@ -0,0 +1,18 @@
+# Generated by Django 5.1.6 on 2025-05-26 15:48
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('epg', '0013_alter_epgsource_refresh_interval'),
+ ]
+
+ operations = [
+ migrations.AddField(
+ model_name='epgsource',
+ name='extracted_file_path',
+ field=models.CharField(blank=True, help_text='Path to extracted XML file after decompression', max_length=1024, null=True),
+ ),
+ ]
diff --git a/apps/epg/models.py b/apps/epg/models.py
index dce4e21b..8abfb26f 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)
+ extracted_file_path = models.CharField(max_length=1024, blank=True, null=True,
+ help_text="Path to extracted XML file after decompression")
refresh_interval = models.IntegerField(default=0)
refresh_task = models.ForeignKey(
PeriodicTask, on_delete=models.SET_NULL, null=True, blank=True
@@ -59,8 +61,46 @@ class EPGSource(models.Model):
return self.name
def get_cache_file(self):
- # Decide on file extension
- file_ext = ".gz" if self.url.lower().endswith('.gz') else ".xml"
+ import mimetypes
+
+ # Use a temporary extension for initial download
+ # The actual extension will be determined after content inspection
+ file_ext = ".tmp"
+
+ # If file_path is already set and contains an extension, use that
+ # This handles cases where we've already detected the proper type
+ if self.file_path and os.path.exists(self.file_path):
+ _, existing_ext = os.path.splitext(self.file_path)
+ if existing_ext:
+ file_ext = existing_ext
+ else:
+ # Try to detect the MIME type and map to extension
+ mime_type, _ = mimetypes.guess_type(self.file_path)
+ if mime_type:
+ if mime_type == 'application/gzip' or mime_type == 'application/x-gzip':
+ file_ext = '.gz'
+ elif mime_type == 'application/zip':
+ file_ext = '.zip'
+ elif mime_type == 'application/xml' or mime_type == 'text/xml':
+ file_ext = '.xml'
+ # For files without mime type detection, try peeking at content
+ else:
+ try:
+ with open(self.file_path, 'rb') as f:
+ header = f.read(4)
+ # Check for gzip magic number (1f 8b)
+ if header[:2] == b'\x1f\x8b':
+ file_ext = '.gz'
+ # Check for zip magic number (PK..)
+ elif header[:2] == b'PK':
+ file_ext = '.zip'
+ # Check for XML
+ elif header[:5] == b'':
+ file_ext = '.xml'
+ except Exception as e:
+ # If we can't read the file, just keep the default extension
+ pass
+
filename = f"{self.id}{file_ext}"
# Build full path in MEDIA_ROOT/cached_epg
diff --git a/apps/epg/signals.py b/apps/epg/signals.py
index 6f98e84a..e8a004cb 100644
--- a/apps/epg/signals.py
+++ b/apps/epg/signals.py
@@ -3,8 +3,10 @@ from django.dispatch import receiver
from .models import EPGSource
from .tasks import refresh_epg_data, delete_epg_refresh_task_by_id
from django_celery_beat.models import PeriodicTask, IntervalSchedule
+from core.utils import is_protected_path
import json
import logging
+import os
logger = logging.getLogger(__name__)
@@ -95,3 +97,31 @@ def update_status_on_active_change(sender, instance, **kwargs):
except EPGSource.DoesNotExist:
# New record, will use default status
pass
+
+@receiver(post_delete, sender=EPGSource)
+def delete_cached_files(sender, instance, **kwargs):
+ """
+ Delete cached files associated with an EPGSource when it's deleted.
+ Only deletes files that aren't in protected directories.
+ """
+ # Check and delete the main file path if not protected
+ if instance.file_path and os.path.exists(instance.file_path):
+ if is_protected_path(instance.file_path):
+ logger.info(f"Skipping deletion of protected file: {instance.file_path}")
+ else:
+ try:
+ os.remove(instance.file_path)
+ logger.info(f"Deleted cached file: {instance.file_path}")
+ except OSError as e:
+ logger.error(f"Error deleting cached file {instance.file_path}: {e}")
+
+ # Check and delete the extracted file path if it exists, is different from main path, and not protected
+ if instance.extracted_file_path and os.path.exists(instance.extracted_file_path) and instance.extracted_file_path != instance.file_path:
+ if is_protected_path(instance.extracted_file_path):
+ logger.info(f"Skipping deletion of protected extracted file: {instance.extracted_file_path}")
+ else:
+ try:
+ os.remove(instance.extracted_file_path)
+ logger.info(f"Deleted extracted file: {instance.extracted_file_path}")
+ except OSError as e:
+ logger.error(f"Error deleting extracted file {instance.extracted_file_path}: {e}")
diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py
index f3f281e7..1a5f832e 100644
--- a/apps/epg/tasks.py
+++ b/apps/epg/tasks.py
@@ -11,6 +11,7 @@ import gc # Add garbage collection module
import json
from lxml import etree # Using lxml exclusively
import psutil # Add import for memory tracking
+import zipfile
from celery import shared_task
from django.conf import settings
@@ -207,6 +208,29 @@ def fetch_xmltv(source):
# Handle cases with local file but no URL
if not source.url and source.file_path and os.path.exists(source.file_path):
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'):
+ try:
+ # Define the path for the extracted file in the cache directory
+ cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg")
+ os.makedirs(cache_dir, exist_ok=True)
+ xml_path = os.path.join(cache_dir, f"{source.id}.xml")
+
+ # Extract to the cache location keeping the original
+ extracted_path = extract_compressed_file(source.file_path, xml_path, delete_original=False)
+
+ if extracted_path:
+ logger.info(f"Extracted mapped compressed file to: {extracted_path}")
+ # Update to use extracted_file_path instead of changing file_path
+ source.extracted_file_path = extracted_path
+ source.save(update_fields=['extracted_file_path'])
+ else:
+ logger.error(f"Failed to extract mapped compressed file. Using original file: {source.file_path}")
+ except Exception as e:
+ logger.error(f"Failed to extract existing compressed file: {e}")
+ # Continue with the original file if extraction fails
+
# Set the status to success in the database
source.status = 'success'
source.save(update_fields=['status'])
@@ -226,13 +250,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
@@ -321,7 +338,12 @@ 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")
# Check if we have content length for progress tracking
total_size = int(response.headers.get('content-length', 0))
@@ -330,7 +352,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)
@@ -371,11 +394,95 @@ def fetch_xmltv(source):
# Send completion notification
send_epg_update(source.id, "downloading", 100)
+ # Determine the appropriate file extension based on content detection
+ with open(temp_download_path, 'rb') as f:
+ content_sample = f.read(1024) # Just need the first 1KB to detect format
+
+ # Use our helper function to detect the format
+ format_type, is_compressed, file_extension = detect_file_format(
+ file_path=source.url, # Original URL as a hint
+ content=content_sample # Actual file content for detection
+ )
+
+ logger.debug(f"File format detection results: type={format_type}, compressed={is_compressed}, extension={file_extension}")
+
+ # Ensure consistent final paths
+ compressed_path = os.path.join(cache_dir, f"{source.id}{file_extension}" if is_compressed else f"{source.id}.compressed")
+ xml_path = os.path.join(cache_dir, f"{source.id}.xml")
+
+ # 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}")
+
+ 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}")
+
+ # 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
+ if is_compressed:
+ try:
+ logger.info(f"Extracting compressed file {current_file_path}")
+ send_epg_update(source.id, "extracting", 0, message="Extracting downloaded file")
+
+ # Always extract to the standard XML path - set delete_original to True to clean up
+ extracted = extract_compressed_file(current_file_path, xml_path, delete_original=True)
+
+ if extracted:
+ logger.info(f"Successfully extracted to {xml_path}, compressed file deleted")
+ send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully, temporary file removed")
+ # Update to store only the extracted file path since the compressed file is now gone
+ source.file_path = xml_path
+ source.extracted_file_path = None
+ 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 = current_file_path
+ source.extracted_file_path = None
+ 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 = current_file_path
+ source.extracted_file_path = None
+ else:
+ # It's already an XML file
+ source.file_path = current_file_path
+ source.extracted_file_path = None
+
+ # Update the source's file paths
+ source.save(update_fields=['file_path', 'status', 'extracted_file_path'])
+
# Update status to parsing
source.status = 'parsing'
source.save(update_fields=['status'])
- logger.info(f"Cached EPG file saved to {cache_file}")
+ logger.info(f"Cached EPG file saved to {source.file_path}")
return True
except requests.exceptions.HTTPError as e:
@@ -481,14 +588,140 @@ def fetch_xmltv(source):
return False
+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:
+ 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}")
+ # 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"
+
+ # Use our detection helper to determine the file format instead of relying on extension
+ with open(file_path, 'rb') as f:
+ content_sample = f.read(4096) # Read a larger sample to ensure accurate detection
+
+ format_type, is_compressed, _ = detect_file_format(file_path=file_path, content=content_sample)
+
+ if format_type == 'gzip':
+ logger.debug(f"Extracting gzip file: {file_path}")
+ try:
+ # First check if the content is XML by reading a sample
+ with gzip.open(file_path, 'rb') as gz_file:
+ content_sample = gz_file.read(4096) # Read first 4KB for detection
+ detected_format, _, _ = detect_file_format(content=content_sample)
+
+ if detected_format != 'xml':
+ logger.warning(f"GZIP file does not appear to contain XML content: {file_path} (detected as: {detected_format})")
+ # Continue anyway since GZIP only contains one file
+
+ # Reset file pointer and extract the content
+ gz_file.seek(0)
+ with open(extracted_path, 'wb') as out_file:
+ out_file.write(gz_file.read())
+ except Exception as e:
+ logger.error(f"Error extracting GZIP file: {e}", exc_info=True)
+ return None
+
+ logger.info(f"Successfully extracted gzip 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:
+ # 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.info("No files with .xml extension found in ZIP archive, checking content of all files")
+ # Check content of each file to see if any are XML without proper extension
+ for filename in zip_file.namelist():
+ if not filename.endswith('/'): # Skip directories
+ try:
+ # Read a sample of the file content
+ content_sample = zip_file.read(filename, 4096) # Read up to 4KB for detection
+ format_type, _, _ = detect_file_format(content=content_sample)
+ if format_type == 'xml':
+ logger.info(f"Found XML content in file without .xml extension: {filename}")
+ xml_files = [filename]
+ break
+ except Exception as e:
+ logger.warning(f"Error reading file {filename} from ZIP: {e}")
+
+ 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}")
+
+ # 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
+
+ else:
+ logger.error(f"Unsupported or unrecognized compressed file format: {file_path} (detected as: {format_type})")
+ return None
+
+ except Exception as e:
+ logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True)
+ return None
+
+
def parse_channels_only(source):
- file_path = source.file_path
+ # Use extracted file if available, otherwise use the original file path
+ file_path = source.extracted_file_path if source.extracted_file_path else source.file_path
if not file_path:
file_path = source.get_cache_file()
# Send initial parsing notification
send_epg_update(source.id, "parsing_channels", 0)
+ process = None
+ should_log_memory = False
+
try:
# Check if the file exists
if not os.path.exists(file_path):
@@ -517,14 +750,17 @@ def parse_channels_only(source):
return False
# Verify the file was downloaded successfully
- if not os.path.exists(new_path):
- logger.error(f"Failed to fetch EPG data, file still missing at: {new_path}")
+ if not os.path.exists(source.file_path):
+ logger.error(f"Failed to fetch EPG data, file still missing at: {source.file_path}")
# Update status to error
source.status = 'error'
source.last_message = f"Failed to fetch EPG data, file missing after download"
source.save(update_fields=['status', 'last_message'])
send_epg_update(source.id, "parsing_channels", 100, status="error", error="File not found after download")
return False
+
+ # Update file_path with the new location
+ file_path = source.file_path
else:
logger.error(f"No URL provided for EPG source {source.name}, cannot fetch new data")
# Update status to error
@@ -535,7 +771,7 @@ def parse_channels_only(source):
# Initialize process variable for memory tracking only in debug mode
try:
process = None
- # Get current log level as a number rather than string
+ # Get current log level as a number
current_log_level = logger.getEffectiveLevel()
# Only track memory usage when log level is DEBUG (10) or more verbose
@@ -546,8 +782,6 @@ def parse_channels_only(source):
process = psutil.Process()
initial_memory = process.memory_info().rss / 1024 / 1024
logger.debug(f"[parse_channels_only] Initial memory usage: {initial_memory:.2f} MB")
- else:
- logger.debug("Memory tracking disabled in production mode")
except (ImportError, NameError):
process = None
should_log_memory = False
@@ -574,8 +808,7 @@ def parse_channels_only(source):
send_epg_update(source.id, "parsing_channels", 10)
# Stream parsing instead of loading entire file at once
- is_gzipped = file_path.endswith('.gz')
-
+ # This can be simplified since we now always have XML files
epgs_to_create = []
epgs_to_update = []
total_channels = 0
@@ -585,41 +818,28 @@ def parse_channels_only(source):
# Track memory at key points
if process:
- logger.info(f"[parse_channels_only] Memory before opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
+ logger.debug(f"[parse_channels_only] Memory before opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
try:
- # Create a parser with the desired options
- #parser = etree.XMLParser(huge_tree=True, remove_blank_text=True)
-
- # Count channels for progress reporting - use proper lxml approach
- # Open the file first
- logger.info(f"Opening file for initial channel count: {file_path}")
- source_file = gzip.open(file_path, 'rb') if is_gzipped else open(file_path, 'rb')
- if process:
- logger.info(f"[parse_channels_only] Memory after opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
-
- # Count channels
+ # Attempt to count existing channels in the database
try:
total_channels = EPGData.objects.filter(epg_source=source).count()
logger.info(f"Found {total_channels} existing channels for this source")
except Exception as e:
logger.error(f"Error counting channels: {e}")
total_channels = 500 # Default estimate
-
- # Close the file to reset position
- logger.debug(f"Closing initial file handle")
- source_file.close()
if process:
logger.debug(f"[parse_channels_only] Memory after closing initial file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
# Update progress after counting
send_epg_update(source.id, "parsing_channels", 25, total_channels=total_channels)
- # Reset file position for actual processing
- logger.debug(f"Re-opening file for channel parsing: {file_path}")
- source_file = gzip.open(file_path, 'rb') if is_gzipped else open(file_path, 'rb')
+ # Open the file - no need to check file type since it's always XML now
+ logger.debug(f"Opening file for channel parsing: {file_path}")
+ source_file = open(file_path, 'rb')
+
if process:
- logger.debug(f"[parse_channels_only] Memory after re-opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
+ logger.debug(f"[parse_channels_only] Memory after opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB")
# Change iterparse to look for both channel and programme elements
logger.debug(f"Creating iterparse context for channels and programmes")
@@ -632,17 +852,6 @@ def parse_channels_only(source):
for _, elem in channel_parser:
total_elements_processed += 1
- # If we encounter a programme element, we've processed all channels
- # Break out of the loop to avoid memory spike
- if elem.tag == 'programme':
- logger.debug(f"[parse_channels_only] Found first programme element after processing {processed_channels} channels - exiting channel parsing")
- # Clean up the element before breaking
- elem.clear()
- parent = elem.getparent()
- if parent is not None:
- parent.remove(elem)
- break
-
# Only process channel elements
if elem.tag == 'channel':
channel_count += 1
@@ -673,6 +882,7 @@ def parse_channels_only(source):
epg_source=source,
))
logger.debug(f"[parse_channels_only] Added new channel to epgs_to_create 1: {tvg_id} - {display_name}")
+ processed_channels += 1
continue
# We use the cached object to check if the name has changed
@@ -682,6 +892,9 @@ def parse_channels_only(source):
epg_obj.name = display_name
epgs_to_update.append(epg_obj)
logger.debug(f"[parse_channels_only] Added channel to update to epgs_to_update: {tvg_id} - {display_name}")
+ else:
+ # No changes needed, just clear the element
+ logger.debug(f"[parse_channels_only] No changes needed for channel {tvg_id} - {display_name}")
else:
# This is a new channel that doesn't exist in our database
epgs_to_create.append(EPGData(
@@ -701,7 +914,7 @@ def parse_channels_only(source):
logger.info(f"[parse_channels_only] Memory after bulk_create: {process.memory_info().rss / 1024 / 1024:.2f} MB")
del epgs_to_create # Explicit deletion
epgs_to_create = []
- gc.collect()
+ cleanup_memory(log_usage=should_log_memory, force_collection=True)
if process:
logger.info(f"[parse_channels_only] Memory after gc.collect(): {process.memory_info().rss / 1024 / 1024:.2f} MB")
@@ -714,13 +927,13 @@ def parse_channels_only(source):
logger.info(f"[parse_channels_only] Memory after bulk_update: {process.memory_info().rss / 1024 / 1024:.2f} MB")
epgs_to_update = []
# Force garbage collection
- gc.collect()
+ cleanup_memory(log_usage=should_log_memory, force_collection=True)
# Periodically clear the existing_epgs cache to prevent memory buildup
if processed_channels % 1000 == 0:
logger.info(f"[parse_channels_only] Clearing existing_epgs cache at {processed_channels} channels")
existing_epgs.clear()
- gc.collect()
+ cleanup_memory(log_usage=should_log_memory, force_collection=True)
if process:
logger.info(f"[parse_channels_only] Memory after clearing cache: {process.memory_info().rss / 1024 / 1024:.2f} MB")
@@ -734,105 +947,28 @@ def parse_channels_only(source):
processed=processed_channels,
total=total_channels
)
- logger.debug(f"[parse_channels_only] Processed channel: {tvg_id} - {display_name}")
+ if processed_channels > total_channels:
+ logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - processed {processed_channels - total_channels} additional channels")
+ else:
+ logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - processed {processed_channels}/{total_channels}")
if process:
- logger.info(f"[parse_channels_only] Memory before elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
+ logger.debug(f"[parse_channels_only] Memory before elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
# Clear memory
try:
# First clear the element's content
- elem.clear()
-
- # Get the parent before we might lose reference to it
- parent = elem.getparent()
- if parent is not None:
- # Clean up preceding siblings
- while elem.getprevious() is not None:
- del parent[0]
-
- # Try to fully detach this element from parent
- try:
- parent.remove(elem)
- del elem
- del parent
- except (ValueError, KeyError, TypeError):
- # Element might already be removed or detached
- pass
- cleanup_memory(log_usage=should_log_memory, force_collection=True)
+ clear_element(elem)
except Exception as e:
# Just log the error and continue - don't let cleanup errors stop processing
logger.debug(f"[parse_channels_only] Non-critical error during XML element cleanup: {e}")
if process:
- logger.info(f"[parse_channels_only] Memory after elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
- # Check if we should break early to avoid excessive sleep
- if processed_channels >= total_channels and total_channels > 0:
- logger.info(f"[parse_channels_only] Expected channel numbers hit, continuing - processed {processed_channels}/{total_channels}")
- if process:
- logger.debug(f"[parse_channels_only] Memory usage after {processed_channels}: {process.memory_info().rss / 1024 / 1024:.2f} MB")
+ logger.debug(f"[parse_channels_only] Memory after elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
logger.debug(f"[parse_channels_only] Total elements processed: {total_elements_processed}")
- # Add periodic forced cleanup based on TOTAL ELEMENTS, not just channels
- # This ensures we clean up even if processing many non-channel elements
- if total_elements_processed % 1000 == 0:
- logger.info(f"[parse_channels_only] Performing preventative memory cleanup after {total_elements_processed} elements (found {processed_channels} channels)")
- # Close and reopen the parser to release memory
- if source_file and channel_parser:
- # First clear element references - safely with checks
- if 'elem' in locals() and elem is not None:
- try:
- elem.clear()
- parent = elem.getparent()
- if parent is not None:
- parent.remove(elem)
- except Exception as e:
- logger.debug(f"Non-critical error during cleanup: {e}")
- # Reset parser state
- del channel_parser
- channel_parser = None
- gc.collect()
-
- # Perform thorough cleanup
- cleanup_memory(log_usage=should_log_memory, force_collection=True)
-
- # Create a new parser context - continue looking for both tags
- # This doesn't restart from the beginning but continues from current position
- channel_parser = etree.iterparse(source_file, events=('end',), tag=('channel', 'programme'))
- logger.info(f"[parse_channels_only] Recreated parser context after memory cleanup")
-
- # Also do cleanup based on processed channels as before
- elif processed_channels % 1000 == 0 and processed_channels > 0:
- logger.info(f"[parse_channels_only] Performing preventative memory cleanup at {processed_channels} channels")
- # Close and reopen the parser to release memory
- if source_file and channel_parser:
- # First clear element references - safely with checks
- if 'elem' in locals() and elem is not None:
- try:
- elem.clear()
- parent = elem.getparent()
- if parent is not None:
- parent.remove(elem)
- except Exception as e:
- logger.debug(f"Non-critical error during cleanup: {e}")
-
- # Reset parser state
- del channel_parser
- channel_parser = None
- gc.collect()
-
- # Perform thorough cleanup
- cleanup_memory(log_usage=should_log_memory, force_collection=True)
-
- # Create a new parser context
- # This doesn't restart from the beginning but continues from current position
- channel_parser = etree.iterparse(source_file, events=('end',), tag=('channel', 'programme'))
- logger.info(f"[parse_channels_only] Recreated parser context after memory cleanup")
-
- if processed_channels == total_channels:
- if process:
- logger.info(f"[parse_channels_only] Processed all channels current memory: {process.memory_info().rss / 1024 / 1024:.2f} MB")
- else:
- logger.info(f"[parse_channels_only] Processed all channels")
+ else:
+ clear_element(elem)
+ continue
except (etree.XMLSyntaxError, Exception) as xml_error:
logger.error(f"[parse_channels_only] XML parsing failed: {xml_error}")
@@ -843,8 +979,9 @@ def parse_channels_only(source):
send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(xml_error))
return False
if process:
- logger.debug(f"[parse_channels_only] Memory before final batch creation: {process.memory_info().rss / 1024 / 1024:.2f} MB")
-
+ logger.info(f"[parse_channels_only] Processed {processed_channels} channels current memory: {process.memory_info().rss / 1024 / 1024:.2f} MB")
+ else:
+ logger.info(f"[parse_channels_only] Processed {processed_channels} channels")
# Process any remaining items
if epgs_to_create:
EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True)
@@ -893,8 +1030,9 @@ def parse_channels_only(source):
send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(e))
return False
finally:
- # Add more detailed cleanup in finally block
- logger.debug("In finally block, ensuring cleanup")
+ # Cleanup memory and close file
+ if process:
+ logger.debug(f"[parse_channels_only] Memory before cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB")
try:
if 'channel_parser' in locals():
del channel_parser
@@ -936,6 +1074,7 @@ def parse_programs_for_tvg_id(epg_id):
source_file = None
program_parser = None
programs_to_create = []
+ programs_processed = 0
try:
# Add memory tracking only in trace mode or higher
try:
@@ -943,16 +1082,14 @@ def parse_programs_for_tvg_id(epg_id):
# Get current log level as a number
current_log_level = logger.getEffectiveLevel()
- # Only track memory usage when log level is TRACE or more verbose
- should_log_memory = current_log_level <= 5
+ # Only track memory usage when log level is TRACE or more verbose or if running in DEBUG mode
+ should_log_memory = current_log_level <= 5 or settings.DEBUG
if should_log_memory:
process = psutil.Process()
initial_memory = process.memory_info().rss / 1024 / 1024
logger.info(f"[parse_programs_for_tvg_id] Initial memory usage: {initial_memory:.2f} MB")
mem_before = initial_memory
- else:
- logger.debug("Memory tracking disabled in production mode")
except ImportError:
process = None
should_log_memory = False
@@ -971,7 +1108,7 @@ def parse_programs_for_tvg_id(epg_id):
# This is faster for most database engines
ProgramData.objects.filter(epg=epg).delete()
- file_path = epg_source.file_path
+ file_path = epg_source.extracted_file_path if epg_source.extracted_file_path else epg_source.file_path
if not file_path:
file_path = epg_source.get_cache_file()
@@ -979,15 +1116,19 @@ def parse_programs_for_tvg_id(epg_id):
if not os.path.exists(file_path):
logger.error(f"EPG file not found at: {file_path}")
- # Update the file path in the database
- new_path = epg_source.get_cache_file()
- logger.info(f"Updating file_path from '{file_path}' to '{new_path}'")
- epg_source.file_path = new_path
- epg_source.save(update_fields=['file_path'])
+ if epg_source.url:
+ # Update the file path in the database
+ new_path = epg_source.get_cache_file()
+ logger.info(f"Updating file_path from '{file_path}' to '{new_path}'")
+ epg_source.file_path = new_path
+ epg_source.save(update_fields=['file_path'])
+ logger.info(f"Fetching new EPG data from URL: {epg_source.url}")
+ else:
+ logger.info(f"EPG source does not have a URL, using existing file path: {file_path} to rebuild cache")
# Fetch new data before continuing
- if epg_source.url:
- logger.info(f"Fetching new EPG data from URL: {epg_source.url}")
+ if epg_source:
+
# Properly check the return value from fetch_xmltv
fetch_success = fetch_xmltv(epg_source)
@@ -1003,14 +1144,20 @@ def parse_programs_for_tvg_id(epg_id):
return
# Also check if the file exists after download
- if not os.path.exists(new_path):
- logger.error(f"Failed to fetch EPG data, file still missing at: {new_path}")
+ if not os.path.exists(epg_source.file_path):
+ logger.error(f"Failed to fetch EPG data, file still missing at: {epg_source.file_path}")
epg_source.status = 'error'
epg_source.last_message = f"Failed to download EPG data, file missing after download"
epg_source.save(update_fields=['status', 'last_message'])
send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="File not found after download")
release_task_lock('parse_epg_programs', epg_id)
return
+
+ # Update file_path with the new location
+ if epg_source.extracted_file_path:
+ file_path = epg_source.extracted_file_path
+ else:
+ file_path = epg_source.file_path
else:
logger.error(f"No URL provided for EPG source {epg_source.name}, cannot fetch new data")
# Update status to error
@@ -1021,29 +1168,26 @@ def parse_programs_for_tvg_id(epg_id):
release_task_lock('parse_epg_programs', epg_id)
return
- file_path = new_path
-
# Use streaming parsing to reduce memory usage
- is_gzipped = file_path.endswith('.gz')
-
- logger.info(f"Parsing programs for tvg_id={epg.tvg_id} from {file_path}")
+ # No need to check file type anymore since it's always XML
+ logger.debug(f"Parsing programs for tvg_id={epg.tvg_id} from {file_path}")
# Memory usage tracking
if process:
try:
mem_before = process.memory_info().rss / 1024 / 1024
- logger.info(f"[parse_programs_for_tvg_id] Memory before parsing {epg.tvg_id} - {mem_before:.2f} MB")
+ logger.debug(f"[parse_programs_for_tvg_id] Memory before parsing {epg.tvg_id} - {mem_before:.2f} MB")
except Exception as e:
logger.warning(f"Error tracking memory: {e}")
mem_before = 0
programs_to_create = []
batch_size = 1000 # Process in batches to limit memory usage
- programs_processed = 0
try:
- # Open the file properly
- source_file = gzip.open(file_path, 'rb') if is_gzipped else open(file_path, 'rb')
+ # Open the file directly - no need to check compression
+ logger.debug(f"Opening file for parsing: {file_path}")
+ source_file = open(file_path, 'rb')
# Stream parse the file using lxml's iterparse
program_parser = etree.iterparse(source_file, events=('end',), tag='programme')
@@ -1091,22 +1235,8 @@ def parse_programs_for_tvg_id(epg_id):
custom_properties=custom_properties_json
))
programs_processed += 1
- del custom_props
- del custom_properties_json
- del start_time
- del end_time
- del title
- del desc
- del sub_title
- elem.clear()
- parent = elem.getparent()
- if parent is not None:
- while elem.getprevious() is not None:
- del parent[0]
- parent.remove(elem)
- del elem
- del parent
- #gc.collect()
+ # Clear the element to free memory
+ clear_element(elem)
# Batch processing
if len(programs_to_create) >= batch_size:
ProgramData.objects.bulk_create(programs_to_create)
@@ -1120,41 +1250,10 @@ def parse_programs_for_tvg_id(epg_id):
logger.error(f"Error processing program for {epg.tvg_id}: {e}", exc_info=True)
else:
# Immediately clean up non-matching elements to reduce memory pressure
- elem.clear()
- parent = elem.getparent()
- if parent is not None:
- while elem.getprevious() is not None:
- del parent[0]
- try:
- parent.remove(elem)
- except (ValueError, KeyError, TypeError):
- pass
- del elem
+ if elem is not None:
+ clear_element(elem)
continue
- # Important: Clear the element to avoid memory leaks using a more robust approach
- try:
- # First clear the element's content
- elem.clear()
- # Get the parent before we might lose reference to it
- parent = elem.getparent()
- if parent is not None:
- # Clean up preceding siblings
- while elem.getprevious() is not None:
- del parent[0]
- # Try to fully detach this element from parent
- try:
- parent.remove(elem)
- del elem
- del parent
- except (ValueError, KeyError, TypeError):
- # Element might already be removed or detached
- pass
-
- except Exception as e:
- # Just log the error and continue - don't let cleanup errors stop processing
- logger.trace(f"Non-critical error during XML element cleanup: {e}")
-
# Make sure to close the file and release parser resources
if source_file:
source_file.close()
@@ -1165,6 +1264,9 @@ def parse_programs_for_tvg_id(epg_id):
gc.collect()
+ except zipfile.BadZipFile as zip_error:
+ logger.error(f"Bad ZIP file: {zip_error}")
+ raise
except etree.XMLSyntaxError as xml_error:
logger.error(f"XML syntax error parsing program data: {xml_error}")
raise
@@ -1246,8 +1348,6 @@ def parse_programs_for_source(epg_source, tvg_id=None):
process = psutil.Process()
initial_memory = process.memory_info().rss / 1024 / 1024
logger.info(f"[parse_programs_for_source] Initial memory usage: {initial_memory:.2f} MB")
- else:
- logger.debug("Memory tracking disabled in production mode")
except ImportError:
logger.warning("psutil not available for memory tracking")
process = None
@@ -1569,3 +1669,82 @@ def extract_custom_properties(prog):
custom_props[kw.replace('-', '_')] = True
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]
+ parent.remove(elem)
+ except Exception as e:
+ logger.warning(f"Error clearing XML element: {e}", exc_info=True)
+
+
+def detect_file_format(file_path=None, content=None):
+ """
+ Detect file format by examining content or file path.
+
+ Args:
+ file_path: Path to file (optional)
+ content: Raw file content bytes (optional)
+
+ Returns:
+ tuple: (format_type, is_compressed, file_extension)
+ format_type: 'gzip', 'zip', 'xml', or 'unknown'
+ is_compressed: Boolean indicating if the file is compressed
+ file_extension: Appropriate file extension including dot (.gz, .zip, .xml)
+ """
+ # Default return values
+ format_type = 'unknown'
+ is_compressed = False
+ file_extension = '.tmp'
+
+ # First priority: check content magic numbers as they're most reliable
+ if content:
+ # We only need the first few bytes for magic number detection
+ header = content[:20] if len(content) >= 20 else content
+
+ # Check for gzip magic number (1f 8b)
+ if len(header) >= 2 and header[:2] == b'\x1f\x8b':
+ return 'gzip', True, '.gz'
+
+ # Check for zip magic number (PK..)
+ if len(header) >= 2 and header[:2] == b'PK':
+ return 'zip', True, '.zip'
+
+ # Check for XML - either standard XML header or XMLTV-specific tag
+ if len(header) >= 5 and (b'' in header):
+ return 'xml', False, '.xml'
+
+ # Second priority: check file extension - focus on the final extension for compression
+ if file_path:
+ logger.debug(f"Detecting file format for: {file_path}")
+
+ # Handle compound extensions like .xml.gz - prioritize compression extensions
+ lower_path = file_path.lower()
+
+ # Check for compression extensions explicitly
+ if lower_path.endswith('.gz') or lower_path.endswith('.gzip'):
+ return 'gzip', True, '.gz'
+ elif lower_path.endswith('.zip'):
+ return 'zip', True, '.zip'
+ elif lower_path.endswith('.xml'):
+ return 'xml', False, '.xml'
+
+ # Fallback to mimetypes only if direct extension check doesn't work
+ import mimetypes
+ mime_type, _ = mimetypes.guess_type(file_path)
+ logger.debug(f"Guessed MIME type: {mime_type}")
+ if mime_type:
+ if mime_type == 'application/gzip' or mime_type == 'application/x-gzip':
+ return 'gzip', True, '.gz'
+ elif mime_type == 'application/zip':
+ return 'zip', True, '.zip'
+ elif mime_type == 'application/xml' or mime_type == 'text/xml':
+ return 'xml', False, '.xml'
+
+ # If we reach here, we couldn't reliably determine the format
+ return format_type, is_compressed, file_extension
diff --git a/apps/hdhr/api_views.py b/apps/hdhr/api_views.py
index 62dea356..95bacace 100644
--- a/apps/hdhr/api_views.py
+++ b/apps/hdhr/api_views.py
@@ -100,13 +100,19 @@ class DiscoverAPIView(APIView):
f"Calculated tuner count: {tuner_count} (limited profiles: {limited_tuners}, custom streams: {custom_stream_count}, unlimited: {has_unlimited})"
)
+ # Create a unique DeviceID for the HDHomeRun device based on profile ID or a default value
+ device_ID = "12345678" # Default DeviceID
+ friendly_name = "Dispatcharr HDHomeRun"
+ if profile is not None:
+ device_ID = f"dispatcharr-hdhr-{profile}"
+ friendly_name = f"Dispatcharr HDHomeRun - {profile}"
if not device:
data = {
- "FriendlyName": "Dispatcharr HDHomeRun",
+ "FriendlyName": friendly_name,
"ModelNumber": "HDTC-2US",
"FirmwareName": "hdhomerun3_atsc",
"FirmwareVersion": "20200101",
- "DeviceID": "12345678",
+ "DeviceID": device_ID,
"DeviceAuth": "test_auth_token",
"BaseURL": base_url,
"LineupURL": f"{base_url}/lineup.json",
diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py
index cebcc545..3d0a53d9 100644
--- a/apps/proxy/ts_proxy/server.py
+++ b/apps/proxy/ts_proxy/server.py
@@ -464,6 +464,28 @@ class ProxyServer:
def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None):
"""Initialize a channel without redundant active key"""
try:
+ # IMPROVED: First check if channel is already being initialized by another process
+ if self.redis_client:
+ metadata_key = RedisKeys.channel_metadata(channel_id)
+ if self.redis_client.exists(metadata_key):
+ metadata = self.redis_client.hgetall(metadata_key)
+ if b'state' in metadata:
+ state = metadata[b'state'].decode('utf-8')
+ active_states = [ChannelState.INITIALIZING, ChannelState.CONNECTING,
+ ChannelState.WAITING_FOR_CLIENTS, ChannelState.ACTIVE]
+ if state in active_states:
+ logger.info(f"Channel {channel_id} already being initialized with state {state}")
+ # Create buffer and client manager only if we don't have them
+ if channel_id not in self.stream_buffers:
+ self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=self.redis_client)
+ if channel_id not in self.client_managers:
+ self.client_managers[channel_id] = ClientManager(
+ channel_id,
+ redis_client=self.redis_client,
+ worker_id=self.worker_id
+ )
+ return True
+
# Create buffer and client manager instances
buffer = StreamBuffer(channel_id, redis_client=self.redis_client)
client_manager = ClientManager(
@@ -476,6 +498,20 @@ class ProxyServer:
self.stream_buffers[channel_id] = buffer
self.client_managers[channel_id] = client_manager
+ # IMPROVED: Set initializing state in Redis BEFORE any other operations
+ if self.redis_client:
+ # Set early initialization state to prevent race conditions
+ metadata_key = RedisKeys.channel_metadata(channel_id)
+ initial_metadata = {
+ "state": ChannelState.INITIALIZING,
+ "init_time": str(time.time()),
+ "owner": self.worker_id
+ }
+ if stream_id:
+ initial_metadata["stream_id"] = str(stream_id)
+ self.redis_client.hset(metadata_key, mapping=initial_metadata)
+ logger.info(f"Set early initializing state for channel {channel_id}")
+
# Get channel URL from Redis if available
channel_url = url
channel_user_agent = user_agent
diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py
index 97e02c16..817a7b82 100644
--- a/apps/proxy/ts_proxy/stream_generator.py
+++ b/apps/proxy/ts_proxy/stream_generator.py
@@ -120,9 +120,19 @@ class StreamGenerator:
yield create_ts_packet('error', f"Error: {error_message}")
return False
else:
+ # Improved logging to track initialization progress
+ init_time = "unknown"
+ if b'init_time' in metadata:
+ try:
+ init_time_float = float(metadata[b'init_time'].decode('utf-8'))
+ init_duration = time.time() - init_time_float
+ init_time = f"{init_duration:.1f}s ago"
+ except:
+ pass
+
# Still initializing - send keepalive if needed
if time.time() - last_keepalive >= keepalive_interval:
- status_msg = f"Initializing: {state}"
+ status_msg = f"Initializing: {state} (started {init_time})"
keepalive_packet = create_ts_packet('keepalive', status_msg)
logger.debug(f"[{self.client_id}] Sending keepalive packet during initialization, state={state}")
yield keepalive_packet
diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py
index 7949c33b..4d632bfd 100644
--- a/apps/proxy/ts_proxy/views.py
+++ b/apps/proxy/ts_proxy/views.py
@@ -67,6 +67,7 @@ def stream_ts(request, channel_id):
# Check if we need to reinitialize the channel
needs_initialization = True
channel_state = None
+ channel_initializing = False
# Get current channel state from Redis if available
if proxy_server.redis_client:
@@ -77,28 +78,46 @@ def stream_ts(request, channel_id):
if state_field in metadata:
channel_state = metadata[state_field].decode("utf-8")
- # Only skip initialization if channel is in a healthy state
- valid_states = [
- ChannelState.ACTIVE,
+ # IMPROVED: Check for *any* state that indicates initialization is in progress
+ active_states = [
+ ChannelState.INITIALIZING,
+ ChannelState.CONNECTING,
ChannelState.WAITING_FOR_CLIENTS,
+ ChannelState.ACTIVE,
]
- if channel_state in valid_states:
- # Verify the owner is still active
+ if channel_state in active_states:
+ # Channel is being initialized or already active - no need for reinitialization
+ needs_initialization = False
+ logger.debug(
+ f"[{client_id}] Channel {channel_id} already in state {channel_state}, skipping initialization"
+ )
+
+ # Special handling for initializing/connecting states
+ if channel_state in [
+ ChannelState.INITIALIZING,
+ ChannelState.CONNECTING,
+ ]:
+ channel_initializing = True
+ logger.debug(
+ f"[{client_id}] Channel {channel_id} is still initializing, client will wait for completion"
+ )
+ else:
+ # Only check for owner if channel is in a valid state
owner_field = ChannelMetadataField.OWNER.encode("utf-8")
if owner_field in metadata:
owner = metadata[owner_field].decode("utf-8")
owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
if proxy_server.redis_client.exists(owner_heartbeat_key):
- # Owner is active and channel is in good state
+ # Owner is still active, so we don't need to reinitialize
needs_initialization = False
- logger.info(
- f"[{client_id}] Channel {channel_id} in state {channel_state} with active owner {owner}"
+ logger.debug(
+ f"[{client_id}] Channel {channel_id} has active owner {owner}"
)
# Start initialization if needed
- channel_initializing = False
if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
- # Force cleanup of any previous instance
+ logger.info(f"[{client_id}] Starting channel {channel_id} initialization")
+ # Force cleanup of any previous instance if in terminal state
if channel_state in [
ChannelState.ERROR,
ChannelState.STOPPING,
@@ -109,9 +128,6 @@ def stream_ts(request, channel_id):
)
proxy_server.stop_channel(channel_id)
- # Initialize the channel (but don't wait for completion)
- logger.info(f"[{client_id}] Starting channel {channel_id} initialization")
-
# Use max retry attempts and connection timeout from config
max_retries = ConfigHelper.max_retries()
retry_timeout = ConfigHelper.connection_timeout()
diff --git a/core/tasks.py b/core/tasks.py
index eee5b18f..0fdaedf7 100644
--- a/core/tasks.py
+++ b/core/tasks.py
@@ -175,12 +175,12 @@ def scan_and_process_files():
epg_skipped += 1
continue
- if not filename.endswith('.xml') and not filename.endswith('.gz'):
+ if not filename.endswith('.xml') and not filename.endswith('.gz') and not filename.endswith('.zip'):
# Use trace level if not first scan
if _first_scan_completed:
- logger.trace(f"Skipping {filename}: Not an XML or GZ file")
+ logger.trace(f"Skipping {filename}: Not an XML, GZ or zip file")
else:
- logger.debug(f"Skipping {filename}: Not an XML or GZ file")
+ logger.debug(f"Skipping {filename}: Not an XML, GZ or zip file")
epg_skipped += 1
continue
diff --git a/core/utils.py b/core/utils.py
index 039b0695..9951ce26 100644
--- a/core/utils.py
+++ b/core/utils.py
@@ -303,3 +303,30 @@ def cleanup_memory(log_usage=False, force_collection=True):
except (ImportError, Exception):
pass
logger.trace("Memory cleanup complete for django")
+
+def is_protected_path(file_path):
+ """
+ Determine if a file path is in a protected directory that shouldn't be deleted.
+
+ Args:
+ file_path (str): The file path to check
+
+ Returns:
+ bool: True if the path is protected, False otherwise
+ """
+ if not file_path:
+ return False
+
+ # List of protected directory prefixes
+ protected_dirs = [
+ '/data/epgs', # EPG files mapped from host
+ '/data/uploads', # User uploaded files
+ '/data/m3us' # M3U files mapped from host
+ ]
+
+ # Check if the path starts with any protected directory
+ for protected_dir in protected_dirs:
+ if file_path.startswith(protected_dir):
+ return True
+
+ return False
diff --git a/frontend/src/pages/Guide.jsx b/frontend/src/pages/Guide.jsx
index b64c2517..c0042c31 100644
--- a/frontend/src/pages/Guide.jsx
+++ b/frontend/src/pages/Guide.jsx
@@ -78,31 +78,14 @@ export default function TVChannelGuide({ startDate, endDate }) {
const fetched = await API.getGrid(); // GETs your EPG grid
console.log(`Received ${fetched.length} programs`);
- // Unique tvg_ids from returned programs
- const programIds = [...new Set(fetched.map((p) => p.tvg_id))];
+ // Include ALL channels, sorted by channel number - don't filter by EPG data
+ const sortedChannels = Object.values(channels)
+ .sort((a, b) => (a.channel_number || Infinity) - (b.channel_number || Infinity));
- // Filter your Redux/Zustand channels by matching tvg_id
- const filteredChannels = Object.values(channels)
- // Include channels with matching tvg_ids OR channels with null epg_data
- .filter(
- (ch) =>
- programIds.includes(tvgsById[ch.epg_data_id]?.tvg_id) ||
- programIds.includes(ch.uuid) ||
- ch.epg_data_id === null
- )
- // Add sorting by channel_number
- .sort(
- (a, b) =>
- (a.channel_number || Infinity) - (b.channel_number || Infinity)
- );
+ console.log(`Using all ${sortedChannels.length} available channels`);
- console.log(
- `found ${filteredChannels.length} channels with matching tvg_ids`
- );
-
- setGuideChannels(filteredChannels);
- setFilteredChannels(filteredChannels); // Initialize filtered channels
- console.log(fetched);
+ setGuideChannels(sortedChannels);
+ setFilteredChannels(sortedChannels); // Initialize filtered channels
setPrograms(fetched);
setLoading(false);
};
@@ -135,9 +118,12 @@ export default function TVChannelGuide({ startDate, endDate }) {
if (selectedProfileId !== 'all') {
// Get the profile's enabled channels
const profileChannels = profiles[selectedProfileId]?.channels || [];
- const enabledChannelIds = profileChannels
- .filter((pc) => pc.enabled)
- .map((pc) => pc.id);
+ // Check if channels is a Set (from the error message, it likely is)
+ const enabledChannelIds = Array.isArray(profileChannels)
+ ? profileChannels.filter((pc) => pc.enabled).map((pc) => pc.id)
+ : profiles[selectedProfileId]?.channels instanceof Set
+ ? Array.from(profiles[selectedProfileId].channels)
+ : [];
result = result.filter((channel) =>
enabledChannelIds.includes(channel.id)
@@ -1210,13 +1196,51 @@ export default function TVChannelGuide({ startDate, endDate }) {
paddingLeft: 0, // Remove any padding that might push content
}}
>
- {channelPrograms.map((program) => {
- return (
+ {channelPrograms.length > 0 ? (
+ channelPrograms.map((program) => (
{renderProgram(program, start)}
- );
- })}
+ ))
+ ) : (
+ // Simple placeholder for channels with no program data - 2 hour blocks
+ <>
+ {/* Generate repeating placeholder blocks every 2 hours across the timeline */}
+ {Array.from({ length: Math.ceil(hourTimeline.length / 2) }).map((_, index) => (
+
+
+
+ No Program Information Available
+
+
+
+ ))}
+
+ )}
);
diff --git a/version.py b/version.py
index 18a1ef5d..62c02ebc 100644
--- a/version.py
+++ b/version.py
@@ -1,5 +1,5 @@
"""
Dispatcharr version information.
"""
-__version__ = '0.5.0' # Follow semantic versioning (MAJOR.MINOR.PATCH)
+__version__ = '0.5.1' # Follow semantic versioning (MAJOR.MINOR.PATCH)
__timestamp__ = None # Set during CI/CD build process