From 4595cea40096a69e36fff208571dd6bb058eaa8f Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sat, 3 May 2025 18:39:59 -0500 Subject: [PATCH] Status update overhaul for epg files. --- ...7_epgsource_status_epgsource_last_error.py | 23 + .../migrations/0010_merge_20250503_2147.py | 14 + apps/epg/models.py | 10 + apps/epg/serializers.py | 16 +- apps/epg/tasks.py | 551 ++++++++++++++---- frontend/src/WebSocket.jsx | 43 ++ frontend/src/components/tables/EPGsTable.jsx | 97 ++- frontend/src/store/epgs.jsx | 45 ++ 8 files changed, 666 insertions(+), 133 deletions(-) create mode 100644 apps/epg/migrations/0007_epgsource_status_epgsource_last_error.py create mode 100644 apps/epg/migrations/0010_merge_20250503_2147.py diff --git a/apps/epg/migrations/0007_epgsource_status_epgsource_last_error.py b/apps/epg/migrations/0007_epgsource_status_epgsource_last_error.py new file mode 100644 index 00000000..050b1b4f --- /dev/null +++ b/apps/epg/migrations/0007_epgsource_status_epgsource_last_error.py @@ -0,0 +1,23 @@ +# Generated by Django + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0006_epgsource_refresh_interval_epgsource_refresh_task'), + ] + + operations = [ + migrations.AddField( + model_name='epgsource', + name='status', + field=models.CharField(choices=[('idle', 'Idle'), ('fetching', 'Fetching'), ('parsing', 'Parsing'), ('error', 'Error'), ('success', 'Success')], default='idle', max_length=20), + ), + migrations.AddField( + model_name='epgsource', + name='last_error', + field=models.TextField(blank=True, null=True), + ), + ] diff --git a/apps/epg/migrations/0010_merge_20250503_2147.py b/apps/epg/migrations/0010_merge_20250503_2147.py new file mode 100644 index 00000000..a65117ba --- /dev/null +++ b/apps/epg/migrations/0010_merge_20250503_2147.py @@ -0,0 +1,14 @@ +# Generated by Django 5.1.6 on 2025-05-03 21:47 + +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0007_epgsource_status_epgsource_last_error'), + ('epg', '0009_alter_epgsource_created_at_and_more'), + ] + + operations = [ + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index a0e5343b..d4661690 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -9,6 +9,14 @@ class EPGSource(models.Model): ('xmltv', 'XMLTV URL'), ('schedules_direct', 'Schedules Direct API'), ] + STATUS_CHOICES = [ + ('idle', 'Idle'), + ('fetching', 'Fetching'), + ('parsing', 'Parsing'), + ('error', 'Error'), + ('success', 'Success'), + ] + name = models.CharField(max_length=255, unique=True) source_type = models.CharField(max_length=20, choices=SOURCE_TYPE_CHOICES) url = models.URLField(blank=True, null=True) # For XMLTV @@ -19,6 +27,8 @@ class EPGSource(models.Model): refresh_task = models.ForeignKey( PeriodicTask, on_delete=models.SET_NULL, null=True, blank=True ) + status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='idle') + last_error = models.TextField(blank=True, null=True) created_at = models.DateTimeField( auto_now_add=True, help_text="Time when this source was created" diff --git a/apps/epg/serializers.py b/apps/epg/serializers.py index e4ff932e..efcd1aba 100644 --- a/apps/epg/serializers.py +++ b/apps/epg/serializers.py @@ -8,7 +8,21 @@ class EPGSourceSerializer(serializers.ModelSerializer): class Meta: model = EPGSource - fields = ['id', 'name', 'source_type', 'url', 'api_key', 'is_active', 'epg_data_ids', 'refresh_interval', 'created_at', 'updated_at'] + fields = [ + 'id', + 'name', + 'source_type', + 'url', + 'api_key', + 'is_active', + 'file_path', + 'refresh_interval', + 'status', + 'last_error', + 'created_at', + 'updated_at', + 'epg_data_ids' + ] def get_epg_data_ids(self, obj): return list(obj.epgs.values_list('id', flat=True)) diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 55337277..7755bd7d 100644 --- a/apps/epg/tasks.py +++ b/apps/epg/tasks.py @@ -6,6 +6,7 @@ import os import uuid import requests import xml.etree.ElementTree as ET +import time # Add import for tracking download progress from datetime import datetime, timedelta, timezone as dt_timezone from celery import shared_task @@ -24,6 +25,30 @@ from core.utils import acquire_task_lock, release_task_lock logger = logging.getLogger(__name__) +def send_epg_update(source_id, action, progress, **kwargs): + """Send WebSocket update about EPG download/parsing progress""" + # Start with the base data dictionary + data = { + "progress": progress, + "type": "epg_refresh", + "source": source_id, + "action": action, + } + + # Add the additional key-value pairs from kwargs + data.update(kwargs) + + # Now, send the updated data dictionary + channel_layer = get_channel_layer() + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + 'data': data + } + ) + + @shared_task def refresh_all_epg_data(): logger.info("Starting refresh_epg_data task.") @@ -36,31 +61,75 @@ def refresh_all_epg_data(): logger.info("Finished refresh_epg_data task.") return "EPG data refreshed." + @shared_task def refresh_epg_data(source_id): if not acquire_task_lock('refresh_epg_data', source_id): logger.debug(f"EPG refresh for {source_id} already running") return - source = EPGSource.objects.get(id=source_id) - if not source.is_active: - logger.info(f"EPG source {source_id} is not active. Skipping.") - return + try: + source = EPGSource.objects.get(id=source_id) + if not source.is_active: + logger.info(f"EPG source {source_id} is not active. Skipping.") + return - logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})") - if source.source_type == 'xmltv': - fetch_xmltv(source) - parse_channels_only(source) - parse_programs_for_source(source) - elif source.source_type == 'schedules_direct': - fetch_schedules_direct(source) + logger.info(f"Processing EPGSource: {source.name} (type: {source.source_type})") + if source.source_type == 'xmltv': + fetch_success = fetch_xmltv(source) + if not fetch_success: + logger.error(f"Failed to fetch XMLTV for source {source.name}") + release_task_lock('refresh_epg_data', source_id) + return - source.save(update_fields=['updated_at']) + parse_channels_success = parse_channels_only(source) + if not parse_channels_success: + logger.error(f"Failed to parse channels for source {source.name}") + release_task_lock('refresh_epg_data', source_id) + return + + parse_programs_for_source(source) + + elif source.source_type == 'schedules_direct': + fetch_schedules_direct(source) + + source.save(update_fields=['updated_at']) + except Exception as e: + logger.error(f"Error in refresh_epg_data for source {source_id}: {e}", exc_info=True) + try: + source = EPGSource.objects.get(id=source_id) + source.status = 'error' + source.last_error = f"Error refreshing EPG data: {str(e)}" + source.save(update_fields=['status', 'last_error']) + send_epg_update(source_id, "refresh", 100, status="error", error=str(e)) + except Exception as inner_e: + logger.error(f"Error updating source status: {inner_e}") + finally: + release_task_lock('refresh_epg_data', source_id) - release_task_lock('refresh_epg_data', source_id) 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}") + + # Set the status to success in the database + source.status = 'success' + source.save(update_fields=['status']) + + # Send a download complete notification + send_epg_update(source.id, "downloading", 100, status="success") + + # Return True to indicate successful fetch, processing will continue with parse_channels_only + return True + + # Handle cases where no URL is provided and no valid file path exists if not source.url: + # Update source status for missing URL + source.status = 'error' + source.last_error = "No URL provided and no valid local file exists" + source.save(update_fields=['status', 'last_error']) + send_epg_update(source.id, "downloading", 100, status="error", error="No URL provided and no valid local file exists") return False if os.path.exists(source.get_cache_file()): @@ -79,46 +148,132 @@ def fetch_xmltv(source): logger.debug(f"Using default user agent: {user_agent}") except (ValueError, Exception) as e: logger.warning(f"Error retrieving default user agent, using fallback: {e}") + headers = { 'User-Agent': user_agent } - response = requests.get(source.url, headers=headers, timeout=30) + # Update status to fetching before starting download + source.status = 'fetching' + source.save(update_fields=['status']) - # Handle 404 specifically - if response.status_code == 404: - logger.error(f"EPG URL not found (404): {source.url}") - # Just log the error without marking inactive, will retry on next scheduled run - logger.warning(f"EPG source '{source.name}' encountered a 404 error - will retry on next scheduled run") + # Send initial download notification + send_epg_update(source.id, "downloading", 0) - # Notify users through the WebSocket about the EPG fetch failure - channel_layer = get_channel_layer() - async_to_sync(channel_layer.group_send)( - 'updates', - { - 'type': 'update', - 'data': { - "success": False, - "type": "epg_fetch_error", - "source_id": source.id, - "source_name": source.name, - "error_code": 404, - "message": f"EPG source '{source.name}' returned 404 error - will retry on next scheduled run" + # Use streaming response to track download progress + with requests.get(source.url, headers=headers, stream=True, timeout=30) as response: + # Handle 404 specifically + if response.status_code == 404: + logger.error(f"EPG URL not found (404): {source.url}") + # Update status to error in the database + source.status = 'error' + source.last_error = f"EPG source '{source.name}' returned 404 error - will retry on next scheduled run" + source.save(update_fields=['status', 'last_error']) + + # Notify users through the WebSocket about the EPG fetch failure + channel_layer = get_channel_layer() + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + 'data': { + "success": False, + "type": "epg_fetch_error", + "source_id": source.id, + "source_name": source.name, + "error_code": 404, + "message": f"EPG source '{source.name}' returned 404 error - will retry on next scheduled run" + } } - } - ) - return False + ) + # Ensure we update the download progress to 100 with error status + send_epg_update(source.id, "downloading", 100, status="error", error="URL not found (404)") + return False - response.raise_for_status() - logger.debug("XMLTV data fetched successfully.") + # For all other error status codes + if response.status_code >= 400: + error_message = f"HTTP error {response.status_code}" + user_message = f"EPG source '{source.name}' encountered HTTP error {response.status_code}" - cache_file = source.get_cache_file() + # Update status to error in the database + source.status = 'error' + source.last_error = user_message + source.save(update_fields=['status', 'last_error']) - # Save raw data - with open(cache_file, 'wb') as f: - f.write(response.content) - logger.info(f"Cached EPG file saved to {cache_file}") - return True + # Notify users through the WebSocket + channel_layer = get_channel_layer() + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + 'data': { + "success": False, + "type": "epg_fetch_error", + "source_id": source.id, + "source_name": source.name, + "error_code": response.status_code, + "message": user_message + } + } + ) + # Update download progress + send_epg_update(source.id, "downloading", 100, status="error", error=user_message) + return False + + response.raise_for_status() + logger.debug("XMLTV data fetched successfully.") + + cache_file = source.get_cache_file() + + # Check if we have content length for progress tracking + total_size = int(response.headers.get('content-length', 0)) + downloaded = 0 + start_time = time.time() + last_update_time = start_time + + with open(cache_file, 'wb') as f: + for chunk in response.iter_content(chunk_size=8192): + if chunk: + f.write(chunk) + + downloaded += len(chunk) + elapsed_time = time.time() - start_time + + # Calculate download speed in KB/s + speed = downloaded / elapsed_time / 1024 if elapsed_time > 0 else 0 + + # Calculate progress percentage + if total_size and total_size > 0: + progress = min(100, int((downloaded / total_size) * 100)) + else: + # If no content length header, estimate progress + progress = min(95, int((downloaded / (10 * 1024 * 1024)) * 100)) # Assume 10MB if unknown + + # Time remaining (in seconds) + time_remaining = (total_size - downloaded) / (speed * 1024) if speed > 0 and total_size > 0 else 0 + + # Only send updates every 0.5 seconds to avoid flooding + current_time = time.time() + if current_time - last_update_time >= 0.5 and progress > 0: + last_update_time = current_time + send_epg_update( + source.id, + "downloading", + progress, + speed=speed, + elapsed_time=elapsed_time, + time_remaining=time_remaining + ) + + # Send completion notification + send_epg_update(source.id, "downloading", 100) + + # Update status to parsing + source.status = 'parsing' + source.save(update_fields=['status']) + + logger.info(f"Cached EPG file saved to {cache_file}") + return True except requests.exceptions.HTTPError as e: logger.error(f"HTTP Error fetching XMLTV from {source.name}: {e}", exc_info=True) @@ -140,6 +295,11 @@ def fetch_xmltv(source): elif status_code >= 500: user_message = f"EPG source '{source.name}' server error (HTTP {status_code}) - will retry later" + # Update source status to error with the error message + source.status = 'error' + source.last_error = user_message + source.save(update_fields=['status', 'last_error']) + # Notify users through the WebSocket about the EPG fetch failure channel_layer = get_channel_layer() async_to_sync(channel_layer.group_send)( @@ -157,9 +317,50 @@ def fetch_xmltv(source): } } ) + + # Ensure we update the download progress to 100 with error status + send_epg_update(source.id, "downloading", 100, status="error", error=user_message) + return False + except requests.exceptions.ConnectionError as e: + # Handle connection errors separately + error_message = str(e) + user_message = f"Connection error: Unable to connect to EPG source '{source.name}'" + logger.error(f"Connection error fetching XMLTV from {source.name}: {e}", exc_info=True) + + # Update source status + source.status = 'error' + source.last_error = user_message + source.save(update_fields=['status', 'last_error']) + + # Send notifications + channel_layer = get_channel_layer() + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + 'data': { + "success": False, + "type": "epg_fetch_error", + "source_id": source.id, + "source_name": source.name, + "error_code": "connection_error", + "message": user_message + } + } + ) + send_epg_update(source.id, "downloading", 100, status="error", error=user_message) return False except Exception as e: + error_message = str(e) logger.error(f"Error fetching XMLTV from {source.name}: {e}", exc_info=True) + + # Update source status for general exceptions too + source.status = 'error' + source.last_error = f"Error: {error_message}" + source.save(update_fields=['status', 'last_error']) + + # Ensure we update the download progress to 100 with error status + send_epg_update(source.id, "downloading", 100, status="error", error=f"Error: {error_message}") return False @@ -168,36 +369,57 @@ def parse_channels_only(source): if not file_path: file_path = source.get_cache_file() - # Check if the file exists - if not os.path.exists(file_path): - logger.error(f"EPG file does not exist at path: {file_path}") + # Send initial parsing notification + send_epg_update(source.id, "parsing_channels", 0) - # Update the source's file_path to the default cache location - new_path = source.get_cache_file() - logger.info(f"Updating file_path from '{file_path}' to '{new_path}'") - source.file_path = new_path - source.save(update_fields=['file_path']) - - # If the source has a URL, fetch the data before continuing - if source.url: - logger.info(f"Fetching new EPG data from URL: {source.url}") - fetch_xmltv(source) - - # 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}") - return - else: - logger.error(f"No URL provided for EPG source {source.name}, cannot fetch new data") - return - - file_path = new_path - - logger.info(f"Parsing channels from EPG file: {file_path}") - existing_epgs = {e.tvg_id: e for e in EPGData.objects.filter(epg_source=source)} - - # Read entire file (decompress if .gz) try: + # Check if the file exists + if not os.path.exists(file_path): + logger.error(f"EPG file does not exist at path: {file_path}") + + # Update the source's file_path to the default cache location + new_path = source.get_cache_file() + logger.info(f"Updating file_path from '{file_path}' to '{new_path}'") + source.file_path = new_path + source.save(update_fields=['file_path']) + + # If the source has a URL, fetch the data before continuing + if source.url: + logger.info(f"Fetching new EPG data from URL: {source.url}") + if not fetch_xmltv(source): # Check if fetch was successful + logger.error(f"Failed to fetch EPG data from URL: {source.url}") + # Update status to error + source.status = 'error' + source.last_error = f"Failed to fetch EPG data from URL" + source.save(update_fields=['status', 'last_error']) + # Send error notification + send_epg_update(source.id, "parsing_channels", 100, status="error", error="Failed to fetch EPG data") + 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}") + # Update status to error + source.status = 'error' + source.last_error = f"Failed to fetch EPG data, file missing after download" + source.save(update_fields=['status', 'last_error']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error="File not found after download") + return False + else: + logger.error(f"No URL provided for EPG source {source.name}, cannot fetch new data") + # Update status to error + source.status = 'error' + source.last_error = f"No URL provided, cannot fetch EPG data" + source.save(update_fields=['status', 'last_error']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error="No URL provided") + return False + + file_path = new_path + + logger.info(f"Parsing channels from EPG file: {file_path}") + existing_epgs = {e.tvg_id: e for e in EPGData.objects.filter(epg_source=source)} + + # Read entire file (decompress if .gz) if file_path.endswith('.gz'): with open(file_path, 'rb') as gz_file: decompressed = gzip.decompress(gz_file.read()) @@ -205,54 +427,87 @@ def parse_channels_only(source): else: with open(file_path, 'r', encoding='utf-8') as xml_file: xml_data = xml_file.read() + + # Update progress to show file read completed + send_epg_update(source.id, "parsing_channels", 25) + + root = ET.fromstring(xml_data) + channels = root.findall('channel') + + epgs_to_create = [] + epgs_to_update = [] + + logger.info(f"Found {len(channels)} entries in {file_path}") + + # Update progress to show parsing started + send_epg_update(source.id, "parsing_channels", 50) + + total_channels = len(channels) + for i, channel_elem in enumerate(channels): + tvg_id = channel_elem.get('id', '').strip() + if not tvg_id: + continue # skip blank/invalid IDs + + display_name = channel_elem.findtext('display-name', default=tvg_id).strip() + + if tvg_id in existing_epgs: + epg_obj = existing_epgs[tvg_id] + if epg_obj.name != display_name: + epg_obj.name = display_name + epgs_to_update.append(epg_obj) + else: + epgs_to_create.append(EPGData( + tvg_id=tvg_id, + name=display_name, + epg_source=source, + )) + + # Send occasional progress updates + if i % 100 == 0 or i == total_channels - 1: + progress = 50 + int((i / total_channels) * 40) # Scale to 50-90% range + send_epg_update(source.id, "parsing_channels", progress) + + # Update progress before database operations + send_epg_update(source.id, "parsing_channels", 90) + + if epgs_to_create: + EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True) + + if epgs_to_update: + EPGData.objects.bulk_update(epgs_to_update, ["name"]) + + # Send completion notification + send_epg_update(source.id, "parsing_channels", 100, status="success") + + channel_layer = get_channel_layer() + async_to_sync(channel_layer.group_send)( + 'updates', + { + 'type': 'update', + "data": {"success": True, "type": "epg_channels"} + } + ) + + logger.info("Finished parsing channel info.") + return True + except FileNotFoundError: logger.error(f"EPG file not found at: {file_path}") - return + # Update status to error + source.status = 'error' + source.last_error = f"EPG file not found: {file_path}" + source.save(update_fields=['status', 'last_error']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error="File not found") + return False except Exception as e: logger.error(f"Error reading EPG file {file_path}: {e}", exc_info=True) - return + # Update status to error + source.status = 'error' + source.last_error = f"Error parsing EPG file: {str(e)}" + source.save(update_fields=['status', 'last_error']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(e)) + return False - root = ET.fromstring(xml_data) - channels = root.findall('channel') - - epgs_to_create = [] - epgs_to_update = [] - - logger.info(f"Found {len(channels)} entries in {file_path}") - for channel_elem in channels: - tvg_id = channel_elem.get('id', '').strip() - if not tvg_id: - continue # skip blank/invalid IDs - - display_name = channel_elem.findtext('display-name', default=tvg_id).strip() - - if tvg_id in existing_epgs: - epg_obj = existing_epgs[tvg_id] - if epg_obj.name != display_name: - epg_obj.name = display_name - epgs_to_update.append(epg_obj) - else: - epgs_to_create.append(EPGData( - tvg_id=tvg_id, - name=display_name, - epg_source=source, - )) - - if epgs_to_create: - EPGData.objects.bulk_create(epgs_to_create, ignore_conflicts=True) - if epgs_to_update: - EPGData.objects.bulk_update(epgs_to_update, ["name"]) - - channel_layer = get_channel_layer() - async_to_sync(channel_layer.group_send)( - 'updates', - { - 'type': 'update', - "data": {"success": True, "type": "epg_channels"} - } - ) - - logger.info("Finished parsing channel info.") @shared_task def parse_programs_for_tvg_id(epg_id): @@ -433,12 +688,70 @@ def parse_programs_for_tvg_id(epg_id): logger.info(f"Completed program parsing for tvg_id={epg.tvg_id}.") + def parse_programs_for_source(epg_source, tvg_id=None): - file_path = epg_source.file_path - epg_entries = EPGData.objects.filter(epg_source=epg_source) - for epg in epg_entries: - if epg.tvg_id: - parse_programs_for_tvg_id(epg.id) + # Send initial programs parsing notification + send_epg_update(epg_source.id, "parsing_programs", 0) + + try: + epg_entries = EPGData.objects.filter(epg_source=epg_source) + total_entries = epg_entries.count() + processed = 0 + + if total_entries == 0: + logger.info(f"No EPG entries found for source: {epg_source.name}") + # Update status - this is not an error, just no entries + epg_source.status = 'success' + epg_source.save(update_fields=['status']) + send_epg_update(epg_source.id, "parsing_programs", 100, status="success") + return True + + logger.info(f"Parsing programs for {total_entries} EPG entries from source: {epg_source.name}") + + failed_entries = [] + for epg in epg_entries: + if epg.tvg_id: + try: + result = parse_programs_for_tvg_id(epg.id) + if result == "Task already running": + logger.info(f"Program parse for {epg.id} already in progress, skipping") + + processed += 1 + progress = min(95, int((processed / total_entries) * 100)) if total_entries > 0 else 50 + send_epg_update(epg_source.id, "parsing_programs", progress) + except Exception as e: + logger.error(f"Error parsing programs for tvg_id={epg.tvg_id}: {e}", exc_info=True) + failed_entries.append(f"{epg.tvg_id}: {str(e)}") + + # If any entries failed, mark the source as error but continue + if failed_entries: + epg_source.status = 'error' + epg_source.last_error = f"Failed to parse some entries: {', '.join(failed_entries[:5])}" + ( + "..." if len(failed_entries) > 5 else "") + epg_source.save(update_fields=['status', 'last_error']) + send_epg_update(epg_source.id, "parsing_programs", 100, status="error", + error=f"Failed to parse {len(failed_entries)} of {total_entries} entries") + return False + + # All successful + epg_source.status = 'success' + epg_source.save(update_fields=['status']) + + # Send completion notification with updated status + send_epg_update(epg_source.id, "parsing_programs", 100, status="success") + + logger.info(f"Completed parsing all programs for source: {epg_source.name}") + return True + + except Exception as e: + logger.error(f"Error in parse_programs_for_source: {e}", exc_info=True) + # Update status to error + epg_source.status = 'error' + epg_source.last_error = f"Error parsing programs: {str(e)}" + epg_source.save(update_fields=['status', 'last_error']) + send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error=str(e)) + return False + def fetch_schedules_direct(source): logger.info(f"Fetching Schedules Direct data from source: {source.name}") diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index 73d8473b..0533f13a 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -116,6 +116,19 @@ export const WebsocketProvider = ({ children }) => { message: 'EPG channels updated!', color: 'green.5', }); + + // If source_id is provided, update that specific EPG's status + if (event.data.source_id) { + const epgsState = useEPGsStore.getState(); + const epg = epgsState.epgs[event.data.source_id]; + if (epg) { + epgsState.updateEPG({ + ...epg, + status: 'success' + }); + } + } + fetchEPGData(); break; @@ -156,6 +169,36 @@ export const WebsocketProvider = ({ children }) => { color: 'orange.5', autoClose: 8000, }); + + // Update EPG status in store + if (event.data.source_id) { + const epgsState = useEPGsStore.getState(); + const epg = epgsState.epgs[event.data.source_id]; + if (epg) { + epgsState.updateEPG({ + ...epg, + status: 'error', + last_error: event.data.message + }); + } + } + break; + + case 'epg_refresh': + // Update the store with progress information + const epgsState = useEPGsStore.getState(); + epgsState.updateEPGProgress(event.data); + + // If progress is complete (100%), show a notification and refresh EPG data + if (event.data.progress === 100 && event.data.action === "parsing_programs") { + notifications.show({ + title: 'EPG Processing Complete', + message: 'EPG data has been updated successfully', + color: 'green.5', + }); + + fetchEPGData(); + } break; default: diff --git a/frontend/src/components/tables/EPGsTable.jsx b/frontend/src/components/tables/EPGsTable.jsx index 1f76eb33..e14a75f0 100644 --- a/frontend/src/components/tables/EPGsTable.jsx +++ b/frontend/src/components/tables/EPGsTable.jsx @@ -14,18 +14,34 @@ import { Flex, useMantineTheme, Switch, + Badge, + Progress, + Stack, } from '@mantine/core'; import { notifications } from '@mantine/notifications'; import { IconSquarePlus } from '@tabler/icons-react'; import { RefreshCcw, SquareMinus, SquarePen } from 'lucide-react'; import dayjs from 'dayjs'; +// Helper function to get status badge color +const getStatusColor = (status) => { + switch (status) { + case 'idle': return 'gray'; + case 'fetching': return 'blue'; + case 'parsing': return 'indigo'; + case 'error': return 'red'; + case 'success': return 'green'; + default: return 'gray'; + } +}; + const EPGsTable = () => { const [epg, setEPG] = useState(null); const [epgModalOpen, setEPGModalOpen] = useState(false); const [rowSelection, setRowSelection] = useState([]); const epgs = useEPGsStore((s) => s.epgs); + const refreshProgress = useEPGsStore((s) => s.refreshProgress); const theme = useMantineTheme(); @@ -36,6 +52,35 @@ const EPGsTable = () => { }); }; + const buildProgressDisplay = (data) => { + const progress = refreshProgress[data.id] || null; + + if (!progress) return null; + + let label = ''; + switch (progress.action) { + case 'downloading': + label = 'Downloading'; + break; + case 'parsing_channels': + label = 'Parsing Channels'; + break; + case 'parsing_programs': + label = 'Parsing Programs'; + break; + default: + return null; + } + + return ( + + {label}: {parseInt(progress.progress)}% + + {progress.speed && Speed: {parseInt(progress.speed)} KB/s} + + ); + }; + const columns = useMemo( //column definitions... () => [ @@ -51,24 +96,50 @@ const EPGsTable = () => { size: 120, minSize: 100, }, + { + header: 'Status', + accessorKey: 'status', + size: 100, + minSize: 80, + Cell: ({ row }) => { + const data = row.original; + + // Check if there's an active progress for this EPG + if (refreshProgress[data.id] && refreshProgress[data.id].progress < 100) { + return buildProgressDisplay(data); + } + + // Return the original status display if no active progress + return ( + + {data.status} + + ); + }, + }, { header: 'URL / API Key', accessorKey: 'url', size: 200, minSize: 120, enableSorting: false, - Cell: ({ cell }) => ( -
- {cell.getValue()} -
- ), + Cell: ({ cell, row }) => { + const value = cell.getValue() || row.original.api_key || ''; + return ( + +
+ {value} +
+ + ); + }, }, { header: 'Active', @@ -97,7 +168,7 @@ const EPGsTable = () => { enableSorting: false, }, ], - [] + [refreshProgress] ); //optionally access the underlying virtualizer instance diff --git a/frontend/src/store/epgs.jsx b/frontend/src/store/epgs.jsx index bcb9fd88..0a0fd8f7 100644 --- a/frontend/src/store/epgs.jsx +++ b/frontend/src/store/epgs.jsx @@ -7,6 +7,7 @@ const useEPGsStore = create((set) => ({ tvgsById: {}, isLoading: false, error: null, + refreshProgress: {}, fetchEPGs: async () => { set({ isLoading: true, error: null }); @@ -62,6 +63,50 @@ const useEPGsStore = create((set) => ({ return { epgs: updatedEPGs }; }), + + updateEPGProgress: (data) => + set((state) => { + // Early exit if source doesn't exist in our EPGs store + if (!state.epgs[data.source] && !data.status) { + return state; + } + + // Create a new refreshProgress object that includes the current update + const newRefreshProgress = { + ...state.refreshProgress, + [data.source]: { + action: data.action, + progress: data.progress, + speed: data.speed, + elapsed_time: data.elapsed_time, + time_remaining: data.time_remaining, + status: data.status || 'in_progress' + } + }; + + // Set the EPG source status based on the update + // First prioritize explicit status values from the backend + const sourceStatus = data.status ? data.status // Use explicit status if provided + : data.action === "downloading" ? "fetching" + : data.action === "parsing_channels" || data.action === "parsing_programs" ? "parsing" + : data.progress === 100 ? "success" // Mark as success when progress is 100% + : state.epgs[data.source]?.status || 'idle'; + + // Create a new epgs object with the updated source status + const newEpgs = { + ...state.epgs, + [data.source]: { + ...state.epgs[data.source], + status: sourceStatus, + last_error: data.status === 'error' ? (data.error || 'Unknown error') : state.epgs[data.source]?.last_error + } + }; + + return { + refreshProgress: newRefreshProgress, + epgs: newEpgs + }; + }), })); export default useEPGsStore;