From 759cbe2f7d457cee61374667833c9554c33a2141 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sat, 24 May 2025 15:07:49 -0500 Subject: [PATCH 01/21] Search entire epg for channels to accomodate non standard epg files. --- apps/epg/tasks.py | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index f3f281e7..088016f3 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -632,17 +632,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 @@ -818,8 +807,6 @@ def parse_channels_only(source): # Reset parser state del channel_parser channel_parser = None - gc.collect() - # Perform thorough cleanup cleanup_memory(log_usage=should_log_memory, force_collection=True) @@ -833,6 +820,20 @@ def parse_channels_only(source): 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: + # Just clear the element to avoid memory leaks + if elem is not None: + 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 + continue except (etree.XMLSyntaxError, Exception) as xml_error: logger.error(f"[parse_channels_only] XML parsing failed: {xml_error}") From 3f17c90a8b92a79c41dd1b17f88a6efed2f2e396 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sat, 24 May 2025 17:31:52 -0500 Subject: [PATCH 02/21] Cleaned up some old unneeded code. --- apps/epg/tasks.py | 20 +++----------------- 1 file changed, 3 insertions(+), 17 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 088016f3..80bd18df 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -588,27 +588,13 @@ def parse_channels_only(source): logger.info(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") @@ -616,10 +602,10 @@ def parse_channels_only(source): 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}") + logger.debug(f"Opening file for channel parsing: {file_path}") source_file = gzip.open(file_path, 'rb') if is_gzipped else 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") From f77b3b975675efa98f7b86fc71d7f27d61e9b064 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sat, 24 May 2025 17:40:45 -0500 Subject: [PATCH 03/21] Little more cleanup. --- apps/epg/tasks.py | 58 ++--------------------------------------------- 1 file changed, 2 insertions(+), 56 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 80bd18df..26fc215b 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -732,7 +732,7 @@ def parse_channels_only(source): except (ValueError, KeyError, TypeError): # Element might already be removed or detached pass - cleanup_memory(log_usage=should_log_memory, force_collection=True) + #cleanup_memory(log_usage=should_log_memory, force_collection=True) except Exception as e: # Just log the error and continue - don't let cleanup errors stop processing @@ -746,60 +746,6 @@ def parse_channels_only(source): 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] 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 - # 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: @@ -860,7 +806,7 @@ def parse_channels_only(source): send_websocket_update('updates', 'update', {"success": True, "type": "epg_channels"}) logger.info(f"Finished parsing channel info. Found {processed_channels} channels.") - + time.sleep(10) return True except FileNotFoundError: From 6c3102a60cbf6934e29d5274e0ef96be8d023b2b Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sat, 24 May 2025 17:46:17 -0500 Subject: [PATCH 04/21] Removed sleep from debugging. --- apps/epg/tasks.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 26fc215b..9a3d76c1 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -806,7 +806,7 @@ def parse_channels_only(source): send_websocket_update('updates', 'update', {"success": True, "type": "epg_channels"}) logger.info(f"Finished parsing channel info. Found {processed_channels} channels.") - time.sleep(10) + return True except FileNotFoundError: From d7023bcdac7f9eb4f4401cb3bbe4e267c5f7aafe Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 13:19:49 -0500 Subject: [PATCH 05/21] Some more code cleanup --- apps/epg/tasks.py | 120 ++++++++++++++-------------------------------- 1 file changed, 36 insertions(+), 84 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 9a3d76c1..e7bd06ad 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -585,7 +585,7 @@ 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: # Attempt to count existing channels in the database @@ -648,6 +648,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 @@ -657,6 +658,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( @@ -676,7 +680,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") @@ -689,13 +693,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") @@ -709,62 +713,27 @@ 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} - {display_name} - processed {processed_channels - total_channels} additional channels") + else: + logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - {display_name} - 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}") - 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: - # Just clear the element to avoid memory leaks - if elem is not None: - 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 + clear_element(elem) continue except (etree.XMLSyntaxError, Exception) as xml_error: @@ -776,8 +745,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) @@ -826,8 +796,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 @@ -1053,41 +1024,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() @@ -1502,3 +1442,15 @@ 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) From 391a1d9707aa6543cd08f32b167573ce4cbc1ca9 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 13:29:39 -0500 Subject: [PATCH 06/21] A little more cleanup. --- apps/epg/tasks.py | 18 ++---------------- 1 file changed, 2 insertions(+), 16 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index e7bd06ad..b8ae7fdf 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -995,22 +995,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) From c1eb3a6ecfbc27ace1a3d91ea2937cffe08c27e9 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 14:05:59 -0500 Subject: [PATCH 07/21] Add Zip file support for EPG. --- apps/epg/tasks.py | 51 +++++++++++++++++++++++++++++++++++++---------- core/tasks.py | 6 +++--- 2 files changed, 43 insertions(+), 14 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index b8ae7fdf..40d168b1 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 @@ -546,8 +547,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 @@ -575,6 +574,7 @@ def parse_channels_only(source): # Stream parsing instead of loading entire file at once is_gzipped = file_path.endswith('.gz') + is_zipped = file_path.endswith('.zip') epgs_to_create = [] epgs_to_update = [] @@ -603,7 +603,19 @@ def parse_channels_only(source): # Reset file position for actual processing logger.debug(f"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 based on its type + if is_gzipped: + source_file = gzip.open(file_path, 'rb') + elif is_zipped: + # For ZIP files, need to open the first file in the archive + zip_archive = zipfile.ZipFile(file_path, 'r') + # Use the first file in the archive + first_file = zip_archive.namelist()[0] + source_file = zip_archive.open(first_file, 'r') + else: + source_file = open(file_path, 'rb') + if process: logger.debug(f"[parse_channels_only] Memory after opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB") @@ -736,6 +748,14 @@ def parse_channels_only(source): clear_element(elem) continue + except zipfile.BadZipFile as zip_error: + logger.error(f"Bad ZIP file: {zip_error}") + # Update status to error + source.status = 'error' + source.last_message = f"Error parsing ZIP file: {str(zip_error)}" + source.save(update_fields=['status', 'last_message']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(zip_error)) + return False except (etree.XMLSyntaxError, Exception) as xml_error: logger.error(f"[parse_channels_only] XML parsing failed: {xml_error}") # Update status to error @@ -847,16 +867,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 @@ -929,6 +947,7 @@ def parse_programs_for_tvg_id(epg_id): # Use streaming parsing to reduce memory usage is_gzipped = file_path.endswith('.gz') + is_zipped = file_path.endswith('.zip') logger.info(f"Parsing programs for tvg_id={epg.tvg_id} from {file_path}") @@ -946,8 +965,17 @@ def parse_programs_for_tvg_id(epg_id): 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 based on its type + if is_gzipped: + source_file = gzip.open(file_path, 'rb') + elif is_zipped: + # For ZIP files, need to open the first file in the archive + zip_archive = zipfile.ZipFile(file_path, 'r') + # Use the first file in the archive + first_file = zip_archive.namelist()[0] + source_file = zip_archive.open(first_file, 'r') + else: + source_file = open(file_path, 'rb') # Stream parse the file using lxml's iterparse program_parser = etree.iterparse(source_file, events=('end',), tag='programme') @@ -1024,6 +1052,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 @@ -1105,8 +1136,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 diff --git a/core/tasks.py b/core/tasks.py index eee5b18f..c2297569 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 From 1ab04e31a440a40a475b0cef8aff6a208ffea729 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 14:52:05 -0500 Subject: [PATCH 08/21] Show channels without epg data. --- frontend/src/pages/Guide.jsx | 75 +++++++++++++++++++++++------------- 1 file changed, 48 insertions(+), 27 deletions(-) diff --git a/frontend/src/pages/Guide.jsx b/frontend/src/pages/Guide.jsx index b64c2517..2331f2b5 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); }; @@ -1210,13 +1193,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 + + + + ))} + + )} ); From 363a1a80805c63be40dbfeb557b4f9a617a72060 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 15:01:04 -0500 Subject: [PATCH 09/21] Fixes selecting a profile leads to webui crash. --- frontend/src/pages/Guide.jsx | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/frontend/src/pages/Guide.jsx b/frontend/src/pages/Guide.jsx index 2331f2b5..c0042c31 100644 --- a/frontend/src/pages/Guide.jsx +++ b/frontend/src/pages/Guide.jsx @@ -118,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) From 0322a5c904186d3e4eba640b028bc7e8219ea3d8 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 15:47:22 -0500 Subject: [PATCH 10/21] Smarter logging for file type. --- apps/epg/tasks.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 40d168b1..3e27c6af 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -601,19 +601,19 @@ def parse_channels_only(source): # 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"Opening file for channel parsing: {file_path}") - # Open the file based on its type if is_gzipped: + logger.debug(f"Opening gzipped file for channel parsing: {file_path}") source_file = gzip.open(file_path, 'rb') elif is_zipped: + logger.debug(f"Opening zipped file for channel parsing: {file_path}") # For ZIP files, need to open the first file in the archive zip_archive = zipfile.ZipFile(file_path, 'r') # Use the first file in the archive first_file = zip_archive.namelist()[0] source_file = zip_archive.open(first_file, 'r') else: + logger.debug(f"Opening file for channel parsing: {file_path}") source_file = open(file_path, 'rb') if process: @@ -967,8 +967,10 @@ def parse_programs_for_tvg_id(epg_id): try: # Open the file based on its type if is_gzipped: + logger.debug(f"Opening GZ file for parsing: {file_path}") source_file = gzip.open(file_path, 'rb') elif is_zipped: + logger.debug(f"Opening ZIP file for parsing: {file_path}") # For ZIP files, need to open the first file in the archive zip_archive = zipfile.ZipFile(file_path, 'r') # Use the first file in the archive From d270e988bd3facc2fc75be06d9151d2ef42b1407 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 16:50:03 -0500 Subject: [PATCH 11/21] Greatly improve filetype detection. --- apps/epg/models.py | 42 ++++++++++++++++++++++++++++++++++++++++-- apps/epg/tasks.py | 42 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 81 insertions(+), 3 deletions(-) diff --git a/apps/epg/models.py b/apps/epg/models.py index dce4e21b..27e85e29 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -59,8 +59,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/tasks.py b/apps/epg/tasks.py index 3e27c6af..03cb5e41 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -372,6 +372,43 @@ def fetch_xmltv(source): # Send completion notification send_epg_update(source.id, "downloading", 100) + # Determine the appropriate file extension based on content type or URL + content_type = response.headers.get('Content-Type', '').lower() + original_url = source.url.lower() + + # Default extension is xml + file_extension = '.xml' + + # Check content type first + if 'application/x-gzip' in content_type or 'application/gzip' in content_type: + file_extension = '.gz' + elif 'application/zip' in content_type: + file_extension = '.zip' + # If content type didn't help, try the URL + elif original_url.endswith('.gz'): + file_extension = '.gz' + elif original_url.endswith('.zip'): + file_extension = '.zip' + + # Update the cache file path with the correct extension + base_cache_path = os.path.splitext(cache_file)[0] + new_cache_file = f"{base_cache_path}{file_extension}" + + # Actually rename the file on disk + logger.debug(f"Renaming file from {cache_file} to {new_cache_file}") + try: + os.rename(cache_file, new_cache_file) + cache_file = new_cache_file # Update the variable to point to the renamed file + except OSError as e: + logger.error(f"Failed to rename temporary file: {e}") + # If rename fails, keep using the original file + logger.warning(f"Continuing with temporary file: {cache_file}") + new_cache_file = cache_file # Fall back to the tmp file + + # Update the source's file_path to reflect the correct extension + source.file_path = new_cache_file + source.save(update_fields=['file_path', 'status']) + # Update status to parsing source.status = 'parsing' source.save(update_fields=['status']) @@ -490,6 +527,9 @@ def parse_channels_only(source): # 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): @@ -860,6 +900,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: @@ -962,7 +1003,6 @@ def parse_programs_for_tvg_id(epg_id): programs_to_create = [] batch_size = 1000 # Process in batches to limit memory usage - programs_processed = 0 try: # Open the file based on its type From 7dbd41afa89f0641e4d5fa18ce71cdc8f864d244 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 17:39:41 -0500 Subject: [PATCH 12/21] Extract compressed files after downloading and delete original. --- apps/epg/tasks.py | 190 +++++++++++++++++++++++++++++++++------------- 1 file changed, 139 insertions(+), 51 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 03cb5e41..371e6c2d 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -208,6 +208,19 @@ 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: + extracted_path = extract_compressed_file(source.file_path) + if extracted_path: + logger.info(f"Extracted existing compressed file to: {extracted_path}") + source.file_path = extracted_path + source.save(update_fields=['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']) @@ -405,15 +418,42 @@ def fetch_xmltv(source): logger.warning(f"Continuing with temporary file: {cache_file}") new_cache_file = cache_file # Fall back to the tmp file - # Update the source's file_path to reflect the correct extension - source.file_path = new_cache_file + # Now extract the file if it's compressed + extracted_file = None + if file_extension in ('.gz', '.zip'): + try: + logger.info(f"Extracting compressed file {new_cache_file}") + send_epg_update(source.id, "extracting", 0, message="Extracting downloaded file") + + extracted_file = extract_compressed_file(new_cache_file) + + if extracted_file: + logger.info(f"Successfully extracted to {extracted_file}") + send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully: {os.path.basename(extracted_file)}") + # Update the source's file_path to the extracted XML file + source.file_path = extracted_file + else: + logger.error("Extraction failed, using compressed file") + send_epg_update(source.id, "extracting", 100, status="error", message="Extraction failed, using compressed file") + # Use the compressed file + source.file_path = new_cache_file + except Exception as e: + logger.error(f"Error extracting file: {str(e)}", exc_info=True) + send_epg_update(source.id, "extracting", 100, status="error", message=f"Error during extraction: {str(e)}") + # Use the compressed file if extraction fails + source.file_path = new_cache_file + else: + # It's already an XML file + source.file_path = new_cache_file + + # Update the source's file_path to reflect the correct file source.save(update_fields=['file_path', 'status']) # 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: @@ -519,6 +559,83 @@ def fetch_xmltv(source): return False +def extract_compressed_file(file_path, delete_original=True): + """ + Extracts a compressed file (.gz or .zip) to an XML file. + + Args: + file_path: Path to the compressed file + delete_original: Whether to delete the original compressed file after successful extraction + + Returns: + Path to the extracted XML file, or None if extraction failed + """ + try: + base_path = os.path.splitext(file_path)[0] + extracted_path = f"{base_path}.xml" + + # Make sure the output path doesn't already exist + if os.path.exists(extracted_path): + try: + os.remove(extracted_path) + except Exception as e: + logger.warning(f"Failed to remove existing extracted file: {e}") + # Try with a unique filename instead + extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml" + + if file_path.endswith('.gz'): + logger.debug(f"Extracting gzip file: {file_path}") + with gzip.open(file_path, 'rb') as gz_file: + with open(extracted_path, 'wb') as out_file: + out_file.write(gz_file.read()) + logger.info(f"Successfully extracted gzip file to: {extracted_path}") + + # 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 file_path.endswith('.zip'): + logger.debug(f"Extracting zip file: {file_path}") + with zipfile.ZipFile(file_path, 'r') as zip_file: + # Find the first XML file in the ZIP archive + xml_files = [f for f in zip_file.namelist() if f.lower().endswith('.xml')] + + if not xml_files: + logger.error("No XML file found in ZIP archive") + return None + + # Extract the first XML file + xml_content = zip_file.read(xml_files[0]) + with open(extracted_path, 'wb') as out_file: + out_file.write(xml_content) + + logger.info(f"Successfully extracted zip file to: {extracted_path}") + + # 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 compressed file format: {file_path}") + return None + + except Exception as e: + logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True) + return None + + def parse_channels_only(source): file_path = source.file_path if not file_path: @@ -558,14 +675,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 @@ -576,7 +696,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 @@ -613,9 +733,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') - is_zipped = file_path.endswith('.zip') - + # This can be simplified since we now always have XML files epgs_to_create = [] epgs_to_update = [] total_channels = 0 @@ -641,20 +759,9 @@ def parse_channels_only(source): # Update progress after counting send_epg_update(source.id, "parsing_channels", 25, total_channels=total_channels) - # Open the file based on its type - if is_gzipped: - logger.debug(f"Opening gzipped file for channel parsing: {file_path}") - source_file = gzip.open(file_path, 'rb') - elif is_zipped: - logger.debug(f"Opening zipped file for channel parsing: {file_path}") - # For ZIP files, need to open the first file in the archive - zip_archive = zipfile.ZipFile(file_path, 'r') - # Use the first file in the archive - first_file = zip_archive.namelist()[0] - source_file = zip_archive.open(first_file, 'r') - else: - logger.debug(f"Opening file for channel parsing: {file_path}") - source_file = 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 opening file: {process.memory_info().rss / 1024 / 1024:.2f} MB") @@ -788,14 +895,6 @@ def parse_channels_only(source): clear_element(elem) continue - except zipfile.BadZipFile as zip_error: - logger.error(f"Bad ZIP file: {zip_error}") - # Update status to error - source.status = 'error' - source.last_message = f"Error parsing ZIP file: {str(zip_error)}" - source.save(update_fields=['status', 'last_message']) - send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(zip_error)) - return False except (etree.XMLSyntaxError, Exception) as xml_error: logger.error(f"[parse_channels_only] XML parsing failed: {xml_error}") # Update status to error @@ -966,14 +1065,17 @@ 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 + 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 @@ -984,12 +1086,8 @@ 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') - is_zipped = file_path.endswith('.zip') - + # No need to check file type anymore since it's always XML logger.info(f"Parsing programs for tvg_id={epg.tvg_id} from {file_path}") # Memory usage tracking @@ -1005,19 +1103,9 @@ def parse_programs_for_tvg_id(epg_id): batch_size = 1000 # Process in batches to limit memory usage try: - # Open the file based on its type - if is_gzipped: - logger.debug(f"Opening GZ file for parsing: {file_path}") - source_file = gzip.open(file_path, 'rb') - elif is_zipped: - logger.debug(f"Opening ZIP file for parsing: {file_path}") - # For ZIP files, need to open the first file in the archive - zip_archive = zipfile.ZipFile(file_path, 'r') - # Use the first file in the archive - first_file = zip_archive.namelist()[0] - source_file = zip_archive.open(first_file, 'r') - else: - source_file = 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') From 182a009d696f95355cf07f92cff9b1dab2d0620a Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sun, 25 May 2025 18:14:27 -0500 Subject: [PATCH 13/21] Track extracted files for mapped epg files. --- .../0014_epgsource_original_file_path.py | 18 +++ apps/epg/models.py | 2 + apps/epg/tasks.py | 143 ++++++++++-------- core/tasks.py | 97 +++++++++++- 4 files changed, 195 insertions(+), 65 deletions(-) create mode 100644 apps/epg/migrations/0014_epgsource_original_file_path.py diff --git a/apps/epg/migrations/0014_epgsource_original_file_path.py b/apps/epg/migrations/0014_epgsource_original_file_path.py new file mode 100644 index 00000000..e6c04fec --- /dev/null +++ b/apps/epg/migrations/0014_epgsource_original_file_path.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 on 2025-05-25 23:00 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0013_alter_epgsource_refresh_interval'), + ] + + operations = [ + migrations.AddField( + model_name='epgsource', + name='original_file_path', + field=models.CharField(blank=True, help_text='Original path to compressed file before extraction', max_length=1024, null=True), + ), + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index 27e85e29..8fd4b2f5 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -32,6 +32,8 @@ class EPGSource(models.Model): api_key = models.CharField(max_length=255, blank=True, null=True) # For Schedules Direct is_active = models.BooleanField(default=True) file_path = models.CharField(max_length=1024, blank=True, null=True) + original_file_path = models.CharField(max_length=1024, blank=True, null=True, + help_text="Original path to compressed file before extraction") refresh_interval = models.IntegerField(default=0) refresh_task = models.ForeignKey( PeriodicTask, on_delete=models.SET_NULL, null=True, blank=True diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 371e6c2d..cdf942e8 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -240,13 +240,6 @@ def fetch_xmltv(source): send_epg_update(source.id, "downloading", 100, status="error", error="No URL provided and no valid local file exists") return False - # Clean up existing cache file - if os.path.exists(source.get_cache_file()): - try: - os.remove(source.get_cache_file()) - except Exception as e: - logger.warning(f"Failed to remove existing cache file: {e}") - logger.info(f"Fetching XMLTV data from source: {source.name}") try: # Get default user agent from settings @@ -335,7 +328,16 @@ def fetch_xmltv(source): response.raise_for_status() logger.debug("XMLTV data fetched successfully.") - cache_file = source.get_cache_file() + # Define base paths for consistent file naming + cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg") + os.makedirs(cache_dir, exist_ok=True) + + # Create temporary download file with .tmp extension + temp_download_path = os.path.join(cache_dir, f"{source.id}.tmp") + + # Define final paths based on content type + compressed_path = os.path.join(cache_dir, f"{source.id}.compressed"); + xml_path = os.path.join(cache_dir, f"{source.id}.xml"); # Check if we have content length for progress tracking total_size = int(response.headers.get('content-length', 0)) @@ -344,7 +346,8 @@ def fetch_xmltv(source): last_update_time = start_time update_interval = 0.5 # Only update every 0.5 seconds - with open(cache_file, 'wb') as f: + # Download to temporary file + with open(temp_download_path, 'wb') as f: for chunk in response.iter_content(chunk_size=16384): # Increased chunk size for better performance if chunk: f.write(chunk) @@ -389,65 +392,79 @@ def fetch_xmltv(source): content_type = response.headers.get('Content-Type', '').lower() original_url = source.url.lower() - # Default extension is xml - file_extension = '.xml' + # Is this file compressed? + is_compressed = False + if 'application/x-gzip' in content_type or 'application/gzip' in content_type or 'application/zip' in content_type: + is_compressed = True + elif original_url.endswith('.gz') or original_url.endswith('.zip'): + is_compressed = True - # Check content type first - if 'application/x-gzip' in content_type or 'application/gzip' in content_type: - file_extension = '.gz' - elif 'application/zip' in content_type: - file_extension = '.zip' - # If content type didn't help, try the URL - elif original_url.endswith('.gz'): - file_extension = '.gz' - elif original_url.endswith('.zip'): - file_extension = '.zip' + # Clean up old files before saving new ones + if os.path.exists(compressed_path): + try: + os.remove(compressed_path) + logger.debug(f"Removed old compressed file: {compressed_path}") + except OSError as e: + logger.warning(f"Failed to remove old compressed file: {e}") - # Update the cache file path with the correct extension - base_cache_path = os.path.splitext(cache_file)[0] - new_cache_file = f"{base_cache_path}{file_extension}" + if os.path.exists(xml_path): + try: + os.remove(xml_path) + logger.debug(f"Removed old XML file: {xml_path}") + except OSError as e: + logger.warning(f"Failed to remove old XML file: {e}") - # Actually rename the file on disk - logger.debug(f"Renaming file from {cache_file} to {new_cache_file}") - try: - os.rename(cache_file, new_cache_file) - cache_file = new_cache_file # Update the variable to point to the renamed file - except OSError as e: - logger.error(f"Failed to rename temporary file: {e}") - # If rename fails, keep using the original file - logger.warning(f"Continuing with temporary file: {cache_file}") - new_cache_file = cache_file # Fall back to the tmp file + # Rename the temp file to appropriate final path + if is_compressed: + try: + os.rename(temp_download_path, compressed_path) + logger.debug(f"Renamed temp file to compressed file: {compressed_path}") + current_file_path = compressed_path + except OSError as e: + logger.error(f"Failed to rename temp file to compressed file: {e}") + current_file_path = temp_download_path # Fall back to using temp file + else: + try: + os.rename(temp_download_path, xml_path) + logger.debug(f"Renamed temp file to XML file: {xml_path}") + current_file_path = xml_path + except OSError as e: + logger.error(f"Failed to rename temp file to XML file: {e}") + current_file_path = temp_download_path # Fall back to using temp file # Now extract the file if it's compressed - extracted_file = None - if file_extension in ('.gz', '.zip'): + if is_compressed: try: - logger.info(f"Extracting compressed file {new_cache_file}") + logger.info(f"Extracting compressed file {current_file_path}") send_epg_update(source.id, "extracting", 0, message="Extracting downloaded file") - extracted_file = extract_compressed_file(new_cache_file) + # Always extract to the standard XML path + extracted = extract_compressed_file(current_file_path, xml_path) - if extracted_file: - logger.info(f"Successfully extracted to {extracted_file}") - send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully: {os.path.basename(extracted_file)}") + if extracted: + logger.info(f"Successfully extracted to {xml_path}") + send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully") # Update the source's file_path to the extracted XML file - source.file_path = extracted_file + source.file_path = xml_path + + # Store the original compressed file path + source.original_file_path = current_file_path else: logger.error("Extraction failed, using compressed file") send_epg_update(source.id, "extracting", 100, status="error", message="Extraction failed, using compressed file") # Use the compressed file - source.file_path = new_cache_file + source.file_path = current_file_path except Exception as e: logger.error(f"Error extracting file: {str(e)}", exc_info=True) send_epg_update(source.id, "extracting", 100, status="error", message=f"Error during extraction: {str(e)}") # Use the compressed file if extraction fails - source.file_path = new_cache_file + source.file_path = current_file_path else: # It's already an XML file - source.file_path = new_cache_file + source.file_path = current_file_path # Update the source's file_path to reflect the correct file - source.save(update_fields=['file_path', 'status']) + source.save(update_fields=['file_path', 'status', 'original_file_path']) # Update status to parsing source.status = 'parsing' @@ -559,29 +576,37 @@ def fetch_xmltv(source): return False -def extract_compressed_file(file_path, delete_original=True): +def extract_compressed_file(file_path, output_path=None, delete_original=False): """ Extracts a compressed file (.gz or .zip) to an XML file. Args: file_path: Path to the compressed file + output_path: Specific path where the file should be extracted (optional) delete_original: Whether to delete the original compressed file after successful extraction Returns: Path to the extracted XML file, or None if extraction failed """ try: - base_path = os.path.splitext(file_path)[0] - extracted_path = f"{base_path}.xml" + if output_path is None: + base_path = os.path.splitext(file_path)[0] + extracted_path = f"{base_path}.xml" + else: + extracted_path = output_path # Make sure the output path doesn't already exist if os.path.exists(extracted_path): try: os.remove(extracted_path) + logger.info(f"Removed existing extracted file: {extracted_path}") except Exception as e: logger.warning(f"Failed to remove existing extracted file: {e}") - # Try with a unique filename instead - extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml" + # If we can't delete the existing file and no specific output was requested, + # create a unique filename instead + if output_path is None: + base_path = os.path.splitext(file_path)[0] + extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml" if file_path.endswith('.gz'): logger.debug(f"Extracting gzip file: {file_path}") @@ -1588,14 +1613,14 @@ def extract_custom_properties(prog): return custom_props -def clear_element(elem): - """Clear an XML element and its parent to free memory.""" - try: - elem.clear() - parent = elem.getparent() - if parent is not None: - while elem.getprevious() is not None: - del parent[0] + + + + + + + + while elem.getprevious() is not None: if parent is not None: parent = elem.getparent() elem.clear() try: """Clear an XML element and its parent to free memory."""def clear_element(elem): del parent[0] parent.remove(elem) except Exception as e: logger.warning(f"Error clearing XML element: {e}", exc_info=True) diff --git a/core/tasks.py b/core/tasks.py index c2297569..062ffaf2 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -7,6 +7,9 @@ import logging import re import time import os +import gzip +import zipfile +import uuid from core.utils import RedisClient, send_websocket_update from apps.proxy.ts_proxy.channel_status import ChannelStatus from apps.m3u.models import M3UAccount @@ -16,11 +19,13 @@ from apps.epg.tasks import refresh_epg_data from .models import CoreSettings from apps.channels.models import Stream, ChannelStream from django.db import transaction +from django.conf import settings logger = logging.getLogger(__name__) EPG_WATCH_DIR = '/data/epgs' M3U_WATCH_DIR = '/data/m3us' +EPG_CACHE_DIR = os.path.join(settings.MEDIA_ROOT, "cached_epg") MIN_AGE_SECONDS = 6 STARTUP_SKIP_AGE = 30 REDIS_PREFIX = "processed_file:" @@ -46,6 +51,58 @@ def throttled_log(logger_method, message, key=None, *args, **kwargs): logger_method(message, *args, **kwargs) _last_log_times[key] = now +def extract_compressed_file(file_path): + """ + Extracts a compressed file (.gz or .zip) to an XML file in the cache directory. + + Args: + file_path: Path to the compressed file + + Returns: + Path to the extracted XML file, or None if extraction failed + """ + try: + # Create cache directory if it doesn't exist + os.makedirs(EPG_CACHE_DIR, exist_ok=True) + + # Generate a unique filename for the extracted file + extracted_filename = f"extracted_{uuid.uuid4().hex[:8]}.xml" + extracted_path = os.path.join(EPG_CACHE_DIR, extracted_filename) + + if file_path.endswith('.gz'): + logger.debug(f"Extracting gzip file: {file_path}") + with gzip.open(file_path, 'rb') as gz_file: + with open(extracted_path, 'wb') as out_file: + out_file.write(gz_file.read()) + logger.info(f"Successfully extracted gzip file to: {extracted_path}") + return extracted_path + + elif file_path.endswith('.zip'): + logger.debug(f"Extracting zip file: {file_path}") + with zipfile.ZipFile(file_path, 'r') as zip_file: + # Find the first XML file in the ZIP archive + xml_files = [f for f in zip_file.namelist() if f.lower().endswith('.xml')] + + if not xml_files: + logger.error("No XML file found in ZIP archive") + return None + + # Extract the first XML file + xml_content = zip_file.read(xml_files[0]) + with open(extracted_path, 'wb') as out_file: + out_file.write(xml_content) + + logger.info(f"Successfully extracted zip file to: {extracted_path}") + return extracted_path + + else: + logger.error(f"Unsupported compressed file format: {file_path}") + return None + + except Exception as e: + logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True) + return None + @shared_task def beat_periodic_task(): fetch_channel_stats() @@ -192,7 +249,7 @@ def scan_and_process_files(): # Instead of assuming old files were processed, check if they exist in the database if not stored_mtime and age > STARTUP_SKIP_AGE: # Check if this file is already in the database - existing_epg = EPGSource.objects.filter(file_path=filepath).exists() + existing_epg = EPGSource.objects.filter(original_file_path=filepath).exists() if existing_epg: # Use trace level if not first scan if _first_scan_completed: @@ -227,11 +284,39 @@ def scan_and_process_files(): continue try: - epg_source, created = EPGSource.objects.get_or_create(file_path=filepath, defaults={ - "name": filename, - "source_type": "xmltv", - "is_active": CoreSettings.get_auto_import_mapped_files() in [True, "true", "True"], - }) + extracted_path = None + is_compressed = filename.endswith('.gz') or filename.endswith('.zip') + + # Extract compressed files + if is_compressed: + logger.info(f"Detected compressed EPG file: {filename}, extracting") + extracted_path = extract_compressed_file(filepath) + + if not extracted_path: + logger.error(f"Failed to extract compressed file: {filename}") + epg_errors += 1 + continue + + logger.info(f"Successfully extracted {filename} to {extracted_path}") + + # Set the file path to use (either extracted or original) + file_to_use = extracted_path if extracted_path else filepath + + epg_source, created = EPGSource.objects.get_or_create( + original_file_path=filepath, + defaults={ + "name": filename, + "source_type": "xmltv", + "is_active": CoreSettings.get_auto_import_mapped_files() in [True, "true", "True"], + "file_path": file_to_use + } + ) + + # If the source already exists but we extracted a new file, update its file_path + if not created and extracted_path: + epg_source.file_path = extracted_path + epg_source.save(update_fields=['file_path']) + logger.info(f"Updated existing EPG source with new extracted file: {extracted_path}") redis_client.set(redis_key, mtime, ex=REDIS_TTL) From 8f4e05b0b82e08a7ccbfc917ce4a5f703dda8594 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 26 May 2025 15:10:54 -0500 Subject: [PATCH 14/21] Add extracted_file_path to EPGSource model and update extraction logic --- ... => 0014_epgsource_extracted_file_path.py} | 6 +- apps/epg/models.py | 4 +- apps/epg/tasks.py | 168 +++++++++++++----- core/tasks.py | 101 +---------- 4 files changed, 140 insertions(+), 139 deletions(-) rename apps/epg/migrations/{0014_epgsource_original_file_path.py => 0014_epgsource_extracted_file_path.py} (54%) diff --git a/apps/epg/migrations/0014_epgsource_original_file_path.py b/apps/epg/migrations/0014_epgsource_extracted_file_path.py similarity index 54% rename from apps/epg/migrations/0014_epgsource_original_file_path.py rename to apps/epg/migrations/0014_epgsource_extracted_file_path.py index e6c04fec..9ee1170b 100644 --- a/apps/epg/migrations/0014_epgsource_original_file_path.py +++ b/apps/epg/migrations/0014_epgsource_extracted_file_path.py @@ -1,4 +1,4 @@ -# Generated by Django 5.1.6 on 2025-05-25 23:00 +# Generated by Django 5.1.6 on 2025-05-26 15:48 from django.db import migrations, models @@ -12,7 +12,7 @@ class Migration(migrations.Migration): operations = [ migrations.AddField( model_name='epgsource', - name='original_file_path', - field=models.CharField(blank=True, help_text='Original path to compressed file before extraction', max_length=1024, null=True), + 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 8fd4b2f5..8abfb26f 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -32,8 +32,8 @@ class EPGSource(models.Model): api_key = models.CharField(max_length=255, blank=True, null=True) # For Schedules Direct is_active = models.BooleanField(default=True) file_path = models.CharField(max_length=1024, blank=True, null=True) - original_file_path = models.CharField(max_length=1024, blank=True, null=True, - help_text="Original path to compressed file before extraction") + 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 diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index cdf942e8..5238e9d3 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -212,11 +212,21 @@ def fetch_xmltv(source): # 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: - extracted_path = extract_compressed_file(source.file_path) + # 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 existing compressed file to: {extracted_path}") - source.file_path = extracted_path - source.save(update_fields=['file_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 @@ -331,13 +341,9 @@ def fetch_xmltv(source): # Define base paths for consistent file naming cache_dir = os.path.join(settings.MEDIA_ROOT, "cached_epg") os.makedirs(cache_dir, exist_ok=True) - + # Create temporary download file with .tmp extension temp_download_path = os.path.join(cache_dir, f"{source.id}.tmp") - - # Define final paths based on content type - compressed_path = os.path.join(cache_dir, f"{source.id}.compressed"); - xml_path = os.path.join(cache_dir, f"{source.id}.xml"); # Check if we have content length for progress tracking total_size = int(response.headers.get('content-length', 0)) @@ -388,16 +394,21 @@ def fetch_xmltv(source): # Send completion notification send_epg_update(source.id, "downloading", 100) - # Determine the appropriate file extension based on content type or URL - content_type = response.headers.get('Content-Type', '').lower() - original_url = source.url.lower() + # 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 - # Is this file compressed? - is_compressed = False - if 'application/x-gzip' in content_type or 'application/gzip' in content_type or 'application/zip' in content_type: - is_compressed = True - elif original_url.endswith('.gz') or original_url.endswith('.zip'): - is_compressed = True + # 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): @@ -439,32 +450,33 @@ def fetch_xmltv(source): send_epg_update(source.id, "extracting", 0, message="Extracting downloaded file") # Always extract to the standard XML path - extracted = extract_compressed_file(current_file_path, xml_path) + extracted = extract_compressed_file(current_file_path, xml_path, delete_original=False) if extracted: logger.info(f"Successfully extracted to {xml_path}") send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully") - # Update the source's file_path to the extracted XML file - source.file_path = xml_path - - # Store the original compressed file path - source.original_file_path = current_file_path + # Update to store both paths properly + source.file_path = current_file_path + source.extracted_file_path = xml_path else: logger.error("Extraction failed, using compressed file") send_epg_update(source.id, "extracting", 100, status="error", message="Extraction failed, using compressed file") # Use the compressed file source.file_path = 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_path to reflect the correct file - source.save(update_fields=['file_path', 'status', 'original_file_path']) + # Update the source's file paths + source.save(update_fields=['file_path', 'status', 'extracted_file_path']) # Update status to parsing source.status = 'parsing' @@ -608,7 +620,13 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): base_path = os.path.splitext(file_path)[0] extracted_path = f"{base_path}_{uuid.uuid4().hex[:8]}.xml" - if file_path.endswith('.gz'): + # 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}") with gzip.open(file_path, 'rb') as gz_file: with open(extracted_path, 'wb') as out_file: @@ -625,7 +643,7 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): return extracted_path - elif file_path.endswith('.zip'): + 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 @@ -653,7 +671,7 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): return extracted_path else: - logger.error(f"Unsupported compressed file format: {file_path}") + logger.error(f"Unsupported or unrecognized compressed file format: {file_path} (detected as: {format_type})") return None except Exception as e: @@ -662,7 +680,8 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): 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() @@ -1058,7 +1077,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() @@ -1113,13 +1132,13 @@ def parse_programs_for_tvg_id(epg_id): # Use streaming parsing to reduce memory usage # No need to check file type anymore since it's always XML - logger.info(f"Parsing programs for tvg_id={epg.tvg_id} from {file_path}") + 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 @@ -1613,14 +1632,81 @@ def extract_custom_properties(prog): return custom_props - - - - - - - - while elem.getprevious() is not None: if parent is not None: parent = elem.getparent() elem.clear() try: """Clear an XML element and its parent to free memory."""def clear_element(elem): del parent[0] +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/core/tasks.py b/core/tasks.py index 062ffaf2..0fdaedf7 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -7,9 +7,6 @@ import logging import re import time import os -import gzip -import zipfile -import uuid from core.utils import RedisClient, send_websocket_update from apps.proxy.ts_proxy.channel_status import ChannelStatus from apps.m3u.models import M3UAccount @@ -19,13 +16,11 @@ from apps.epg.tasks import refresh_epg_data from .models import CoreSettings from apps.channels.models import Stream, ChannelStream from django.db import transaction -from django.conf import settings logger = logging.getLogger(__name__) EPG_WATCH_DIR = '/data/epgs' M3U_WATCH_DIR = '/data/m3us' -EPG_CACHE_DIR = os.path.join(settings.MEDIA_ROOT, "cached_epg") MIN_AGE_SECONDS = 6 STARTUP_SKIP_AGE = 30 REDIS_PREFIX = "processed_file:" @@ -51,58 +46,6 @@ def throttled_log(logger_method, message, key=None, *args, **kwargs): logger_method(message, *args, **kwargs) _last_log_times[key] = now -def extract_compressed_file(file_path): - """ - Extracts a compressed file (.gz or .zip) to an XML file in the cache directory. - - Args: - file_path: Path to the compressed file - - Returns: - Path to the extracted XML file, or None if extraction failed - """ - try: - # Create cache directory if it doesn't exist - os.makedirs(EPG_CACHE_DIR, exist_ok=True) - - # Generate a unique filename for the extracted file - extracted_filename = f"extracted_{uuid.uuid4().hex[:8]}.xml" - extracted_path = os.path.join(EPG_CACHE_DIR, extracted_filename) - - if file_path.endswith('.gz'): - logger.debug(f"Extracting gzip file: {file_path}") - with gzip.open(file_path, 'rb') as gz_file: - with open(extracted_path, 'wb') as out_file: - out_file.write(gz_file.read()) - logger.info(f"Successfully extracted gzip file to: {extracted_path}") - return extracted_path - - elif file_path.endswith('.zip'): - logger.debug(f"Extracting zip file: {file_path}") - with zipfile.ZipFile(file_path, 'r') as zip_file: - # Find the first XML file in the ZIP archive - xml_files = [f for f in zip_file.namelist() if f.lower().endswith('.xml')] - - if not xml_files: - logger.error("No XML file found in ZIP archive") - return None - - # Extract the first XML file - xml_content = zip_file.read(xml_files[0]) - with open(extracted_path, 'wb') as out_file: - out_file.write(xml_content) - - logger.info(f"Successfully extracted zip file to: {extracted_path}") - return extracted_path - - else: - logger.error(f"Unsupported compressed file format: {file_path}") - return None - - except Exception as e: - logger.error(f"Error extracting {file_path}: {str(e)}", exc_info=True) - return None - @shared_task def beat_periodic_task(): fetch_channel_stats() @@ -235,9 +178,9 @@ def scan_and_process_files(): 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, GZ, or ZIP file") + logger.trace(f"Skipping {filename}: Not an XML, GZ or zip file") else: - logger.debug(f"Skipping {filename}: Not an XML, GZ, or ZIP file") + logger.debug(f"Skipping {filename}: Not an XML, GZ or zip file") epg_skipped += 1 continue @@ -249,7 +192,7 @@ def scan_and_process_files(): # Instead of assuming old files were processed, check if they exist in the database if not stored_mtime and age > STARTUP_SKIP_AGE: # Check if this file is already in the database - existing_epg = EPGSource.objects.filter(original_file_path=filepath).exists() + existing_epg = EPGSource.objects.filter(file_path=filepath).exists() if existing_epg: # Use trace level if not first scan if _first_scan_completed: @@ -284,39 +227,11 @@ def scan_and_process_files(): continue try: - extracted_path = None - is_compressed = filename.endswith('.gz') or filename.endswith('.zip') - - # Extract compressed files - if is_compressed: - logger.info(f"Detected compressed EPG file: {filename}, extracting") - extracted_path = extract_compressed_file(filepath) - - if not extracted_path: - logger.error(f"Failed to extract compressed file: {filename}") - epg_errors += 1 - continue - - logger.info(f"Successfully extracted {filename} to {extracted_path}") - - # Set the file path to use (either extracted or original) - file_to_use = extracted_path if extracted_path else filepath - - epg_source, created = EPGSource.objects.get_or_create( - original_file_path=filepath, - defaults={ - "name": filename, - "source_type": "xmltv", - "is_active": CoreSettings.get_auto_import_mapped_files() in [True, "true", "True"], - "file_path": file_to_use - } - ) - - # If the source already exists but we extracted a new file, update its file_path - if not created and extracted_path: - epg_source.file_path = extracted_path - epg_source.save(update_fields=['file_path']) - logger.info(f"Updated existing EPG source with new extracted file: {extracted_path}") + epg_source, created = EPGSource.objects.get_or_create(file_path=filepath, defaults={ + "name": filename, + "source_type": "xmltv", + "is_active": CoreSettings.get_auto_import_mapped_files() in [True, "true", "True"], + }) redis_client.set(redis_key, mtime, ex=REDIS_TTL) From 5d2c604a4a884b88a90509bfc032aaf836ccadea Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 26 May 2025 15:49:51 -0500 Subject: [PATCH 15/21] Peak inside gz files to see if they contain xml files. Check file list for zip files to see if they contain xml files. --- apps/epg/tasks.py | 53 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 42 insertions(+), 11 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 5238e9d3..3d903193 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -449,15 +449,15 @@ def fetch_xmltv(source): 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 - extracted = extract_compressed_file(current_file_path, xml_path, delete_original=False) + # 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}") - send_epg_update(source.id, "extracting", 100, message=f"File extracted successfully") - # Update to store both paths properly - source.file_path = current_file_path - source.extracted_file_path = xml_path + 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") @@ -628,11 +628,26 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): if format_type == 'gzip': logger.debug(f"Extracting gzip file: {file_path}") - with gzip.open(file_path, 'rb') as gz_file: - with open(extracted_path, 'wb') as out_file: - out_file.write(gz_file.read()) + 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: @@ -649,6 +664,22 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): # 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 From 45239b744c981072bed145ef0c07b58612eeef00 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Mon, 26 May 2025 16:19:57 -0500 Subject: [PATCH 16/21] Delete cached files when deleting epg account. --- apps/epg/signals.py | 30 ++++++++++++++++++++++++++++++ core/utils.py | 27 +++++++++++++++++++++++++++ 2 files changed, 57 insertions(+) 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/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 From 3fa5301894e89c5715a6d684d9b53d7d04d59ee8 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 27 May 2025 09:54:28 -0500 Subject: [PATCH 17/21] If cached file doesn't exist for compressed mapped file, properly re-exctract. --- apps/epg/tasks.py | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 3d903193..a7f96363 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -633,11 +633,11 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): 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: @@ -645,9 +645,9 @@ def extract_compressed_file(file_path, output_path=None, delete_original=False): 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: @@ -1116,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) @@ -1150,7 +1154,10 @@ def parse_programs_for_tvg_id(epg_id): return # Update file_path with the new location - file_path = epg_source.file_path + 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 From 68a7f7f4fd1d1b2ce74e910c9e92a68325fdc643 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 27 May 2025 10:46:49 -0500 Subject: [PATCH 18/21] Add profile name to hdhr friendly name and device id. --- apps/hdhr/api_views.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/apps/hdhr/api_views.py b/apps/hdhr/api_views.py index b4f895d4..278efc36 100644 --- a/apps/hdhr/api_views.py +++ b/apps/hdhr/api_views.py @@ -84,13 +84,19 @@ class DiscoverAPIView(APIView): logger.debug(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", From 4dbb363211010cfbcb7b9951a4d75ea7feb32fe0 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 27 May 2025 18:00:35 -0500 Subject: [PATCH 19/21] Removed display-name from logging as it may not exist. --- apps/epg/tasks.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index a7f96363..1a5f832e 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -948,9 +948,9 @@ def parse_channels_only(source): total=total_channels ) if processed_channels > total_channels: - logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - {display_name} - processed {processed_channels - total_channels} additional 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} - {display_name} - processed {processed_channels}/{total_channels}") + logger.debug(f"[parse_channels_only] Processed channel {tvg_id} - processed {processed_channels}/{total_channels}") if process: logger.debug(f"[parse_channels_only] Memory before elem cleanup: {process.memory_info().rss / 1024 / 1024:.2f} MB") # Clear memory From 50048518a9aa7ca32474c8613110d0801054f7ed Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Tue, 27 May 2025 19:05:26 -0500 Subject: [PATCH 20/21] Fixes bug where mulitiple channel initializations can occur which leads to choppy streams and zombie channels. --- apps/proxy/ts_proxy/server.py | 36 +++++++++++++++++++++++++ apps/proxy/ts_proxy/stream_generator.py | 12 ++++++++- apps/proxy/ts_proxy/views.py | 29 ++++++++++++-------- 3 files changed, 65 insertions(+), 12 deletions(-) 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 ef232fd2..b90e1585 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -53,6 +53,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: @@ -63,30 +64,36 @@ 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, ChannelState.WAITING_FOR_CLIENTS] - if channel_state in valid_states: - # Verify the owner is still 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 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, ChannelState.STOPPED]: logger.warning(f"[{client_id}] Channel {channel_id} in state {channel_state}, forcing cleanup") 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() From d37865398344f522504266faa4dad5a0bd9cd72f Mon Sep 17 00:00:00 2001 From: GitHub Actions Date: Wed, 28 May 2025 21:29:07 +0000 Subject: [PATCH 21/21] Release v0.5.1 --- version.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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