mirror of
https://github.com/Dispatcharr/Dispatcharr.git
synced 2026-07-20 16:51:10 +00:00
merged in dev
This commit is contained in:
dekzter
commit
6504db3bd4
12 changed files with 655 additions and 269 deletions
18
apps/epg/migrations/0014_epgsource_extracted_file_path.py
Normal file
18
apps/epg/migrations/0014_epgsource_extracted_file_path.py
Normal file
|
|
@ -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),
|
||||
),
|
||||
]
|
||||
|
|
@ -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'<?xml' or header[:5] == b'<tv>':
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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}")
|
||||
|
|
|
|||
|
|
@ -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'<?xml' in header or b'<tv>' 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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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) => (
|
||||
<div key={`${channel.id}-${program.id}-${program.start_time}`}>
|
||||
{renderProgram(program, start)}
|
||||
</div>
|
||||
);
|
||||
})}
|
||||
))
|
||||
) : (
|
||||
// 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) => (
|
||||
<Box
|
||||
key={`placeholder-${channel.id}-${index}`}
|
||||
style={{
|
||||
position: 'absolute',
|
||||
left: index * 2 * HOUR_WIDTH + 2,
|
||||
top: 0,
|
||||
width: 2 * HOUR_WIDTH - 4,
|
||||
height: rowHeight - 4,
|
||||
display: 'flex',
|
||||
alignItems: 'center',
|
||||
justifyContent: 'center',
|
||||
}}
|
||||
>
|
||||
<Paper
|
||||
style={{
|
||||
width: '100%',
|
||||
height: '100%',
|
||||
display: 'flex',
|
||||
alignItems: 'center',
|
||||
justifyContent: 'center',
|
||||
backgroundColor: '#2D3748',
|
||||
color: '#A0AEC0',
|
||||
opacity: 0.8,
|
||||
border: '1px dashed #718096',
|
||||
}}
|
||||
>
|
||||
<Text size="sm" align="center">
|
||||
No Program Information Available
|
||||
</Text>
|
||||
</Paper>
|
||||
</Box>
|
||||
))}
|
||||
</>
|
||||
)}
|
||||
</Box>
|
||||
</Box>
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue