diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 13535c8c..b612fa2c 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -185,6 +185,30 @@ class ChannelViewSet(viewsets.ModelViewSet): context['include_streams'] = include_streams return context + @action(detail=False, methods=['patch'], url_path='edit/bulk') + def edit_bulk(self, request): + data_list = request.data + if not isinstance(data_list, list): + return Response({"error": "Expected a list of channel objects objects"}, status=status.HTTP_400_BAD_REQUEST) + + updated_channels = [] + try: + with transaction.atomic(): + for item in data_list: + channel = Channel.objects.id(id=item.pop('id')) + for key, value in item.items(): + setattr(channel, key, value) + + channel.save(update_fields=item.keys()) + updated_channels.append(channel) + except Exception as e: + logger.error("Error during bulk channel edit", e) + return Response({"error": e}, status=500) + + response_data = ChannelSerializer(updated_channels, many=True).data + + return Response(response_data, status=status.HTTP_200_OK) + @action(detail=False, methods=['get'], url_path='ids') def get_ids(self, request, *args, **kwargs): # Get the filtered queryset @@ -204,12 +228,13 @@ class ChannelViewSet(viewsets.ModelViewSet): operation_description="Auto-assign channel_number in bulk by an ordered list of channel IDs.", request_body=openapi.Schema( type=openapi.TYPE_OBJECT, - required=["channel_order"], + required=["channel_ids"], properties={ - "channel_order": openapi.Schema( + "starting_number": openapi.Schema(type=openapi.TYPE_STRING, description="Starting channel number to assign"), + "channel_ids": openapi.Schema( type=openapi.TYPE_ARRAY, items=openapi.Items(type=openapi.TYPE_INTEGER), - description="List of channel IDs in the new order" + description="Channel IDs to assign" ) } ), @@ -218,9 +243,11 @@ class ChannelViewSet(viewsets.ModelViewSet): @action(detail=False, methods=['post'], url_path='assign') def assign(self, request): with transaction.atomic(): - channel_order = request.data.get('channel_order', []) - for order, channel_id in enumerate(channel_order, start=1): - Channel.objects.filter(id=channel_id).update(channel_number=order) + channel_ids = request.data.get('channel_ids', []) + channel_num = request.data.get('starting_number', 1) + for channel_id in channel_ids: + Channel.objects.filter(id=channel_id).update(channel_number=channel_num) + channel_num = channel_num + 1 return Response({"message": "Channels have been auto-assigned!"}, status=status.HTTP_200_OK) @@ -285,6 +312,10 @@ class ChannelViewSet(viewsets.ModelViewSet): {"error": f"Channel number {channel_number} is already in use. Please choose a different number."}, status=status.HTTP_400_BAD_REQUEST ) + #Get the tvc_guide_stationid from custom properties if it exists + tvc_guide_stationid = None + if 'tvc-guide-stationid' in stream_custom_props: + tvc_guide_stationid = stream_custom_props['tvc-guide-stationid'] @@ -292,6 +323,7 @@ class ChannelViewSet(viewsets.ModelViewSet): 'channel_number': channel_number, 'name': name, 'tvg_id': stream.tvg_id, + 'tvc_guide_stationid': tvc_guide_stationid, 'channel_group_id': channel_group.id, 'streams': [stream_id], } diff --git a/apps/channels/migrations/0018_channelgroupm3uaccount_custom_properties_and_more.py b/apps/channels/migrations/0018_channelgroupm3uaccount_custom_properties_and_more.py new file mode 100644 index 00000000..51507843 --- /dev/null +++ b/apps/channels/migrations/0018_channelgroupm3uaccount_custom_properties_and_more.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 on 2025-04-27 14:12 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dispatcharr_channels', '0017_alter_channelgroup_name'), + ] + + operations = [ + migrations.AddField( + model_name='channelgroupm3uaccount', + name='custom_properties', + field=models.TextField(blank=True, null=True), + ), + ] diff --git a/apps/channels/migrations/0019_channel_tvc_guide_stationid.py b/apps/channels/migrations/0019_channel_tvc_guide_stationid.py new file mode 100644 index 00000000..86f76b82 --- /dev/null +++ b/apps/channels/migrations/0019_channel_tvc_guide_stationid.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 on 2025-05-04 00:02 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('dispatcharr_channels', '0018_channelgroupm3uaccount_custom_properties_and_more'), + ] + + operations = [ + migrations.AddField( + model_name='channel', + name='tvc_guide_stationid', + field=models.CharField(blank=True, max_length=255, null=True), + ), + ] diff --git a/apps/channels/models.py b/apps/channels/models.py index fc6af558..191eb45e 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -1,6 +1,5 @@ from django.db import models from django.core.exceptions import ValidationError -from core.models import StreamProfile from django.conf import settings from core.models import StreamProfile, CoreSettings from core.utils import RedisClient @@ -237,6 +236,8 @@ class Channel(models.Model): help_text="Channel group this channel belongs to." ) tvg_id = models.CharField(max_length=255, blank=True, null=True) + tvc_guide_stationid = models.CharField(max_length=255, blank=True, null=True) + epg_data = models.ForeignKey( EPGData, on_delete=models.SET_NULL, @@ -488,6 +489,7 @@ class ChannelGroupM3UAccount(models.Model): on_delete=models.CASCADE, related_name='channel_group' ) + custom_properties = models.TextField(null=True, blank=True) enabled = models.BooleanField(default=True) class Meta: diff --git a/apps/channels/serializers.py b/apps/channels/serializers.py index c3772f34..712d8569 100644 --- a/apps/channels/serializers.py +++ b/apps/channels/serializers.py @@ -45,6 +45,7 @@ class StreamSerializer(serializers.ModelSerializer): 'local_file', 'current_viewers', 'updated_at', + 'last_seen', 'stream_profile_id', 'is_custom', 'channel_group', @@ -151,6 +152,7 @@ class ChannelSerializer(serializers.ModelSerializer): 'name', 'channel_group_id', 'tvg_id', + 'tvc_guide_stationid', 'epg_data_id', 'streams', 'stream_profile_id', diff --git a/apps/epg/api_views.py b/apps/epg/api_views.py index 526172f1..70bfa87d 100644 --- a/apps/epg/api_views.py +++ b/apps/epg/api_views.py @@ -18,7 +18,9 @@ logger = logging.getLogger(__name__) # 1) EPG Source API (CRUD) # ───────────────────────────── class EPGSourceViewSet(viewsets.ModelViewSet): - """Handles CRUD operations for EPG sources""" + """ + API endpoint that allows EPG sources to be viewed or edited. + """ queryset = EPGSource.objects.all() serializer_class = EPGSourceSerializer permission_classes = [IsAuthenticated] @@ -50,6 +52,21 @@ class EPGSourceViewSet(viewsets.ModelViewSet): return Response(serializer.data, status=status.HTTP_201_CREATED) + def partial_update(self, request, *args, **kwargs): + """Handle partial updates with special logic for is_active field""" + instance = self.get_object() + + # Check if we're toggling is_active + if 'is_active' in request.data and instance.is_active != request.data['is_active']: + # Set appropriate status based on new is_active value + if request.data['is_active']: + request.data['status'] = 'idle' + else: + request.data['status'] = 'disabled' + + # Continue with regular partial update + return super().partial_update(request, *args, **kwargs) + # ───────────────────────────── # 2) Program API (CRUD) # ───────────────────────────── 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/migrations/0011_update_epgsource_fields.py b/apps/epg/migrations/0011_update_epgsource_fields.py new file mode 100644 index 00000000..44c38a76 --- /dev/null +++ b/apps/epg/migrations/0011_update_epgsource_fields.py @@ -0,0 +1,42 @@ +# Generated by Django 5.1.6 on 2025-05-04 21:43 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('epg', '0010_merge_20250503_2147'), + ] + + operations = [ + # Change updated_at field + migrations.AlterField( + model_name='epgsource', + name='updated_at', + field=models.DateTimeField(blank=True, help_text='Time when this source was last successfully refreshed', null=True), + ), + + # Add new last_message field + migrations.AddField( + model_name='epgsource', + name='last_message', + field=models.TextField(blank=True, help_text='Last status message, including success results or error information', null=True), + ), + + # Copy data from last_error to last_message + migrations.RunPython( + code=lambda apps, schema_editor: apps.get_model('epg', 'EPGSource').objects.all().update( + last_message=models.F('last_error') + ), + reverse_code=lambda apps, schema_editor: apps.get_model('epg', 'EPGSource').objects.all().update( + last_error=models.F('last_message') + ), + ), + + # Remove the old field + migrations.RemoveField( + model_name='epgsource', + name='last_error', + ), + ] diff --git a/apps/epg/models.py b/apps/epg/models.py index a0e5343b..ed8f2708 100644 --- a/apps/epg/models.py +++ b/apps/epg/models.py @@ -9,6 +9,23 @@ class EPGSource(models.Model): ('xmltv', 'XMLTV URL'), ('schedules_direct', 'Schedules Direct API'), ] + + STATUS_IDLE = 'idle' + STATUS_FETCHING = 'fetching' + STATUS_PARSING = 'parsing' + STATUS_ERROR = 'error' + STATUS_SUCCESS = 'success' + STATUS_DISABLED = 'disabled' + + STATUS_CHOICES = [ + (STATUS_IDLE, 'Idle'), + (STATUS_FETCHING, 'Fetching'), + (STATUS_PARSING, 'Parsing'), + (STATUS_ERROR, 'Error'), + (STATUS_SUCCESS, 'Success'), + (STATUS_DISABLED, 'Disabled'), + ] + 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,13 +36,23 @@ 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=STATUS_IDLE + ) + last_message = models.TextField( + null=True, + blank=True, + help_text="Last status message, including success results or error information" + ) created_at = models.DateTimeField( auto_now_add=True, help_text="Time when this source was created" ) updated_at = models.DateTimeField( - auto_now=True, - help_text="Time when this source was last updated" + null=True, blank=True, + help_text="Time when this source was last successfully refreshed" ) def __str__(self): @@ -46,6 +73,15 @@ class EPGSource(models.Model): return cache + def save(self, *args, **kwargs): + # Prevent auto_now behavior by handling updated_at manually + if 'update_fields' in kwargs and 'updated_at' not in kwargs['update_fields']: + # Don't modify updated_at for regular updates + kwargs.setdefault('update_fields', []) + if 'updated_at' in kwargs['update_fields']: + kwargs['update_fields'].remove('updated_at') + super().save(*args, **kwargs) + class EPGData(models.Model): # Removed the Channel foreign key. We now just store the original tvg_id # and a name (which might simply be the tvg_id if no real channel exists). diff --git a/apps/epg/serializers.py b/apps/epg/serializers.py index e4ff932e..09390237 100644 --- a/apps/epg/serializers.py +++ b/apps/epg/serializers.py @@ -8,10 +8,24 @@ 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_message', + 'created_at', + 'updated_at', + 'epg_data_ids' + ] def get_epg_data_ids(self, obj): - return list(obj.epgs.values_list('id', flat=True)) + return list(obj.epgs.values_list('id', flat=True)) class ProgramDataSerializer(serializers.ModelSerializer): class Meta: diff --git a/apps/epg/signals.py b/apps/epg/signals.py index 82db7fad..601ec68b 100644 --- a/apps/epg/signals.py +++ b/apps/epg/signals.py @@ -1,4 +1,4 @@ -from django.db.models.signals import post_save, post_delete +from django.db.models.signals import post_save, post_delete, pre_save from django.dispatch import receiver from .models import EPGSource from .tasks import refresh_epg_data @@ -54,3 +54,25 @@ def delete_refresh_task(sender, instance, **kwargs): """ if instance.refresh_task: instance.refresh_task.delete() + +@receiver(pre_save, sender=EPGSource) +def update_status_on_active_change(sender, instance, **kwargs): + """ + When an EPGSource's is_active field changes, update the status accordingly. + """ + if instance.pk: # Only for existing records, not new ones + try: + # Get the current record from the database + old_instance = EPGSource.objects.get(pk=instance.pk) + + # If is_active changed, update the status + if old_instance.is_active != instance.is_active: + if instance.is_active: + # When activating, set status to idle + instance.status = 'idle' + else: + # When deactivating, set status to disabled + instance.status = 'disabled' + except EPGSource.DoesNotExist: + # New record, will use default status + pass diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index cb985a51..52f8f5a1 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,32 +61,76 @@ 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_message = f"Error refreshing EPG data: {str(e)}" + source.save(update_fields=['status', 'last_message']) + 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: - return + # Update source status for missing URL + source.status = 'error' + source.last_message = "No URL provided and no valid local file exists" + source.save(update_fields=['status', 'last_message']) + 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()): os.remove(source.get_cache_file()) @@ -79,83 +148,369 @@ 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) - response.raise_for_status() - logger.debug("XMLTV data fetched successfully.") + # Update status to fetching before starting download + source.status = 'fetching' + source.save(update_fields=['status']) - cache_file = source.get_cache_file() + # Send initial download notification + send_epg_update(source.id, "downloading", 0) - # Save raw data - with open(cache_file, 'wb') as f: - f.write(response.content) - logger.info(f"Cached EPG file saved to {cache_file}") + # 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_message = f"EPG source '{source.name}' returned 404 error - will retry on next scheduled run" + source.save(update_fields=['status', 'last_message']) + # 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" + } + } + ) + # 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 + + # 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}" + + # Update status to error in the database + source.status = 'error' + source.last_message = user_message + source.save(update_fields=['status', 'last_message']) + + # 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) + + # Get error details + status_code = e.response.status_code if hasattr(e, 'response') and e.response else 'unknown' + error_message = str(e) + + # Create a user-friendly message + user_message = f"EPG source '{source.name}' encountered HTTP error {status_code}" + + # Add specific handling for common HTTP errors + if status_code == 404: + user_message = f"EPG source '{source.name}' URL not found (404) - will retry on next scheduled run" + elif status_code == 401 or status_code == 403: + user_message = f"EPG source '{source.name}' access denied (HTTP {status_code}) - check credentials" + elif status_code == 429: + user_message = f"EPG source '{source.name}' rate limited (429) - try again later" + 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_message = user_message + source.save(update_fields=['status', 'last_message']) + + # 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": status_code, + "message": user_message, + "details": error_message + } + } + ) + + # 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_message = user_message + source.save(update_fields=['status', 'last_message']) + + # 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_message = f"Error: {error_message}" + source.save(update_fields=['status', 'last_message']) + + # 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 + def parse_channels_only(source): file_path = source.file_path if not file_path: file_path = source.get_cache_file() - 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)} + # Send initial parsing notification + send_epg_update(source.id, "parsing_channels", 0) - # 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()) - xml_data = decompressed.decode('utf-8') - else: - with open(file_path, 'r', encoding='utf-8') as xml_file: - xml_data = xml_file.read() + 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}") - root = ET.fromstring(xml_data) - channels = root.findall('channel') + # 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']) - epgs_to_create = [] - epgs_to_update = [] + # 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_success = fetch_xmltv(source) # Store the result - 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 + # Only proceed if fetch was successful AND file exists + if not fetch_success: + logger.error(f"Failed to fetch EPG data from URL: {source.url}") + # Update status to error + source.status = 'error' + source.last_message = f"Failed to fetch EPG data from URL" + source.save(update_fields=['status', 'last_message']) + # Send error notification + send_epg_update(source.id, "parsing_channels", 100, status="error", error="Failed to fetch EPG data") + return False - display_name = channel_elem.findtext('display-name', default=tvg_id).strip() + # 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_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 + 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_message = f"No URL provided, cannot fetch EPG data" + source.save(update_fields=['status', 'last_message']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error="No URL provided") + return False - 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) + 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()) + xml_data = decompressed.decode('utf-8') else: - epgs_to_create.append(EPGData( - tvg_id=tvg_id, - name=display_name, - epg_source=source, - )) + with open(file_path, 'r', encoding='utf-8') as xml_file: + xml_data = xml_file.read() - 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"]) + # Update progress to show file read completed + send_epg_update(source.id, "parsing_channels", 25) - channel_layer = get_channel_layer() - async_to_sync(channel_layer.group_send)( - 'updates', - { - 'type': 'update', - "data": {"success": True, "type": "epg_channels"} - } - ) + 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}") + # Update status to error + source.status = 'error' + source.last_message = f"EPG file not found: {file_path}" + source.save(update_fields=['status', 'last_message']) + 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) + # Update status to error + source.status = 'error' + source.last_message = f"Error parsing EPG file: {str(e)}" + source.save(update_fields=['status', 'last_message']) + send_epg_update(source.id, "parsing_channels", 100, status="error", error=str(e)) + return False - logger.info("Finished parsing channel info.") @shared_task def parse_programs_for_tvg_id(epg_id): @@ -179,17 +534,72 @@ def parse_programs_for_tvg_id(epg_id): file_path = epg_source.file_path if not file_path: file_path = epg_source.get_cache_file() - if not os.path.exists(file_path): - fetch_xmltv(epg_source) + + # Check if the file exists + 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']) + + # Fetch new data before continuing + if epg_source.url: + logger.info(f"Fetching new EPG data from URL: {epg_source.url}") + # Properly check the return value from fetch_xmltv + fetch_success = fetch_xmltv(epg_source) + + # If fetch was not successful or the file still doesn't exist, abort + if not fetch_success: + logger.error(f"Failed to fetch EPG data, cannot parse programs for tvg_id: {epg.tvg_id}") + # Update status to error if not already set + epg_source.status = 'error' + epg_source.last_message = f"Failed to download EPG data, cannot parse programs" + epg_source.save(update_fields=['status', 'last_message']) + send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="Failed to download EPG file") + release_task_lock('parse_epg_programs', 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}") + 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 + else: + logger.error(f"No URL provided for EPG source {epg_source.name}, cannot fetch new data") + # Update status to error + epg_source.status = 'error' + epg_source.last_message = f"No URL provided, cannot fetch EPG data" + epg_source.save(update_fields=['status', 'last_message']) + send_epg_update(epg_source.id, "parsing_programs", 100, status="error", error="No URL provided") + release_task_lock('parse_epg_programs', epg_id) + return + + file_path = new_path # 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()) - xml_data = decompressed.decode('utf-8') - else: - with open(file_path, 'r', encoding='utf-8') as xml_file: - xml_data = xml_file.read() + try: + if file_path.endswith('.gz'): + with open(file_path, 'rb') as gz_file: + decompressed = gzip.decompress(gz_file.read()) + xml_data = decompressed.decode('utf-8') + else: + with open(file_path, 'r', encoding='utf-8') as xml_file: + xml_data = xml_file.read() + except FileNotFoundError: + logger.error(f"EPG file not found at: {file_path}") + release_task_lock('parse_epg_programs', epg_id) + return + except Exception as e: + logger.error(f"Error reading EPG file {file_path}: {e}", exc_info=True) + release_task_lock('parse_epg_programs', epg_id) + return root = ET.fromstring(xml_data) @@ -302,12 +712,85 @@ 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 = [] + program_count = 0 + channel_count = 0 + updated_count = 0 + + 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 there were failures, include them in the message but continue + if failed_entries: + epg_source.status = EPGSource.STATUS_SUCCESS # Still mark as success if some processed + error_summary = f"Failed to parse {len(failed_entries)} of {total_entries} entries" + stats_summary = f"Processed {program_count} programs across {channel_count} channels. Updated: {updated_count}." + epg_source.last_message = f"{stats_summary} Warning: {error_summary}" + epg_source.updated_at = timezone.now() + epg_source.save(update_fields=['status', 'last_message', 'updated_at']) + + # Send completion notification with mixed status + send_epg_update(epg_source.id, "parsing_programs", 100, + status="success", + message=epg_source.last_message) + return True + + # If all successful, set a comprehensive success message + epg_source.status = EPGSource.STATUS_SUCCESS + epg_source.last_message = f"Successfully processed {program_count} programs across {channel_count} channels. Updated: {updated_count}." + epg_source.updated_at = timezone.now() + epg_source.save(update_fields=['status', 'last_message', 'updated_at']) + + # Send completion notification with status + send_epg_update(epg_source.id, "parsing_programs", 100, + status="success", + message=epg_source.last_message) + + 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 = EPGSource.STATUS_ERROR + epg_source.last_message = f"Error parsing programs: {str(e)}" + epg_source.save(update_fields=['status', 'last_message']) + send_epg_update(epg_source.id, "parsing_programs", 100, + status="error", + message=epg_source.last_message) + return False + def fetch_schedules_direct(source): logger.info(f"Fetching Schedules Direct data from source: {source.name}") diff --git a/apps/m3u/api_views.py b/apps/m3u/api_views.py index 054bdaa9..6176a0ca 100644 --- a/apps/m3u/api_views.py +++ b/apps/m3u/api_views.py @@ -10,6 +10,7 @@ from django.core.cache import cache import os from rest_framework.decorators import action from django.conf import settings +from .tasks import refresh_m3u_groups # Import all models, including UserAgent. from .models import M3UAccount, M3UFilter, ServerGroup, M3UAccountProfile @@ -56,6 +57,10 @@ class M3UAccountViewSet(viewsets.ModelViewSet): # Now call super().create() to create the instance response = super().create(request, *args, **kwargs) + print(response.data.get('account_type')) + if response.data.get('account_type') == M3UAccount.Types.XC: + refresh_m3u_groups(response.data.get('id')) + # After the instance is created, return the response return response @@ -89,6 +94,21 @@ class M3UAccountViewSet(viewsets.ModelViewSet): # After the instance is created, return the response return response + def partial_update(self, request, *args, **kwargs): + """Handle partial updates with special logic for is_active field""" + instance = self.get_object() + + # Check if we're toggling is_active + if 'is_active' in request.data and instance.is_active != request.data['is_active']: + # Set appropriate status based on new is_active value + if request.data['is_active']: + request.data['status'] = M3UAccount.Status.IDLE + else: + request.data['status'] = M3UAccount.Status.DISABLED + + # Continue with regular partial update + return super().partial_update(request, *args, **kwargs) + class M3UFilterViewSet(viewsets.ModelViewSet): """Handles CRUD operations for M3U filters""" queryset = M3UFilter.objects.all() diff --git a/apps/m3u/migrations/0008_m3uaccount_stale_stream_days.py b/apps/m3u/migrations/0008_m3uaccount_stale_stream_days.py new file mode 100644 index 00000000..69a1397d --- /dev/null +++ b/apps/m3u/migrations/0008_m3uaccount_stale_stream_days.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.6 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('m3u', '0007_remove_m3uaccount_uploaded_file_m3uaccount_file_path'), + ] + + operations = [ + migrations.AddField( + model_name='m3uaccount', + name='stale_stream_days', + field=models.PositiveIntegerField(default=7, help_text='Number of days after which a stream will be removed if not seen in the M3U source.'), + ), + ] diff --git a/apps/m3u/migrations/0009_m3uaccount_account_type_m3uaccount_password_and_more.py b/apps/m3u/migrations/0009_m3uaccount_account_type_m3uaccount_password_and_more.py new file mode 100644 index 00000000..d57f7ccd --- /dev/null +++ b/apps/m3u/migrations/0009_m3uaccount_account_type_m3uaccount_password_and_more.py @@ -0,0 +1,28 @@ +# Generated by Django 5.1.6 on 2025-04-27 12:56 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('m3u', '0008_m3uaccount_stale_stream_days'), + ] + + operations = [ + migrations.AddField( + model_name='m3uaccount', + name='account_type', + field=models.CharField(choices=[('STD', 'Standard'), ('XC', 'Xtream Codes')], default='STD'), + ), + migrations.AddField( + model_name='m3uaccount', + name='password', + field=models.CharField(blank=True, max_length=255, null=True), + ), + migrations.AddField( + model_name='m3uaccount', + name='username', + field=models.CharField(blank=True, max_length=255, null=True), + ), + ] diff --git a/apps/m3u/migrations/0010_add_status_fields_and_remove_auto_now.py b/apps/m3u/migrations/0010_add_status_fields_and_remove_auto_now.py new file mode 100644 index 00000000..e9896041 --- /dev/null +++ b/apps/m3u/migrations/0010_add_status_fields_and_remove_auto_now.py @@ -0,0 +1,28 @@ +# Generated by Django 5.1.6 on 2025-05-04 21:43 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('m3u', '0009_m3uaccount_account_type_m3uaccount_password_and_more'), + ] + + operations = [ + migrations.AddField( + model_name='m3uaccount', + name='last_message', + field=models.TextField(blank=True, null=True, help_text="Last status message, including success results or error information"), + ), + migrations.AddField( + model_name='m3uaccount', + name='status', + field=models.CharField(choices=[('idle', 'Idle'), ('fetching', 'Fetching'), ('parsing', 'Parsing'), ('error', 'Error'), ('success', 'Success')], default='idle', max_length=20), + ), + migrations.AlterField( + model_name='m3uaccount', + name='updated_at', + field=models.DateTimeField(blank=True, help_text='Time when this account was last successfully refreshed', null=True), + ), + ] diff --git a/apps/m3u/models.py b/apps/m3u/models.py index 25a332c6..503ac3da 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -10,6 +10,19 @@ from core.models import CoreSettings, UserAgent CUSTOM_M3U_ACCOUNT_NAME="custom" class M3UAccount(models.Model): + class Types(models.TextChoices): + STADNARD = "STD", "Standard" + XC = "XC", "Xtream Codes" + + class Status(models.TextChoices): + IDLE = "idle", "Idle" + FETCHING = "fetching", "Fetching" + PARSING = "parsing", "Parsing" + ERROR = "error", "Error" + SUCCESS = "success", "Success" + PENDING_SETUP = "pending_setup", "Pending Setup" + DISABLED = "disabled", "Disabled" + """Represents an M3U Account for IPTV streams.""" name = models.CharField( max_length=255, @@ -47,8 +60,18 @@ class M3UAccount(models.Model): help_text="Time when this account was created" ) updated_at = models.DateTimeField( - auto_now=True, - help_text="Time when this account was last updated" + null=True, blank=True, + help_text="Time when this account was last successfully refreshed" + ) + status = models.CharField( + max_length=20, + choices=Status.choices, + default=Status.IDLE + ) + last_message = models.TextField( + null=True, + blank=True, + help_text="Last status message, including success results or error information" ) user_agent = models.ForeignKey( 'core.UserAgent', @@ -69,11 +92,18 @@ class M3UAccount(models.Model): blank=True, related_name='m3u_accounts' ) + account_type = models.CharField(choices=Types.choices, default=Types.STADNARD) + username = models.CharField(max_length=255, null=True, blank=True) + password = models.CharField(max_length=255, null=True, blank=True) custom_properties = models.TextField(null=True, blank=True) refresh_interval = models.IntegerField(default=24) refresh_task = models.ForeignKey( PeriodicTask, on_delete=models.SET_NULL, null=True, blank=True ) + stale_stream_days = models.PositiveIntegerField( + default=7, + help_text="Number of days after which a stream will be removed if not seen in the M3U source." + ) def __str__(self): return self.name @@ -108,6 +138,15 @@ class M3UAccount(models.Model): return user_agent + def save(self, *args, **kwargs): + # Prevent auto_now behavior by handling updated_at manually + if 'update_fields' in kwargs and 'updated_at' not in kwargs['update_fields']: + # Don't modify updated_at for regular updates + kwargs.setdefault('update_fields', []) + if 'updated_at' in kwargs['update_fields']: + kwargs['update_fields'].remove('updated_at') + super().save(*args, **kwargs) + # def get_channel_groups(self): # return ChannelGroup.objects.filter(m3u_account__m3u_account=self) diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index d79b0117..038af628 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -66,8 +66,15 @@ class M3UAccountSerializer(serializers.ModelSerializer): fields = [ 'id', 'name', 'server_url', 'file_path', 'server_group', 'max_streams', 'is_active', 'created_at', 'updated_at', 'filters', 'user_agent', 'profiles', 'locked', - 'channel_groups', 'refresh_interval' + 'channel_groups', 'refresh_interval', 'custom_properties', 'account_type', 'username', 'password', 'stale_stream_days', + 'status', 'last_message', ] + extra_kwargs = { + 'password': { + 'required': False, + 'allow_blank': True, + }, + } def update(self, instance, validated_data): # Pop out channel group memberships so we can handle them manually diff --git a/apps/m3u/signals.py b/apps/m3u/signals.py index 6e46a0ff..e5951ff2 100644 --- a/apps/m3u/signals.py +++ b/apps/m3u/signals.py @@ -1,5 +1,5 @@ # apps/m3u/signals.py -from django.db.models.signals import post_save, post_delete +from django.db.models.signals import post_save, post_delete, pre_save from django.dispatch import receiver from .models import M3UAccount from .tasks import refresh_single_m3u_account, refresh_m3u_groups @@ -13,7 +13,7 @@ def refresh_account_on_save(sender, instance, created, **kwargs): call a Celery task that fetches & parses that single account if it is active or newly created. """ - if created: + if created and instance.account_type != M3UAccount.Types.XC: refresh_m3u_groups.delay(instance.id) @receiver(post_save, sender=M3UAccount) @@ -28,17 +28,10 @@ def create_or_update_refresh_task(sender, instance, **kwargs): period=IntervalSchedule.HOURS ) - if not instance.refresh_task: - refresh_task = PeriodicTask.objects.create( - name=task_name, - interval=interval, - task="apps.m3u.tasks.refresh_single_m3u_account", - kwargs=json.dumps({"account_id": instance.id}), - enabled=instance.refresh_interval != 0, - ) - M3UAccount.objects.filter(id=instance.id).update(refresh_task=refresh_task) - else: - task = instance.refresh_task + # First check if the task already exists to avoid validation errors + try: + task = PeriodicTask.objects.get(name=task_name) + # Task exists, just update it updated_fields = [] if task.enabled != (instance.refresh_interval != 0): @@ -52,6 +45,21 @@ def create_or_update_refresh_task(sender, instance, **kwargs): if updated_fields: task.save(update_fields=updated_fields) + # Ensure instance has the task + if instance.refresh_task_id != task.id: + M3UAccount.objects.filter(id=instance.id).update(refresh_task=task) + + except PeriodicTask.DoesNotExist: + # Create new task if it doesn't exist + refresh_task = PeriodicTask.objects.create( + name=task_name, + interval=interval, + task="apps.m3u.tasks.refresh_single_m3u_account", + kwargs=json.dumps({"account_id": instance.id}), + enabled=instance.refresh_interval != 0, + ) + M3UAccount.objects.filter(id=instance.id).update(refresh_task=refresh_task) + @receiver(post_delete, sender=M3UAccount) def delete_refresh_task(sender, instance, **kwargs): """ @@ -60,3 +68,25 @@ def delete_refresh_task(sender, instance, **kwargs): if instance.refresh_task: instance.refresh_task.interval.delete() instance.refresh_task.delete() + +@receiver(pre_save, sender=M3UAccount) +def update_status_on_active_change(sender, instance, **kwargs): + """ + When an M3UAccount's is_active field changes, update the status accordingly. + """ + if instance.pk: # Only for existing records, not new ones + try: + # Get the current record from the database + old_instance = M3UAccount.objects.get(pk=instance.pk) + + # If is_active changed, update the status + if old_instance.is_active != instance.is_active: + if instance.is_active: + # When activating, set status to idle + instance.status = M3UAccount.Status.IDLE + else: + # When deactivating, set status to disabled + instance.status = M3UAccount.Status.DISABLED + except M3UAccount.DoesNotExist: + # New record, will use default status + pass diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index c6393123..8fb16ba5 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -19,8 +19,9 @@ from django.utils import timezone import time import json from core.utils import RedisClient, acquire_task_lock, release_task_lock -from core.models import CoreSettings +from core.models import CoreSettings, UserAgent from asgiref.sync import async_to_sync +from core.xtream_codes import Client as XCClient logger = logging.getLogger(__name__) @@ -44,6 +45,11 @@ def fetch_m3u_lines(account, use_cache=False): headers = {"User-Agent": user_agent} logger.info(f"Fetching from URL {account.server_url}") + # Set account status to FETCHING before starting download + account.status = M3UAccount.Status.FETCHING + account.last_message = "Starting download..." + account.save(update_fields=['status', 'last_message']) + response = requests.get(account.server_url, headers=headers, stream=True) response.raise_for_status() @@ -70,47 +76,101 @@ def fetch_m3u_lines(account, use_cache=False): progress = (downloaded / total_size) * 100 # Time remaining (in seconds) - time_remaining = (total_size - downloaded) / (speed * 1024) + time_remaining = (total_size - downloaded) / (speed * 1024) if speed > 0 else 0 current_time = time.time() if current_time - last_update_time >= 0.5: last_update_time = current_time if progress > 0: - send_m3u_update(account.id, "downloading", progress, speed=speed, elapsed_time=elapsed_time, time_remaining=time_remaining) + # Update the account's last_message with detailed progress info + progress_msg = f"Downloading: {progress:.1f}% - {speed:.1f} KB/s - {time_remaining:.1f}s remaining" + account.last_message = progress_msg + account.save(update_fields=['last_message']) - send_m3u_update(account.id, "downloading", 100) + send_m3u_update(account.id, "downloading", progress, + speed=speed, + elapsed_time=elapsed_time, + time_remaining=time_remaining, + message=progress_msg) + + # Final update with 100% progress + final_msg = f"Download complete. Size: {total_size/1024/1024:.2f} MB, Time: {time.time() - start_time:.1f}s" + account.last_message = final_msg + account.save(update_fields=['last_message']) + send_m3u_update(account.id, "downloading", 100, message=final_msg) except Exception as e: logger.error(f"Error fetching M3U from URL {account.server_url}: {e}") - return [] + # Update account status and send error notification + account.status = M3UAccount.Status.ERROR + account.last_message = f"Error downloading M3U file: {str(e)}" + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account.id, "downloading", 100, status="error", error=f"Error downloading M3U file: {str(e)}") + return [], False # Return empty list and False for success + + # Check if the file exists and is not empty + if not os.path.exists(file_path) or os.path.getsize(file_path) == 0: + error_msg = f"M3U file not found or empty: {file_path}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + return [], False # Return empty list and False for success + + try: + with open(file_path, 'r', encoding='utf-8') as f: + return f.readlines(), True + except Exception as e: + error_msg = f"Error reading M3U file: {str(e)}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + return [], False - with open(file_path, 'r', encoding='utf-8') as f: - return f.readlines() elif account.file_path: try: if account.file_path.endswith('.gz'): with gzip.open(account.file_path, 'rt', encoding='utf-8') as f: - return f.readlines() + return f.readlines(), True elif account.file_path.endswith('.zip'): with zipfile.ZipFile(account.file_path, 'r') as zip_file: for name in zip_file.namelist(): if name.endswith('.m3u'): with zip_file.open(name) as f: - return [line.decode('utf-8') for line in f.readlines()] - logger.warning(f"No .m3u file found in ZIP archive: {account.file_path}") - return [] + return [line.decode('utf-8') for line in f.readlines()], True + + error_msg = f"No .m3u file found in ZIP archive: {account.file_path}" + logger.warning(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + return [], False else: with open(account.file_path, 'r', encoding='utf-8') as f: - return f.readlines() + return f.readlines(), True except (IOError, OSError, zipfile.BadZipFile, gzip.BadGzipFile) as e: - logger.error(f"Error opening file {account.file_path}: {e}") - return [] + error_msg = f"Error opening file {account.file_path}: {e}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + return [], False - - # Return an empty list if neither server_url nor uploaded_file is available - return [] + # Neither server_url nor uploaded_file is available + error_msg = "No M3U source available (missing URL and file)" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account.id, "downloading", 100, status="error", error=error_msg) + return [], False def parse_extinf_line(line: str) -> dict: """ @@ -177,32 +237,36 @@ def check_field_lengths(streams_to_create): print("") @shared_task -def process_groups(account, group_names): - existing_groups = {group.name: group for group in ChannelGroup.objects.filter(name__in=group_names)} +def process_groups(account, groups): + existing_groups = {group.name: group for group in ChannelGroup.objects.filter(name__in=groups.keys())} logger.info(f"Currently {len(existing_groups)} existing groups") - groups = [] + group_objs = [] groups_to_create = [] - for group_name in group_names: + for group_name, custom_props in groups.items(): logger.info(f"Handling group: {group_name}") - if group_name in existing_groups: - groups.append(existing_groups[group_name]) - else: + if (group_name not in existing_groups) and (group_name not in SKIP_EXTS): groups_to_create.append(ChannelGroup( name=group_name, )) + else: + group_objs.append(existing_groups[group_name]) if groups_to_create: logger.info(f"Creating {len(groups_to_create)} groups") created = ChannelGroup.bulk_create_and_fetch(groups_to_create) logger.info(f"Created {len(created)} groups") - groups.extend(created) + group_objs.extend(created) relations = [] - for group in groups: + for group in group_objs: + # Ensure we include the xc_id in the custom_properties + custom_props = groups.get(group.name, {}) relations.append(ChannelGroupM3UAccount( channel_group=group, m3u_account=account, + custom_properties=json.dumps(custom_props), + enabled=True, # Default to enabled )) ChannelGroupM3UAccount.objects.bulk_create( @@ -210,6 +274,132 @@ def process_groups(account, group_names): ignore_conflicts=True ) +@shared_task +def process_xc_category(account_id, batch, groups, hash_keys): + account = M3UAccount.objects.get(id=account_id) + + streams_to_create = [] + streams_to_update = [] + stream_hashes = {} + + try: + xc_client = XCClient(account.server_url, account.username, account.password, account.get_user_agent()) + + # Log the batch details to help with debugging + logger.debug(f"Processing XC batch: {batch}") + + for group_name, props in batch.items(): + # Check if we have a valid xc_id for this group + if 'xc_id' not in props: + logger.error(f"Missing xc_id for group {group_name} in batch {batch}") + continue + + # Get actual group ID from the mapping + group_id = groups.get(group_name) + if not group_id: + logger.error(f"Group {group_name} not found in enabled groups") + continue + + try: + logger.info(f"Fetching streams for XC category: {group_name} (ID: {props['xc_id']})") + streams = xc_client.get_live_category_streams(props['xc_id']) + + if not streams: + logger.warning(f"No streams found for XC category {group_name} (ID: {props['xc_id']})") + continue + + logger.info(f"Found {len(streams)} streams for category {group_name}") + + for stream in streams: + name = stream["name"] + url = xc_client.get_stream_url(stream["stream_id"]) + tvg_id = stream.get("epg_channel_id", "") + tvg_logo = stream.get("stream_icon", "") + group_title = group_name + + stream_hash = Stream.generate_hash_key(name, url, tvg_id, hash_keys) + stream_props = { + "name": name, + "url": url, + "logo_url": tvg_logo, + "tvg_id": tvg_id, + "m3u_account": account, + "channel_group_id": int(group_id), + "stream_hash": stream_hash, + "custom_properties": json.dumps(stream), + } + + if stream_hash not in stream_hashes: + stream_hashes[stream_hash] = stream_props + except Exception as e: + logger.error(f"Error processing XC category {group_name} (ID: {props['xc_id']}): {str(e)}") + continue + + # Process all found streams + existing_streams = {s.stream_hash: s for s in Stream.objects.filter(stream_hash__in=stream_hashes.keys())} + + for stream_hash, stream_props in stream_hashes.items(): + if stream_hash in existing_streams: + obj = existing_streams[stream_hash] + existing_attr = {field.name: getattr(obj, field.name) for field in Stream._meta.fields if field != 'channel_group_id'} + changed = any(existing_attr[key] != value for key, value in stream_props.items() if key != 'channel_group_id') + + if changed: + for key, value in stream_props.items(): + setattr(obj, key, value) + obj.last_seen = timezone.now() + obj.updated_at = timezone.now() # Update timestamp only for changed streams + streams_to_update.append(obj) + del existing_streams[stream_hash] + else: + # Always update last_seen, even if nothing else changed + obj.last_seen = timezone.now() + # Don't update updated_at for unchanged streams + streams_to_update.append(obj) + existing_streams[stream_hash] = obj + else: + stream_props["last_seen"] = timezone.now() + stream_props["updated_at"] = timezone.now() # Set initial updated_at for new streams + streams_to_create.append(Stream(**stream_props)) + + try: + with transaction.atomic(): + if streams_to_create: + Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True) + if streams_to_update: + # We need to split the bulk update to correctly handle updated_at + # First, get the subset of streams that have content changes + changed_streams = [s for s in streams_to_update if hasattr(s, 'updated_at') and s.updated_at] + unchanged_streams = [s for s in streams_to_update if not hasattr(s, 'updated_at') or not s.updated_at] + + # Update changed streams with all fields including updated_at + if changed_streams: + Stream.objects.bulk_update( + changed_streams, + {key for key in stream_props.keys() if key not in ["m3u_account", "stream_hash"] and key not in hash_keys} | {"last_seen", "updated_at"} + ) + + # Update unchanged streams with only last_seen + if unchanged_streams: + Stream.objects.bulk_update(unchanged_streams, ["last_seen"]) + + if len(existing_streams.keys()) > 0: + Stream.objects.bulk_update(existing_streams.values(), ["last_seen"]) + except Exception as e: + logger.error(f"Bulk create failed for XC streams: {str(e)}") + + retval = f"Batch processed: {len(streams_to_create)} created, {len(streams_to_update)} updated." + + except Exception as e: + logger.error(f"XC category processing error: {str(e)}") + retval = f"Error processing XC batch: {str(e)}" + + # Aggressive garbage collection + del streams_to_create, streams_to_update, stream_hashes, existing_streams + gc.collect() + + return retval + @shared_task def process_m3u_batch(account_id, batch, groups, hash_keys): """Processes a batch of M3U streams using bulk operations.""" @@ -232,23 +422,7 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): logger.debug(f"Skipping stream in disabled group: {group_title}") continue - # if any(url.lower().endswith(ext) for ext in SKIP_EXTS) or len(url) > 2000: - # continue - - # if _matches_filters(name, group_title, account.filters.all()): - # continue - - # if any(compiled_pattern.search(current_info['name']) for ftype, compiled_pattern in compiled_filters if ftype == 'name'): - # excluded_count += 1 - # current_info = None - # continue - stream_hash = Stream.generate_hash_key(name, url, tvg_id, hash_keys) - # if redis_client.exists(f"m3u_refresh:{stream_hash}"): - # # duplicate already processed by another batch - # continue - - # redis_client.set(f"m3u_refresh:{stream_hash}", "true") stream_props = { "name": name, "url": url, @@ -278,12 +452,18 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): for key, value in stream_props.items(): setattr(obj, key, value) obj.last_seen = timezone.now() + obj.updated_at = timezone.now() # Update timestamp only for changed streams streams_to_update.append(obj) del existing_streams[stream_hash] else: + # Always update last_seen, even if nothing else changed + obj.last_seen = timezone.now() + # Don't update updated_at for unchanged streams + streams_to_update.append(obj) existing_streams[stream_hash] = obj else: stream_props["last_seen"] = timezone.now() + stream_props["updated_at"] = timezone.now() # Set initial updated_at for new streams streams_to_create.append(Stream(**stream_props)) try: @@ -291,9 +471,24 @@ def process_m3u_batch(account_id, batch, groups, hash_keys): if streams_to_create: Stream.objects.bulk_create(streams_to_create, ignore_conflicts=True) if streams_to_update: - Stream.objects.bulk_update(streams_to_update, { key for key in stream_props.keys() if key not in ["m3u_account", "stream_hash"] and key not in hash_keys}) - # if len(existing_streams.keys()) > 0: - # Stream.objects.bulk_update(existing_streams.values(), ["last_seen"]) + # We need to split the bulk update to correctly handle updated_at + # First, get the subset of streams that have content changes + changed_streams = [s for s in streams_to_update if hasattr(s, 'updated_at') and s.updated_at] + unchanged_streams = [s for s in streams_to_update if not hasattr(s, 'updated_at') or not s.updated_at] + + # Update changed streams with all fields including updated_at + if changed_streams: + Stream.objects.bulk_update( + changed_streams, + {key for key in stream_props.keys() if key not in ["m3u_account", "stream_hash"] and key not in hash_keys} | {"last_seen", "updated_at"} + ) + + # Update unchanged streams with only last_seen + if unchanged_streams: + Stream.objects.bulk_update(unchanged_streams, ["last_seen"]) + + if len(existing_streams.keys()) > 0: + Stream.objects.bulk_update(existing_streams.values(), ["last_seen"]) except Exception as e: logger.error(f"Bulk create failed: {str(e)}") @@ -312,18 +507,31 @@ def cleanup_streams(account_id): m3u_account__enabled=True, ).values_list('id', flat=True) logger.info(f"Found {len(existing_groups)} active groups") - streams = Stream.objects.filter(m3u_account=account) + # Calculate cutoff date for stale streams + stale_cutoff = timezone.now() - timezone.timedelta(days=account.stale_stream_days) + logger.info(f"Removing streams not seen since {stale_cutoff}") + + # Delete streams that are not in active groups streams_to_delete = Stream.objects.filter( m3u_account=account ).exclude( - channel_group__in=existing_groups # Exclude products having any of the excluded tags + channel_group__in=existing_groups ) - # Delete the filtered products - streams_to_delete.delete() + # Also delete streams that haven't been seen for longer than stale_stream_days + stale_streams = Stream.objects.filter( + m3u_account=account, + last_seen__lt=stale_cutoff + ) - logger.info(f"Cleanup complete") + deleted_count = streams_to_delete.count() + stale_count = stale_streams.count() + + streams_to_delete.delete() + stale_streams.delete() + + logger.info(f"Cleanup complete: {deleted_count} streams removed due to group filter, {stale_count} removed as stale") @shared_task def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): @@ -337,46 +545,191 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False): return f"M3UAccount with ID={account_id} not found or inactive.", None extinf_data = [] - groups = set(["Default Group"]) + groups = {"Default Group": {}} - for line in fetch_m3u_lines(account, use_cache): - line = line.strip() - if line.startswith("#EXTINF"): - parsed = parse_extinf_line(line) - if parsed: - if "group-title" in parsed["attributes"]: - groups.add(parsed["attributes"]["group-title"]) + if account.account_type == M3UAccount.Types.XC: + # Log detailed information about the account + logger.info(f"Processing XC account {account_id} with URL: {account.server_url}") + logger.info(f"Username: {account.username}, Has password: {'Yes' if account.password else 'No'}") - extinf_data.append(parsed) - elif extinf_data and line.startswith("http"): - # Associate URL with the last EXTINF line - extinf_data[-1]["url"] = line + # Validate required fields + if not account.server_url: + error_msg = "Missing server URL for Xtream Codes account" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + + if not account.username or not account.password: + error_msg = "Missing username or password for Xtream Codes account" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + + try: + # Ensure server URL is properly formatted + server_url = account.server_url.rstrip('/') + if not (server_url.startswith('http://') or server_url.startswith('https://')): + server_url = f"http://{server_url}" + + # User agent handling - completely rewritten + try: + # Debug the user agent issue + logger.info(f"Getting user agent for account {account.id}") + + # Use a hardcoded user agent string to avoid any issues with object structure + user_agent_string = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + + try: + # Try to get the user agent directly from the database + if account.user_agent_id: + ua_obj = UserAgent.objects.get(id=account.user_agent_id) + if ua_obj and hasattr(ua_obj, 'user_agent') and ua_obj.user_agent: + user_agent_string = ua_obj.user_agent + logger.info(f"Using user agent from account: {user_agent_string}") + else: + # Get default user agent from CoreSettings + default_ua_id = CoreSettings.get_default_user_agent_id() + logger.info(f"Default user agent ID from settings: {default_ua_id}") + if default_ua_id: + ua_obj = UserAgent.objects.get(id=default_ua_id) + if ua_obj and hasattr(ua_obj, 'user_agent') and ua_obj.user_agent: + user_agent_string = ua_obj.user_agent + logger.info(f"Using default user agent: {user_agent_string}") + except Exception as e: + logger.warning(f"Error getting user agent, using fallback: {str(e)}") + + logger.info(f"Final user agent string: {user_agent_string}") + except Exception as e: + user_agent_string = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" + logger.warning(f"Exception in user agent handling, using fallback: {str(e)}") + + logger.info(f"Creating XCClient with URL: {server_url}, Username: {account.username}, User-Agent: {user_agent_string}") + + # Create XCClient with explicit error handling + try: + xc_client = XCClient(server_url, account.username, account.password, user_agent_string) + logger.info(f"XCClient instance created successfully") + except Exception as e: + error_msg = f"Failed to create XCClient: {str(e)}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + + # Authenticate with detailed error handling + try: + logger.info(f"Authenticating with XC server {server_url}") + auth_result = xc_client.authenticate() + logger.info(f"Authentication response: {auth_result}") + except Exception as e: + error_msg = f"Failed to authenticate with XC server: {str(e)}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + + # Get categories with detailed error handling + try: + logger.info(f"Getting live categories from XC server") + xc_categories = xc_client.get_live_categories() + logger.info(f"Found {len(xc_categories)} categories: {xc_categories}") + + # Validate response + if not isinstance(xc_categories, list): + error_msg = f"Unexpected response from XC server: {xc_categories}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + + if len(xc_categories) == 0: + logger.warning("No categories found in XC server response") + + for category in xc_categories: + cat_name = category.get("category_name", "Unknown Category") + cat_id = category.get("category_id", "0") + logger.info(f"Adding category: {cat_name} (ID: {cat_id})") + groups[cat_name] = { + "xc_id": cat_id, + } + except Exception as e: + error_msg = f"Failed to get categories from XC server: {str(e)}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + except Exception as e: + error_msg = f"Unexpected error in XC processing: {str(e)}" + logger.error(error_msg) + account.status = M3UAccount.Status.ERROR + account.last_message = error_msg + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "processing_groups", 100, status="error", error=error_msg) + release_task_lock('refresh_m3u_account_groups', account_id) + return error_msg, None + else: + # Here's the key change - use the success flag from fetch_m3u_lines + lines, success = fetch_m3u_lines(account, use_cache) + if not success: + # If fetch failed, don't continue processing + release_task_lock('refresh_m3u_account_groups', account_id) + return f"Failed to fetch M3U data for account_id={account_id}.", None + + for line in lines: + line = line.strip() + if line.startswith("#EXTINF"): + parsed = parse_extinf_line(line) + if parsed: + if "group-title" in parsed["attributes"]: + groups[parsed["attributes"]["group-title"]] = {} + + extinf_data.append(parsed) + elif extinf_data and line.startswith("http"): + # Associate URL with the last EXTINF line + extinf_data[-1]["url"] = line + + cache_path = os.path.join(m3u_dir, f"{account_id}.json") + with open(cache_path, 'w', encoding='utf-8') as f: + json.dump({ + "extinf_data": extinf_data, + "groups": groups, + }, f) send_m3u_update(account_id, "processing_groups", 0) - groups = list(groups) - cache_path = os.path.join(m3u_dir, f"{account_id}.json") - with open(cache_path, 'w', encoding='utf-8') as f: - json.dump({ - "extinf_data": extinf_data, - "groups": groups, - }, f) - process_groups(account, groups) release_task_lock('refresh_m3u_account_groups', account_id) - send_m3u_update(account_id, "processing_groups", 100) + if not full_refresh: - channel_layer = get_channel_layer() - async_to_sync(channel_layer.group_send)( - 'updates', - { - 'type': 'update', - "data": {"success": True, "type": "m3u_group_refresh", "account": account_id} - } + # Use update() instead of save() to avoid triggering signals + M3UAccount.objects.filter(id=account_id).update( + status=M3UAccount.Status.PENDING_SETUP, + last_message="M3U groups loaded. Please select groups or refresh M3U to complete setup." ) + send_m3u_update(account_id, "processing_groups", 100, status="pending_setup", message="M3U groups loaded. Please select groups or refresh M3U to complete setup.") return extinf_data, groups @@ -386,23 +739,29 @@ def refresh_single_m3u_account(account_id): if not acquire_task_lock('refresh_single_m3u_account', account_id): return f"Task already running for account_id={account_id}." - # redis_client = RedisClient.get_client() # Record start time start_time = time.time() + streams_created = 0 + streams_updated = 0 + streams_deleted = 0 try: account = M3UAccount.objects.get(id=account_id, is_active=True) if not account.is_active: logger.info(f"Account {account_id} is not active, skipping.") + release_task_lock('refresh_single_m3u_account', account_id) return + # Set status to fetching + account.status = M3UAccount.Status.FETCHING + account.save(update_fields=['status']) + filters = list(account.filters.all()) except M3UAccount.DoesNotExist: release_task_lock('refresh_single_m3u_account', account_id) return f"M3UAccount with ID={account_id} not found or inactive." # Fetch M3U lines and handle potential issues - # lines = fetch_m3u_lines(account) # Extracted fetch logic into separate function extinf_data = [] groups = None @@ -416,14 +775,58 @@ def refresh_single_m3u_account(account_id): if not extinf_data: try: - extinf_data, groups = refresh_m3u_groups(account_id, full_refresh=True) - if not extinf_data or not groups: + logger.info(f"Calling refresh_m3u_groups for account {account_id}") + result = refresh_m3u_groups(account_id, full_refresh=True) + logger.info(f"refresh_m3u_groups result: {result}") + + # Check for completely empty result or missing groups + if not result or result[1] is None: + logger.error(f"Failed to refresh M3U groups for account {account_id}: {result}") release_task_lock('refresh_single_m3u_account', account_id) - return "Failed to update m3u account, task may already be running" - except: + return "Failed to update m3u account - download failed or other error" + + extinf_data, groups = result + + # XC accounts can have empty extinf_data but valid groups + try: + account = M3UAccount.objects.get(id=account_id) + is_xc_account = account.account_type == M3UAccount.Types.XC + except M3UAccount.DoesNotExist: + is_xc_account = False + + # For XC accounts, empty extinf_data is normal at this stage + if not extinf_data and not is_xc_account: + logger.error(f"No streams found for non-XC account {account_id}") + account.status = M3UAccount.Status.ERROR + account.last_message = "No streams found in M3U source" + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "parsing", 100, status="error", error="No streams found") + except Exception as e: + logger.error(f"Exception in refresh_m3u_groups: {str(e)}", exc_info=True) + account.status = M3UAccount.Status.ERROR + account.last_message = f"Error refreshing M3U groups: {str(e)}" + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "parsing", 100, status="error", error=f"Error refreshing M3U groups: {str(e)}") release_task_lock('refresh_single_m3u_account', account_id) return "Failed to update m3u account" + # Only proceed with parsing if we actually have data and no errors were encountered + # Get account type to handle XC accounts differently + try: + is_xc_account = account.account_type == M3UAccount.Types.XC + except Exception: + is_xc_account = False + + # Modified validation logic for different account types + if (not groups) or (not is_xc_account and not extinf_data): + logger.error(f"No data to process for account {account_id}") + account.status = M3UAccount.Status.ERROR + account.last_message = "No data available for processing" + account.save(update_fields=['status', 'last_message']) + send_m3u_update(account_id, "parsing", 100, status="error", error="No data available for processing") + release_task_lock('refresh_single_m3u_account', account_id) + return "Failed to update m3u account, no data available" + hash_keys = CoreSettings.get_m3u_hash_key().split(",") existing_groups = {group.name: group.id for group in ChannelGroup.objects.filter( @@ -431,50 +834,166 @@ def refresh_single_m3u_account(account_id): m3u_account__enabled=True # Filter by the enabled flag in the join table )} - # Break into batches and process in parallel - batches = [extinf_data[i:i + BATCH_SIZE] for i in range(0, len(extinf_data), BATCH_SIZE)] - task_group = group(process_m3u_batch.s(account_id, batch, existing_groups, hash_keys) for batch in batches) + try: + # Set status to parsing + account.status = M3UAccount.Status.PARSING + account.save(update_fields=['status']) - total_batches = len(batches) - completed_batches = 0 - logger.debug(f"Dispatched {len(batches)} parallel tasks for account_id={account_id}.") + if account.account_type == M3UAccount.Types.STADNARD: + # Break into batches and process in parallel + batches = [extinf_data[i:i + BATCH_SIZE] for i in range(0, len(extinf_data), BATCH_SIZE)] + task_group = group(process_m3u_batch.s(account_id, batch, existing_groups, hash_keys) for batch in batches) + else: + # For XC accounts, get the groups with their custom properties containing xc_id + logger.info(f"Processing XC account with groups: {existing_groups}") - # result = task_group.apply_async() - result = task_group.apply_async() + # Get the ChannelGroupM3UAccount entries with their custom_properties + channel_group_relationships = ChannelGroupM3UAccount.objects.filter( + m3u_account=account, + enabled=True + ).select_related('channel_group') - while completed_batches < total_batches: - for async_result in result: - if async_result.ready(): # If the task has completed - task_result = async_result.result # The result of the task - logger.debug(f"Task completed with result: {task_result}") - completed_batches += 1 + filtered_groups = {} + for rel in channel_group_relationships: + group_name = rel.channel_group.name + group_id = rel.channel_group.id - # Calculate progress - progress = int((completed_batches / total_batches) * 100) + # Load the custom properties with the xc_id + try: + custom_props = json.loads(rel.custom_properties) if rel.custom_properties else {} + if 'xc_id' in custom_props: + filtered_groups[group_name] = { + 'xc_id': custom_props['xc_id'], + 'channel_group_id': group_id + } + logger.info(f"Added group {group_name} with xc_id {custom_props['xc_id']}") + else: + logger.warning(f"No xc_id found in custom properties for group {group_name}") + except (json.JSONDecodeError, KeyError) as e: + logger.error(f"Error parsing custom properties for group {group_name}: {str(e)}") - # Send progress update via Channels - # Don't send 100% because we want to clean up after - if progress == 100: - progress = 99 + logger.info(f"Filtered {len(filtered_groups)} groups for processing: {filtered_groups}") - send_m3u_update(account_id, "parsing", progress) + # Batch the groups + filtered_groups_list = list(filtered_groups.items()) + batches = [ + dict(filtered_groups_list[i:i + 2]) + for i in range(0, len(filtered_groups_list), 2) + ] - # Optionally remove completed task from the group to prevent processing it again - result.remove(async_result) - else: - logger.debug(f"Task is still running.") + logger.info(f"Created {len(batches)} batches for XC processing") + task_group = group(process_xc_category.s(account_id, batch, existing_groups, hash_keys) for batch in batches) - # Run cleanup - cleanup_streams(account_id) - send_m3u_update(account_id, "parsing", 100) + total_batches = len(batches) + completed_batches = 0 + streams_processed = 0 # Track total streams processed + logger.debug(f"Dispatched {len(batches)} parallel tasks for account_id={account_id}.") - end_time = time.time() + # result = task_group.apply_async() + result = task_group.apply_async() - # Calculate elapsed time - elapsed_time = end_time - start_time - account.save(update_fields=['updated_at']) + # Wait for all tasks to complete and collect their result IDs + completed_task_ids = set() + while completed_batches < total_batches: + for async_result in result: + if async_result.ready() and async_result.id not in completed_task_ids: # If the task has completed and we haven't counted it + task_result = async_result.result # The result of the task + logger.debug(f"Task completed with result: {task_result}") - print(f"Function took {elapsed_time} seconds to execute.") + # Extract stream counts from result string if available + if isinstance(task_result, str): + try: + created_match = re.search(r"(\d+) created", task_result) + updated_match = re.search(r"(\d+) updated", task_result) + + if created_match and updated_match: + created_count = int(created_match.group(1)) + updated_count = int(updated_match.group(1)) + streams_processed += created_count + updated_count + streams_created += created_count + streams_updated += updated_count + except (AttributeError, ValueError): + pass + + completed_batches += 1 + completed_task_ids.add(async_result.id) # Mark this task as processed + + # Calculate progress + progress = int((completed_batches / total_batches) * 100) + + # Calculate elapsed time and estimated remaining time + current_elapsed = time.time() - start_time + if progress > 0: + estimated_total = (current_elapsed / progress) * 100 + time_remaining = max(0, estimated_total - current_elapsed) + else: + time_remaining = 0 + + # Send progress update via Channels + # Don't send 100% because we want to clean up after + if progress == 100: + progress = 99 + + send_m3u_update( + account_id, + "parsing", + progress, + elapsed_time=current_elapsed, + time_remaining=time_remaining, + streams_processed=streams_processed + ) + + # Optionally remove completed task from the group to prevent processing it again + result.remove(async_result) + else: + logger.debug(f"Task is still running.") + + # Ensure all database transactions are committed before cleanup + logger.info(f"All {total_batches} tasks completed, ensuring DB transactions are committed before cleanup") + # Force a simple DB query to ensure connection sync + Stream.objects.filter(id=-1).exists() # This will never find anything but ensures DB sync + + # Now run cleanup + cleanup_streams(account_id) + + # Calculate elapsed time + elapsed_time = time.time() - start_time + + # Set status to success and update timestamp BEFORE sending the final update + account.status = M3UAccount.Status.SUCCESS + account.last_message = ( + f"Processing completed in {elapsed_time:.1f} seconds. " + f"Streams: {streams_created} created, {streams_updated} updated, {streams_deleted} removed. " + f"Total processed: {streams_processed}." + ) + account.updated_at = timezone.now() + account.save(update_fields=['status', 'last_message', 'updated_at']) + + # Send final update with complete metrics and explicitly include success status + send_m3u_update( + account_id, + "parsing", + 100, + status="success", # Explicitly set status to success + elapsed_time=elapsed_time, + time_remaining=0, + streams_processed=streams_processed, + streams_created=streams_created, + streams_updated=streams_updated, + streams_deleted=streams_deleted, + message=account.last_message + ) + + print(f"Function took {elapsed_time} seconds to execute.") + + except Exception as e: + logger.error(f"Error processing M3U for account {account_id}: {str(e)}") + account.status = M3UAccount.Status.ERROR + account.last_message = f"Error processing M3U: {str(e)}" + account.save(update_fields=['status', 'last_message']) + raise # Re-raise the exception for Celery to handle + + release_task_lock('refresh_single_m3u_account', account_id) # Aggressive garbage collection del existing_groups, extinf_data, groups, batches @@ -484,16 +1003,6 @@ def refresh_single_m3u_account(account_id): if os.path.exists(cache_path): os.remove(cache_path) - release_task_lock('refresh_single_m3u_account', account_id) - - # cursor = 0 - # while True: - # cursor, keys = redis_client.scan(cursor, match=f"m3u_refresh:*", count=BATCH_SIZE) - # if keys: - # redis_client.delete(*keys) # Delete the matching keys - # if cursor == 0: - # break - return f"Dispatched jobs complete." def send_m3u_update(account_id, action, progress, **kwargs): @@ -505,6 +1014,17 @@ def send_m3u_update(account_id, action, progress, **kwargs): "action": action, } + # Add the status and message if not already in kwargs + try: + account = M3UAccount.objects.get(id=account_id) + if account: + if "status" not in kwargs: + data["status"] = account.status + if "message" not in kwargs and account.last_message: + data["message"] = account.last_message + except: + pass # If account can't be retrieved, continue without these fields + # Add the additional key-value pairs from kwargs data.update(kwargs) diff --git a/apps/output/views.py b/apps/output/views.py index 9429d9ca..38b69bde 100644 --- a/apps/output/views.py +++ b/apps/output/views.py @@ -31,11 +31,16 @@ def generate_m3u(request, profile_name=None): if channel.logo: tvg_logo = request.build_absolute_uri(reverse('api:channels:logo-cache', args=[channel.logo.id])) + # create possible gracenote id insertion + tvc_guide_stationid = "" + if channel.tvc_guide_stationid: + tvc_guide_stationid = f'tvc-guide-stationid="{channel.tvc_guide_stationid}" ' + channel_number = channel.channel_number extinf_line = ( f'#EXTINF:-1 tvg-id="{tvg_id}" tvg-name="{tvg_name}" tvg-logo="{tvg_logo}" ' - f'tvg-chno="{channel_number}" group-title="{group_title}",{channel.name}\n' + f'tvg-chno="{channel_number}" {tvc_guide_stationid}group-title="{group_title}",{channel.name}\n' ) base_url = request.build_absolute_uri('/')[:-1] diff --git a/apps/proxy/config.py b/apps/proxy/config.py index 9d38532b..b00bd224 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -48,6 +48,8 @@ class TSConfig(BaseConfig): CHANNEL_INIT_GRACE_PERIOD = 5 # How long to wait for first client after initialization (seconds) CLIENT_HEARTBEAT_INTERVAL = 1 # How often to send client heartbeats (seconds) GHOST_CLIENT_MULTIPLIER = 5.0 # How many heartbeat intervals before client considered ghost (5 would mean 5 secondsif heartbeat interval is 1) + CLIENT_WAIT_TIMEOUT = 30 # Seconds to wait for client to connect + # TS packets are 188 bytes # Make chunk size a multiple of TS packet size for perfect alignment @@ -58,7 +60,7 @@ class TSConfig(BaseConfig): MAX_RECONNECT_ATTEMPTS = 3 # Maximum reconnects to try before switching streams MIN_STABLE_TIME_BEFORE_RECONNECT = 30 # Minimum seconds a stream must be stable to try reconnect FAILOVER_GRACE_PERIOD = 20 # Extra time (seconds) to allow for stream switching before disconnecting clients - + URL_SWITCH_TIMEOUT = 20 # Max time allowed for a stream switch operation diff --git a/apps/proxy/ts_proxy/config_helper.py b/apps/proxy/ts_proxy/config_helper.py index f78ba0b6..773ab378 100644 --- a/apps/proxy/ts_proxy/config_helper.py +++ b/apps/proxy/ts_proxy/config_helper.py @@ -34,12 +34,12 @@ class ConfigHelper: @staticmethod def channel_shutdown_delay(): """Get channel shutdown delay in seconds""" - return ConfigHelper.get('CHANNEL_SHUTDOWN_DELAY', 5) + return ConfigHelper.get('CHANNEL_SHUTDOWN_DELAY', 0) @staticmethod def initial_behind_chunks(): """Get number of chunks to start behind""" - return ConfigHelper.get('INITIAL_BEHIND_CHUNKS', 10) + return ConfigHelper.get('INITIAL_BEHIND_CHUNKS', 4) @staticmethod def keepalive_interval(): @@ -75,3 +75,13 @@ class ConfigHelper: def retry_wait_interval(): """Get wait interval between connection retries in seconds""" return ConfigHelper.get('RETRY_WAIT_INTERVAL', 0.5) # Default to 0.5 second + + @staticmethod + def url_switch_timeout(): + """Get URL switch timeout in seconds (max time allowed for a stream switch operation)""" + return ConfigHelper.get('URL_SWITCH_TIMEOUT', 20) # Default to 20 seconds + + @staticmethod + def failover_grace_period(): + """Get extra time (in seconds) to allow for stream switching before disconnecting clients""" + return ConfigHelper.get('FAILOVER_GRACE_PERIOD', 20) # Default to 20 seconds diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 82060f2f..97e02c16 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -13,6 +13,7 @@ from .utils import create_ts_packet, get_logger from .redis_keys import RedisKeys from .utils import get_logger from .constants import ChannelMetadataField +from .config_helper import ConfigHelper # Add this import logger = get_logger() @@ -95,7 +96,7 @@ class StreamGenerator: def _wait_for_initialization(self): """Wait for channel initialization to complete, sending keepalive packets.""" initialization_start = time.time() - max_init_wait = getattr(Config, 'CLIENT_WAIT_TIMEOUT', 30) + max_init_wait = ConfigHelper.client_wait_timeout() keepalive_interval = 0.5 last_keepalive = 0 proxy_server = ProxyServer.get_instance() @@ -156,7 +157,7 @@ class StreamGenerator: return False # Client state tracking - use config for initial position - initial_behind = getattr(Config, 'INITIAL_BEHIND_CHUNKS', 10) + initial_behind = ConfigHelper.initial_behind_chunks() current_buffer_index = buffer.index self.local_index = max(0, current_buffer_index - initial_behind) @@ -337,8 +338,8 @@ class StreamGenerator: def _is_timeout(self): """Check if the stream has timed out.""" # Get a more generous timeout for stream switching - stream_timeout = getattr(Config, 'STREAM_TIMEOUT', 10) - failover_grace_period = getattr(Config, 'FAILOVER_GRACE_PERIOD', 20) + stream_timeout = ConfigHelper.stream_timeout() + failover_grace_period = ConfigHelper.failover_grace_period() total_timeout = stream_timeout + failover_grace_period # Disconnect after long inactivity @@ -415,7 +416,7 @@ class StreamGenerator: def delayed_shutdown(): # Use the config setting instead of hardcoded value - shutdown_delay = getattr(Config, 'CHANNEL_SHUTDOWN_DELAY', 5) + shutdown_delay = ConfigHelper.channel_shutdown_delay() # Use ConfigHelper logger.info(f"Waiting {shutdown_delay}s before checking if channel should be stopped") gevent.sleep(shutdown_delay) # Replace time.sleep @@ -436,4 +437,4 @@ def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, Returns a function that can be passed to StreamingHttpResponse. """ generator = StreamGenerator(channel_id, client_id, client_ip, client_user_agent, channel_initializing) - return generator.generate + return generator.generate \ No newline at end of file diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 771ffba8..3bc288ac 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -37,6 +37,8 @@ class StreamManager: self.current_response = None self.current_session = None self.url_switching = False + self.url_switch_start_time = 0 + self.url_switch_timeout = ConfigHelper.url_switch_timeout() # Store worker_id for ownership checks self.worker_id = worker_id @@ -144,6 +146,13 @@ class StreamManager: # Main stream switching loop - we'll try different streams if needed while self.running and stream_switch_attempts <= max_stream_switches: + # Check for stuck switching state + if self.url_switching and time.time() - self.url_switch_start_time > self.url_switch_timeout: + logger.warning(f"URL switching state appears stuck for channel {self.channel_id} " + f"({time.time() - self.url_switch_start_time:.1f}s > {self.url_switch_timeout}s timeout). " + f"Resetting switching state.") + self._reset_url_switching_state() + # Check stream type before connecting stream_type = detect_stream_type(self.url) if self.transcode == False and stream_type == StreamType.HLS: @@ -568,44 +577,49 @@ class StreamManager: # CRITICAL: Set a flag to prevent immediate reconnection with old URL self.url_switching = True + self.url_switch_start_time = time.time() - # Check which type of connection we're using and close it properly - if self.transcode or self.socket: - logger.debug("Closing transcode process before URL change") - self._close_socket() - else: - logger.debug("Closing HTTP connection before URL change") - self._close_connection() + try: + # Check which type of connection we're using and close it properly + if self.transcode or self.socket: + logger.debug("Closing transcode process before URL change") + self._close_socket() + else: + logger.debug("Closing HTTP connection before URL change") + self._close_connection() - # Update URL and reset connection state - old_url = self.url - self.url = new_url - self.connected = False + # Update URL and reset connection state + old_url = self.url + self.url = new_url + self.connected = False - # Update stream ID if provided - if stream_id: - old_stream_id = self.current_stream_id - self.current_stream_id = stream_id - # Add stream ID to tried streams for proper tracking - self.tried_stream_ids.add(stream_id) - logger.info(f"Updated stream ID from {old_stream_id} to {stream_id} for channel {self.buffer.channel_id}") + # Update stream ID if provided + if stream_id: + old_stream_id = self.current_stream_id + self.current_stream_id = stream_id + # Add stream ID to tried streams for proper tracking + self.tried_stream_ids.add(stream_id) + logger.info(f"Updated stream ID from {old_stream_id} to {stream_id} for channel {self.buffer.channel_id}") - # Reset retry counter to allow immediate reconnect - self.retry_count = 0 + # Reset retry counter to allow immediate reconnect + self.retry_count = 0 - # Also reset buffer position to prevent stale data after URL change - if hasattr(self.buffer, 'reset_buffer_position'): - try: - self.buffer.reset_buffer_position() - logger.debug("Reset buffer position for clean URL switch") - except Exception as e: - logger.warning(f"Failed to reset buffer position: {e}") + # Also reset buffer position to prevent stale data after URL change + if hasattr(self.buffer, 'reset_buffer_position'): + try: + self.buffer.reset_buffer_position() + logger.debug("Reset buffer position for clean URL switch") + except Exception as e: + logger.warning(f"Failed to reset buffer position: {e}") - # Done with URL switch - self.url_switching = False - logger.info(f"Stream switch completed for channel {self.buffer.channel_id}") - - return True + return True + except Exception as e: + logger.error(f"Error during URL update: {e}", exc_info=True) + return False + finally: + # CRITICAL FIX: Always reset the URL switching flag when done, whether successful or not + self.url_switching = False + logger.info(f"Stream switch completed for channel {self.buffer.channel_id}") def should_retry(self) -> bool: """Check if connection retry is allowed""" @@ -684,8 +698,14 @@ class StreamManager: # Don't try to reconnect if we're already switching URLs if self.url_switching: - logger.info("URL switching already in progress, skipping reconnect") - return + # Add timeout check to prevent permanent deadlock + if time.time() - self.url_switch_start_time > self.url_switch_timeout: + logger.warning(f"URL switching has been in progress too long ({time.time() - self.url_switch_start_time:.1f}s), " + f"resetting switching state and allowing reconnect") + self._reset_url_switching_state() + else: + logger.info("URL switching already in progress, skipping reconnect") + return False # Close existing connection if self.transcode or self.socket: @@ -1060,4 +1080,11 @@ class StreamManager: except Exception as e: logger.error(f"Error trying next stream for channel {self.channel_id}: {e}", exc_info=True) - return False \ No newline at end of file + return False + + # Add a new helper method to safely reset the URL switching state + def _reset_url_switching_state(self): + """Safely reset the URL switching state if it gets stuck""" + self.url_switching = False + self.url_switch_start_time = 0 + logger.info(f"Reset URL switching state for channel {self.channel_id}") \ No newline at end of file diff --git a/core/api_views.py b/core/api_views.py index 7f3ecf57..77473b5d 100644 --- a/core/api_views.py +++ b/core/api_views.py @@ -3,7 +3,7 @@ from rest_framework import viewsets, status from rest_framework.response import Response from django.shortcuts import get_object_or_404 -from .models import UserAgent, StreamProfile, CoreSettings +from .models import UserAgent, StreamProfile, CoreSettings, STREAM_HASH_KEY from .serializers import UserAgentSerializer, StreamProfileSerializer, CoreSettingsSerializer from rest_framework.permissions import IsAuthenticated from rest_framework.decorators import api_view, permission_classes @@ -11,6 +11,7 @@ from drf_yasg.utils import swagger_auto_schema import socket import requests import os +from core.tasks import rehash_streams class UserAgentViewSet(viewsets.ModelViewSet): """ @@ -34,6 +35,15 @@ class CoreSettingsViewSet(viewsets.ModelViewSet): queryset = CoreSettings.objects.all() serializer_class = CoreSettingsSerializer + def update(self, request, *args, **kwargs): + instance = self.get_object() + response = super().update(request, *args, **kwargs) + if instance.key == STREAM_HASH_KEY: + if instance.value != request.data['value']: + rehash_streams.delay(request.data['value'].split(',')) + + return response + @swagger_auto_schema( method='get', operation_description="Endpoint for environment details", diff --git a/core/tasks.py b/core/tasks.py index 83682a69..64a89c7a 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -15,6 +15,8 @@ from apps.epg.models import EPGSource from apps.m3u.tasks import refresh_single_m3u_account from apps.epg.tasks import refresh_epg_data from .models import CoreSettings +from apps.channels.models import Stream, ChannelStream +from django.db import transaction logger = logging.getLogger(__name__) @@ -249,3 +251,35 @@ def fetch_channel_stats(): "data": {"success": True, "type": "channel_stats", "stats": json.dumps({'channels': all_channels, 'count': len(all_channels)})} }, ) + +@shared_task +def rehash_streams(keys): + batch_size = 1000 + queryset = Stream.objects.all() + + hash_keys = {} + total_records = queryset.count() + for start in range(0, total_records, batch_size): + with transaction.atomic(): + batch = queryset[start:start + batch_size] + for obj in batch: + stream_hash = Stream.generate_hash_key(obj.name, obj.url, obj.tvg_id, keys) + if stream_hash in hash_keys: + # Handle duplicate keys and remove any without channels + stream_channels = ChannelStream.objects.filter(stream_id=obj.id).count() + if stream_channels == 0: + obj.delete() + continue + + + existing_stream_channels = ChannelStream.objects.filter(stream_id=hash_keys[stream_hash]).count() + if existing_stream_channels == 0: + Stream.objects.filter(id=hash_keys[stream_hash]).delete() + + obj.stream_hash = stream_hash + obj.save(update_fields=['stream_hash']) + hash_keys[stream_hash] = obj.id + + logger.debug(f"Re-hashed {batch_size} streams") + + logger.debug(f"Re-hashing complete") diff --git a/core/xtream_codes.py b/core/xtream_codes.py new file mode 100644 index 00000000..17f3eaad --- /dev/null +++ b/core/xtream_codes.py @@ -0,0 +1,164 @@ +import requests +import logging +import traceback +import json + +logger = logging.getLogger(__name__) + +class Client: + """Xtream Codes API Client with robust error handling""" + + def __init__(self, server_url, username, password, user_agent=None): + self.server_url = self._normalize_url(server_url) + self.username = username + self.password = password + self.user_agent = user_agent + + # Fix: Properly handle all possible user_agent input types + if user_agent: + if isinstance(user_agent, str): + # Direct string user agent + user_agent_string = user_agent + elif hasattr(user_agent, 'user_agent'): + # UserAgent model object + user_agent_string = user_agent.user_agent + else: + # Fallback for any other type + logger.warning(f"Unexpected user_agent type: {type(user_agent)}, using default") + user_agent_string = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)' + else: + # No user agent provided + user_agent_string = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)' + + self.headers = {'User-Agent': user_agent_string} + self.server_info = None + + def _normalize_url(self, url): + """Normalize server URL by removing trailing slashes and paths""" + if not url: + raise ValueError("Server URL cannot be empty") + + url = url.rstrip('/') + # Remove any path after domain - we'll construct proper API URLs + # Split by protocol first to preserve it + if '://' in url: + protocol, rest = url.split('://', 1) + domain = rest.split('/', 1)[0] + return f"{protocol}://{domain}" + return url + + def _make_request(self, endpoint, params=None): + """Make request with detailed error handling""" + try: + url = f"{self.server_url}/{endpoint}" + logger.debug(f"XC API Request: {url} with params: {params}") + + response = requests.get(url, params=params, headers=self.headers, timeout=30) + response.raise_for_status() + + data = response.json() + logger.debug(f"XC API Response: {url} status code: {response.status_code}") + + # Check for XC-specific error responses + if isinstance(data, dict) and data.get('user_info') is None and 'error' in data: + error_msg = f"XC API Error: {data.get('error', 'Unknown error')}" + logger.error(error_msg) + raise ValueError(error_msg) + + return data + except requests.RequestException as e: + error_msg = f"XC API Request failed: {str(e)}" + logger.error(error_msg) + logger.error(f"Request details: URL={url}, Params={params}") + raise + except ValueError as e: + # This could be from JSON parsing or our explicit raises + logger.error(f"XC API Invalid response: {str(e)}") + raise + except Exception as e: + logger.error(f"XC API Unexpected error: {str(e)}") + logger.error(traceback.format_exc()) + raise + + def authenticate(self): + """Authenticate and validate server response""" + try: + endpoint = "player_api.php" + params = { + 'username': self.username, + 'password': self.password + } + + self.server_info = self._make_request(endpoint, params) + + if not self.server_info or not self.server_info.get('user_info'): + error_msg = "Authentication failed: Invalid response from server" + logger.error(f"{error_msg}. Response: {self.server_info}") + raise ValueError(error_msg) + + logger.info(f"XC Authentication successful for user {self.username}") + return self.server_info + except Exception as e: + logger.error(f"XC Authentication failed: {str(e)}") + logger.error(traceback.format_exc()) + raise + + def get_live_categories(self): + """Get live TV categories""" + try: + if not self.server_info: + self.authenticate() + + endpoint = "player_api.php" + params = { + 'username': self.username, + 'password': self.password, + 'action': 'get_live_categories' + } + + categories = self._make_request(endpoint, params) + + if not isinstance(categories, list): + error_msg = f"Invalid categories response: {categories}" + logger.error(error_msg) + raise ValueError(error_msg) + + logger.info(f"Successfully retrieved {len(categories)} live categories") + logger.debug(f"Categories: {json.dumps(categories[:5])}...") + return categories + except Exception as e: + logger.error(f"Failed to get live categories: {str(e)}") + logger.error(traceback.format_exc()) + raise + + def get_live_category_streams(self, category_id): + """Get streams for a specific category""" + try: + if not self.server_info: + self.authenticate() + + endpoint = "player_api.php" + params = { + 'username': self.username, + 'password': self.password, + 'action': 'get_live_streams', + 'category_id': category_id + } + + streams = self._make_request(endpoint, params) + + if not isinstance(streams, list): + error_msg = f"Invalid streams response for category {category_id}: {streams}" + logger.error(error_msg) + raise ValueError(error_msg) + + logger.info(f"Successfully retrieved {len(streams)} streams for category {category_id}") + return streams + except Exception as e: + logger.error(f"Failed to get streams for category {category_id}: {str(e)}") + logger.error(traceback.format_exc()) + raise + + def get_stream_url(self, stream_id): + """Get the playback URL for a stream""" + return f"{self.server_url}/live/{self.username}/{self.password}/{stream_id}.ts" diff --git a/dispatcharr/consumers.py b/dispatcharr/consumers.py index 8d92c4fa..f7d4a47c 100644 --- a/dispatcharr/consumers.py +++ b/dispatcharr/consumers.py @@ -6,12 +6,35 @@ logger = logging.getLogger(__name__) class MyWebSocketConsumer(AsyncWebsocketConsumer): async def connect(self): - await self.accept() - self.room_name = "updates" - await self.channel_layer.group_add(self.room_name, self.channel_name) + try: + await self.accept() + self.room_name = "updates" + await self.channel_layer.group_add(self.room_name, self.channel_name) + # Send a connection confirmation to the client with consistent format + await self.send(text_data=json.dumps({ + 'type': 'connection_established', + 'data': { + 'success': True, + 'message': 'WebSocket connection established successfully' + } + })) + except Exception as e: + import logging + logger = logging.getLogger(__name__) + logger.error(f"Error in WebSocket connect: {str(e)}") + # If an error occurs during connection, attempt to close + try: + await self.close(code=1011) # Internal server error + except: + pass async def disconnect(self, close_code): - await self.channel_layer.group_discard(self.room_name, self.channel_name) + try: + await self.channel_layer.group_discard(self.room_name, self.channel_name) + except Exception as e: + import logging + logger = logging.getLogger(__name__) + logger.error(f"Error in WebSocket disconnect: {str(e)}") async def receive(self, text_data): data = json.loads(text_data) diff --git a/docker/Dockerfile b/docker/Dockerfile index 26b54975..92705403 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -76,7 +76,15 @@ RUN apt-get update && \ streamlink \ wget \ gnupg2 \ - lsb-release && \ + lsb-release \ + libva-drm2 \ + libva-x11-2 \ + libva-dev \ + libva-wayland2 \ + vainfo \ + i965-va-driver \ + intel-media-va-driver \ + mesa-va-drivers && \ cp /app/docker/nginx.conf /etc/nginx/sites-enabled/default && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* diff --git a/docker/nginx.conf b/docker/nginx.conf index b440f773..65d382c5 100644 --- a/docker/nginx.conf +++ b/docker/nginx.conf @@ -7,7 +7,7 @@ server { proxy_connect_timeout 75; proxy_send_timeout 300; proxy_read_timeout 300; - client_max_body_size 128M; # Allow file uploads up to 128MB + client_max_body_size 0; # Allow file uploads up to 128MB # Serve Django via uWSGI location / { diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index e13c5af8..1c032ab3 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -65,14 +65,20 @@ const App = () => { // Authentication check useEffect(() => { const checkAuth = async () => { - const loggedIn = await initializeAuth(); - if (loggedIn) { - await initData(); - setIsAuthenticated(true); - } else { + try { + const loggedIn = await initializeAuth(); + if (loggedIn) { + await initData(); + setIsAuthenticated(true); + } else { + await logout(); + } + } catch (error) { + console.error("Auth check failed:", error); await logout(); } }; + checkAuth(); }, [initializeAuth, initData, setIsAuthenticated, logout]); diff --git a/frontend/src/WebSocket.jsx b/frontend/src/WebSocket.jsx index f538ee29..d12665fd 100644 --- a/frontend/src/WebSocket.jsx +++ b/frontend/src/WebSocket.jsx @@ -5,20 +5,363 @@ import React, { createContext, useContext, useMemo, + useCallback, } from 'react'; import useStreamsStore from './store/streams'; import { notifications } from '@mantine/notifications'; import useChannelsStore from './store/channels'; import usePlaylistsStore from './store/playlists'; import useEPGsStore from './store/epgs'; -import { Box, Button, Stack } from '@mantine/core'; +import { Box, Button, Stack, Alert, Group } from '@mantine/core'; import API from './api'; +import useSettingsStore from './store/settings'; export const WebsocketContext = createContext([false, () => { }, null]); export const WebsocketProvider = ({ children }) => { const [isReady, setIsReady] = useState(false); const [val, setVal] = useState(null); + const ws = useRef(null); + const reconnectTimerRef = useRef(null); + const [reconnectAttempts, setReconnectAttempts] = useState(0); + const [connectionError, setConnectionError] = useState(null); + const maxReconnectAttempts = 5; + const initialBackoffDelay = 1000; // 1 second initial delay + const env_mode = useSettingsStore((s) => s.environment.env_mode); + + // Calculate reconnection delay with exponential backoff + const getReconnectDelay = useCallback(() => { + return Math.min(initialBackoffDelay * Math.pow(1.5, reconnectAttempts), 30000); // max 30 seconds + }, [reconnectAttempts]); + + // Clear any existing reconnect timers + const clearReconnectTimer = useCallback(() => { + if (reconnectTimerRef.current) { + clearTimeout(reconnectTimerRef.current); + reconnectTimerRef.current = null; + } + }, []); + + // Function to get WebSocket URL that works with both HTTP and HTTPS + const getWebSocketUrl = useCallback(() => { + const protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const host = window.location.hostname; + const appPort = window.location.port; + + // In development mode, connect directly to the WebSocket server on port 8001 + if (env_mode === 'dev') { + return `${protocol}//${host}:8001/ws/`; + } else { + // In production mode, use the same port as the main application + // This allows nginx to handle the WebSocket forwarding + return appPort + ? `${protocol}//${host}:${appPort}/ws/` + : `${protocol}//${host}/ws/`; + } + }, [env_mode]); + + // Function to handle websocket connection + const connectWebSocket = useCallback(() => { + // Clear any existing timers to avoid multiple reconnection attempts + clearReconnectTimer(); + + // Clear old websocket if exists + if (ws.current) { + // Remove event handlers to prevent duplicate events + ws.current.onclose = null; + ws.current.onerror = null; + ws.current.onopen = null; + ws.current.onmessage = null; + + try { + ws.current.close(); + } catch (e) { + console.warn("Error closing existing WebSocket:", e); + } + } + + try { + console.log(`Attempting WebSocket connection (attempt ${reconnectAttempts + 1}/${maxReconnectAttempts})...`); + + // Use the function to get the correct WebSocket URL + const wsUrl = getWebSocketUrl(); + console.log(`Connecting to WebSocket at: ${wsUrl}`); + + // Create new WebSocket connection + const socket = new WebSocket(wsUrl); + + socket.onopen = () => { + console.log("WebSocket connected successfully"); + setIsReady(true); + setConnectionError(null); + setReconnectAttempts(0); + }; + + socket.onerror = (error) => { + console.error("WebSocket connection error:", error); + + // Don't show error notification on initial page load, + // only show it after a connection was established then lost + if (reconnectAttempts > 0 || isReady) { + setConnectionError("Failed to connect to WebSocket server."); + } else { + console.log("Initial connection attempt failed, will retry..."); + } + }; + + socket.onclose = (event) => { + console.warn("WebSocket connection closed", event); + setIsReady(false); + + // Only attempt reconnect if we haven't reached max attempts + if (reconnectAttempts < maxReconnectAttempts) { + const delay = getReconnectDelay(); + setConnectionError(`Connection lost. Reconnecting in ${Math.ceil(delay / 1000)} seconds...`); + console.log(`Scheduling reconnect in ${delay}ms (attempt ${reconnectAttempts + 1}/${maxReconnectAttempts})...`); + + // Store timer reference so we can cancel it if needed + reconnectTimerRef.current = setTimeout(() => { + setReconnectAttempts(prev => prev + 1); + connectWebSocket(); + }, delay); + } else { + setConnectionError("Maximum reconnection attempts reached. Please reload the page."); + console.error("Maximum reconnection attempts reached. WebSocket connection failed."); + } + }; + + // Message handler + socket.onmessage = async (event) => { + try { + const parsedEvent = JSON.parse(event.data); + + // Handle connection_established event + if (parsedEvent.type === 'connection_established') { + console.log('WebSocket connection established:', parsedEvent.data?.message); + // Don't need to do anything else for this event type + return; + } + + // Handle standard message format for other event types + switch (parsedEvent.data?.type) { + case 'epg_file': + fetchEPGs(); + notifications.show({ + title: 'EPG File Detected', + message: `Processing ${parsedEvent.data.filename}`, + }); + break; + + case 'm3u_file': + fetchPlaylists(); + notifications.show({ + title: 'M3U File Detected', + message: `Processing ${parsedEvent.data.filename}`, + }); + break; + + case 'm3u_refresh': + // Update the store with progress information + setRefreshProgress(parsedEvent.data); + + // Update the playlist status whenever we receive a status update + // Not just when progress is 100% or status is pending_setup + if (parsedEvent.data.status && parsedEvent.data.account) { + const playlistsState = usePlaylistsStore.getState(); + const playlist = playlistsState.playlists.find(p => p.id === parsedEvent.data.account); + + if (playlist) { + // When we receive a "success" status with 100% progress, this is a completed refresh + // So we should also update the updated_at timestamp + const updateData = { + ...playlist, + status: parsedEvent.data.status, + last_message: parsedEvent.data.message || playlist.last_message + }; + + // Update the timestamp when we complete a successful refresh + if (parsedEvent.data.status === 'success' && parsedEvent.data.progress === 100) { + updateData.updated_at = new Date().toISOString(); + } + + playlistsState.updatePlaylist(updateData); + } + } + break; + + case 'channel_stats': + setChannelStats(JSON.parse(parsedEvent.data.stats)); + break; + + case 'epg_channels': + notifications.show({ + message: 'EPG channels updated!', + color: 'green.5', + }); + + // If source_id is provided, update that specific EPG's status + if (parsedEvent.data.source_id) { + const epgsState = useEPGsStore.getState(); + const epg = epgsState.epgs[parsedEvent.data.source_id]; + if (epg) { + epgsState.updateEPG({ + ...epg, + status: 'success' + }); + } + } + + fetchEPGData(); + break; + + case 'epg_match': + notifications.show({ + message: parsedEvent.data.message || 'EPG match is complete!', + color: 'green.5', + }); + + // Check if we have associations data and use the more efficient batch API + if (parsedEvent.data.associations && parsedEvent.data.associations.length > 0) { + API.batchSetEPG(parsedEvent.data.associations); + } + break; + + case 'm3u_profile_test': + setProfilePreview(parsedEvent.data.search_preview, parsedEvent.data.result); + break; + + case 'recording_started': + notifications.show({ + title: 'Recording started!', + message: `Started recording channel ${parsedEvent.data.channel}`, + }); + break; + + case 'recording_ended': + notifications.show({ + title: 'Recording finished!', + message: `Stopped recording channel ${parsedEvent.data.channel}`, + }); + break; + + case 'epg_fetch_error': + notifications.show({ + title: 'EPG Source Error', + message: parsedEvent.data.message, + color: 'orange.5', + autoClose: 8000, + }); + + // Update EPG status in store + if (parsedEvent.data.source_id) { + const epgsState = useEPGsStore.getState(); + const epg = epgsState.epgs[parsedEvent.data.source_id]; + if (epg) { + epgsState.updateEPG({ + ...epg, + status: 'error', + last_message: parsedEvent.data.message + }); + } + } + break; + + case 'epg_refresh': + // Update the store with progress information + const epgsState = useEPGsStore.getState(); + epgsState.updateEPGProgress(parsedEvent.data); + + // If we have source_id/account info, update the EPG source status + if (parsedEvent.data.source_id || parsedEvent.data.account) { + const sourceId = parsedEvent.data.source_id || parsedEvent.data.account; + const epg = epgsState.epgs[sourceId]; + + if (epg) { + // Check for any indication of an error (either via status or error field) + const hasError = parsedEvent.data.status === "error" || + !!parsedEvent.data.error || + (parsedEvent.data.message && parsedEvent.data.message.toLowerCase().includes("error")); + + if (hasError) { + // Handle error state + const errorMessage = parsedEvent.data.error || parsedEvent.data.message || "Unknown error occurred"; + + epgsState.updateEPG({ + ...epg, + status: 'error', + last_message: errorMessage + }); + + // Show notification for the error + notifications.show({ + title: 'EPG Refresh Error', + message: errorMessage, + color: 'red.5', + }); + } + // Update status on completion only if no errors + else if (parsedEvent.data.progress === 100) { + epgsState.updateEPG({ + ...epg, + status: parsedEvent.data.status || 'success', + last_message: parsedEvent.data.message || epg.last_message + }); + + // Only show success notification if we've finished parsing programs and had no errors + if (parsedEvent.data.action === "parsing_programs") { + notifications.show({ + title: 'EPG Processing Complete', + message: 'EPG data has been updated successfully', + color: 'green.5', + }); + + fetchEPGData(); + } + } + } + } + break; + + default: + console.error(`Unknown websocket event type: ${parsedEvent.data?.type}`); + break; + } + } catch (error) { + console.error('Error processing WebSocket message:', error, event.data); + } + }; + + ws.current = socket; + } catch (error) { + console.error("Error creating WebSocket connection:", error); + setConnectionError(`WebSocket error: ${error.message}`); + + // Schedule a reconnect if we haven't reached max attempts + if (reconnectAttempts < maxReconnectAttempts) { + const delay = getReconnectDelay(); + reconnectTimerRef.current = setTimeout(() => { + setReconnectAttempts(prev => prev + 1); + connectWebSocket(); + }, delay); + } + } + }, [reconnectAttempts, clearReconnectTimer, getReconnectDelay, getWebSocketUrl, isReady]); + + // Initial connection and cleanup + useEffect(() => { + connectWebSocket(); + + return () => { + clearReconnectTimer(); // Clear any pending reconnect timers + + if (ws.current) { + console.log("Closing WebSocket connection due to component unmount"); + ws.current.onclose = null; // Remove handlers to avoid reconnection + ws.current.close(); + ws.current = null; + } + }; + }, [connectWebSocket, clearReconnectTimer]); const setChannelStats = useChannelsStore((s) => s.setChannelStats); const fetchChannelGroups = useChannelsStore((s) => s.fetchChannelGroups); @@ -28,146 +371,28 @@ export const WebsocketProvider = ({ children }) => { const fetchEPGData = useEPGsStore((s) => s.fetchEPGData); const fetchEPGs = useEPGsStore((s) => s.fetchEPGs); - const ws = useRef(null); - - useEffect(() => { - let wsUrl = `${window.location.host}/ws/`; - if (import.meta.env.DEV) { - wsUrl = `${window.location.hostname}:8001/ws/`; - } - - if (window.location.protocol.match(/https/)) { - wsUrl = `wss://${wsUrl}`; - } else { - wsUrl = `ws://${wsUrl}`; - } - - const socket = new WebSocket(wsUrl); - - socket.onopen = () => { - console.log('websocket connected'); - setIsReady(true); - }; - - // Reconnection logic - socket.onclose = () => { - setIsReady(false); - setTimeout(() => { - const reconnectWs = new WebSocket(wsUrl); - reconnectWs.onopen = () => setIsReady(true); - }, 3000); // Attempt to reconnect every 3 seconds - }; - - socket.onmessage = async (event) => { - event = JSON.parse(event.data); - switch (event.data.type) { - case 'epg_file': - fetchEPGs(); - notifications.show({ - title: 'EPG File Detected', - message: `Processing ${event.data.filename}`, - }); - break; - - case 'm3u_file': - fetchPlaylists(); - notifications.show({ - title: 'M3U File Detected', - message: `Processing ${event.data.filename}`, - }); - break; - - case 'm3u_group_refresh': - fetchChannelGroups(); - fetchPlaylists(); - - notifications.show({ - title: 'Group processing finished!', - autoClose: 5000, - message: ( - - Refresh M3U or filter out groups to pull in streams. - - - ), - color: 'green.5', - }); - break; - - case 'm3u_refresh': - setRefreshProgress(event.data); - break; - - case 'channel_stats': - setChannelStats(JSON.parse(event.data.stats)); - break; - - case 'epg_channels': - notifications.show({ - message: 'EPG channels updated!', - color: 'green.5', - }); - fetchEPGData(); - break; - - case 'epg_match': - notifications.show({ - message: event.data.message || 'EPG match is complete!', - color: 'green.5', - }); - - // Check if we have associations data and use the more efficient batch API - if (event.data.associations && event.data.associations.length > 0) { - API.batchSetEPG(event.data.associations); - } - break; - - case 'm3u_profile_test': - setProfilePreview(event.data.search_preview, event.data.result); - break; - - case 'recording_started': - notifications.show({ - title: 'Recording started!', - message: `Started recording channel ${event.data.channel}`, - }); - break; - - case 'recording_ended': - notifications.show({ - title: 'Recording finished!', - message: `Stopped recording channel ${event.data.channel}`, - }); - break; - - default: - console.error(`Unknown websocket event type: ${event.type}`); - break; - } - }; - - ws.current = socket; - - return () => { - socket.close(); - }; - }, []); - const ret = useMemo(() => { return [isReady, ws.current?.send.bind(ws.current), val]; }, [isReady, val]); return ( + {connectionError && !isReady && reconnectAttempts >= maxReconnectAttempts && ( + + {connectionError} + + + )} + {connectionError && !isReady && reconnectAttempts < maxReconnectAttempts && reconnectAttempts > 0 && ( + + {connectionError} + + )} {children} ); diff --git a/frontend/src/api.js b/frontend/src/api.js index 3cec6e38..60b22634 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -347,6 +347,11 @@ export default class API { payload.tvg_id = null; } + // Ensure tvc_guide_stationid is included properly (not as empty string) + if (payload.tvc_guide_stationid === '') { + payload.tvc_guide_stationid = null; + } + // Handle channel_number properly if (payload.channel_number === '') { payload.channel_number = null; @@ -373,6 +378,31 @@ export default class API { } } + static async updateChannels(ids, values) { + const body = []; + for (const id in ids) { + body.push({ + id, + ...values, + }); + } + + try { + const response = await request( + `${host}/api/channels/channels/edit/bulk/`, + { + method: 'PATCH', + body, + } + ); + + useChannelsStore.getState().updateChannels(response); + return response; + } catch (e) { + errorNotification('Failed to update channels', e); + } + } + static async setChannelEPG(channelId, epgDataId) { try { const response = await request( @@ -403,16 +433,13 @@ export default class API { } } - static async assignChannelNumbers(channelIds) { + static async assignChannelNumbers(channelIds, startingNum = 1) { try { const response = await request(`${host}/api/channels/channels/assign/`, { method: 'POST', - body: { channel_order: channelIds }, + body: { channel_ids: channelIds, starting_number: startingNum }, }); - // Optionally refesh the channel list in Zustand - // await useChannelsStore.getState().fetchChannels(); - return response; } catch (e) { errorNotification('Failed to assign channel #s', e); @@ -655,6 +682,10 @@ export default class API { } static async addPlaylist(values) { + if (values.custom_properties) { + values.custom_properties = JSON.stringify(values.custom_properties); + } + try { let body = null; if (values.file) { @@ -719,10 +750,26 @@ export default class API { } } - static async updatePlaylist(values) { + static async updatePlaylist(values, isToggle = false) { const { id, ...payload } = values; try { + // If this is just toggling the active state, make a simpler request + if ( + isToggle && + 'is_active' in payload && + Object.keys(payload).length === 1 + ) { + const response = await request(`${host}/api/m3u/accounts/${id}/`, { + method: 'PATCH', + body: { is_active: payload.is_active }, + }); + + usePlaylistsStore.getState().updatePlaylist(response); + return response; + } + + // Original implementation for full updates let body = null; if (payload.file) { delete payload.server_url; @@ -740,6 +787,7 @@ export default class API { body = { ...payload }; delete body.file; } + console.log(body); const response = await request(`${host}/api/m3u/accounts/${id}/`, { method: 'PATCH', @@ -803,10 +851,26 @@ export default class API { } } - static async updateEPG(values) { + static async updateEPG(values, isToggle = false) { const { id, ...payload } = values; try { + // If this is just toggling the active state, make a simpler request + if ( + isToggle && + 'is_active' in payload && + Object.keys(payload).length === 1 + ) { + const response = await request(`${host}/api/epg/sources/${id}/`, { + method: 'PATCH', + body: { is_active: payload.is_active }, + }); + + useEPGsStore.getState().updateEPG(response); + return response; + } + + // Original implementation for full updates let body = null; if (payload.files) { body = new FormData(); @@ -1246,10 +1310,13 @@ export default class API { static async switchStream(channelId, streamId) { try { - const response = await request(`${host}/proxy/ts/change_stream/${channelId}`, { - method: 'POST', - body: { stream_id: streamId }, - }); + const response = await request( + `${host}/proxy/ts/change_stream/${channelId}`, + { + method: 'POST', + body: { stream_id: streamId }, + } + ); return response; } catch (e) { @@ -1260,10 +1327,13 @@ export default class API { static async nextStream(channelId, streamId) { try { - const response = await request(`${host}/proxy/ts/next_stream/${channelId}`, { - method: 'POST', - body: { stream_id: streamId }, - }); + const response = await request( + `${host}/proxy/ts/next_stream/${channelId}`, + { + method: 'POST', + body: { stream_id: streamId }, + } + ); return response; } catch (e) { @@ -1301,4 +1371,16 @@ export default class API { errorNotification('Failed to update channel EPGs', e); } } + + static async getChannel(id) { + try { + const response = await request( + `${host}/api/channels/channels/${id}/?include_streams=true` + ); + return response; + } catch (e) { + errorNotification('Failed to fetch channel details', e); + return null; + } + } } diff --git a/frontend/src/components/ConfirmationDialog.jsx b/frontend/src/components/ConfirmationDialog.jsx new file mode 100644 index 00000000..3c0f15e7 --- /dev/null +++ b/frontend/src/components/ConfirmationDialog.jsx @@ -0,0 +1,77 @@ +import { Modal, Group, Text, Button, Checkbox, Box } from '@mantine/core'; +import React, { useState } from 'react'; +import useWarningsStore from '../store/warnings'; + +/** + * A reusable confirmation dialog with option to suppress future warnings + * + * @param {Object} props - Component props + * @param {boolean} props.opened - Whether the dialog is visible + * @param {Function} props.onClose - Function to call when closing without confirming + * @param {Function} props.onConfirm - Function to call when confirming the action + * @param {string} props.title - Dialog title + * @param {string} props.message - Dialog message + * @param {string} props.confirmLabel - Text for the confirm button + * @param {string} props.cancelLabel - Text for the cancel button + * @param {string} props.actionKey - Unique key for this type of action (used for suppression) + * @param {Function} props.onSuppressChange - Called when "don't show again" option changes + * @param {string} [props.size='md'] - Size of the modal + */ +const ConfirmationDialog = ({ + opened, + onClose, + onConfirm, + title = 'Confirm Action', + message = 'Are you sure you want to proceed?', + confirmLabel = 'Confirm', + cancelLabel = 'Cancel', + actionKey, + onSuppressChange, + size = 'md', // Add default size parameter - md is a medium width +}) => { + const suppressWarning = useWarningsStore((s) => s.suppressWarning); + const isWarningSuppressed = useWarningsStore((s) => s.isWarningSuppressed); + const [suppressChecked, setSuppressChecked] = useState( + isWarningSuppressed(actionKey) + ); + + const handleToggleSuppress = (e) => { + setSuppressChecked(e.currentTarget.checked); + if (onSuppressChange) { + onSuppressChange(e.currentTarget.checked); + } + }; + + const handleConfirm = () => { + if (suppressChecked) { + suppressWarning(actionKey); + } + onConfirm(); + }; + + return ( + + {message} + + {actionKey && ( + + )} + + + + + + + ); +}; + +export default ConfirmationDialog; diff --git a/frontend/src/components/FloatingVideo.jsx b/frontend/src/components/FloatingVideo.jsx index 0be42029..46c191eb 100644 --- a/frontend/src/components/FloatingVideo.jsx +++ b/frontend/src/components/FloatingVideo.jsx @@ -1,9 +1,9 @@ // frontend/src/components/FloatingVideo.js -import React, { useEffect, useRef } from 'react'; +import React, { useEffect, useRef, useState } from 'react'; import Draggable from 'react-draggable'; import useVideoStore from '../store/useVideoStore'; import mpegts from 'mpegts.js'; -import { CloseButton, Flex } from '@mantine/core'; +import { CloseButton, Flex, Loader, Text, Box } from '@mantine/core'; export default function FloatingVideo() { const isVisible = useVideoStore((s) => s.isVisible); @@ -12,40 +12,150 @@ export default function FloatingVideo() { const videoRef = useRef(null); const playerRef = useRef(null); const videoContainerRef = useRef(null); + // Convert ref to state so we can use it for rendering + const [isLoading, setIsLoading] = useState(false); + const [loadError, setLoadError] = useState(null); + + // Safely destroy the player to prevent errors + const safeDestroyPlayer = () => { + try { + if (playerRef.current) { + // Set loading to false when destroying player + setIsLoading(false); + setLoadError(null); + + // First unload the source to stop any in-progress fetches + if (videoRef.current) { + // Remove src attribute and force a load to clear any pending requests + videoRef.current.removeAttribute('src'); + videoRef.current.load(); + } + + // Pause the player first + try { + playerRef.current.pause(); + } catch (e) { + // Ignore pause errors + } + + // Use a try-catch block specifically for the destroy call + try { + playerRef.current.destroy(); + } catch (error) { + // Ignore expected abort errors + if (error.name !== 'AbortError' && !error.message?.includes('aborted')) { + console.log("Error during player destruction:", error.message); + } + } finally { + playerRef.current = null; + } + } + } catch (error) { + console.log("Error during player cleanup:", error); + playerRef.current = null; + } + }; useEffect(() => { if (!isVisible || !streamUrl) { + safeDestroyPlayer(); return; } - // If the browser supports MSE for live playback, initialize mpegts.js - if (mpegts.getFeatureList().mseLivePlayback) { - const player = mpegts.createPlayer({ - type: 'mpegts', - url: streamUrl, - isLive: true, - // You can include other custom MPEGTS.js config fields here, e.g.: - // cors: true, - // withCredentials: false, - }); + // Check if we have an existing player and clean it up + safeDestroyPlayer(); - player.attachMediaElement(videoRef.current); - player.load(); - player.play(); + // Set loading state to true when starting a new stream + setIsLoading(true); + setLoadError(null); - // Store player instance so we can clean up later - playerRef.current = player; + // Debug log to help diagnose stream issues + console.log("Attempting to play stream:", streamUrl); + + try { + // If the browser supports MSE for live playback, initialize mpegts.js + if (mpegts.getFeatureList().mseLivePlayback) { + // Set loading flag + setIsLoading(true); + + const player = mpegts.createPlayer({ + type: 'mpegts', // MPEG-TS format + url: streamUrl, + isLive: true, + enableWorker: true, + enableStashBuffer: false, // Try disabling stash buffer for live streams + liveBufferLatencyChasing: true, + liveSync: true, + cors: true, // Enable CORS for cross-domain requests + // Add error recovery options + autoCleanupSourceBuffer: true, + autoCleanupMaxBackwardDuration: 10, + autoCleanupMinBackwardDuration: 5, + reuseRedirectedURL: true, + }); + + player.attachMediaElement(videoRef.current); + + // Add events to track loading state + player.on(mpegts.Events.LOADING_COMPLETE, () => { + setIsLoading(false); + }); + + player.on(mpegts.Events.METADATA_ARRIVED, () => { + setIsLoading(false); + }); + + // Add error event handler + player.on(mpegts.Events.ERROR, (errorType, errorDetail) => { + setIsLoading(false); + + // Filter out aborted errors + if (errorType !== 'NetworkError' || !errorDetail?.includes('aborted')) { + console.error('Player error:', errorType, errorDetail); + setLoadError(`Error: ${errorType}${errorDetail ? ` - ${errorDetail}` : ''}`); + } + }); + + player.load(); + + // Don't auto-play until we've loaded properly + player.on(mpegts.Events.MEDIA_INFO, () => { + setIsLoading(false); + try { + player.play().catch(e => { + console.log("Auto-play prevented:", e); + setLoadError("Auto-play was prevented. Click play to start."); + }); + } catch (e) { + console.log("Error during play:", e); + setLoadError(`Playback error: ${e.message}`); + } + }); + + // Store player instance so we can clean up later + playerRef.current = player; + } + } catch (error) { + setIsLoading(false); + setLoadError(`Initialization error: ${error.message}`); + console.error("Error initializing player:", error); } // Cleanup when component unmounts or streamUrl changes return () => { - if (playerRef.current) { - playerRef.current.destroy(); - playerRef.current = null; - } + safeDestroyPlayer(); }; }, [isVisible, streamUrl]); + // Modified hideVideo handler to clean up player first + const handleClose = () => { + safeDestroyPlayer(); + // Small delay before hiding the video component to ensure cleanup is complete + setTimeout(() => { + hideVideo(); + }, 50); + }; + // If the floating video is hidden or no URL is selected, do not render if (!isVisible || !streamUrl) { return null; @@ -69,15 +179,66 @@ export default function FloatingVideo() { > {/* Simple header row with a close button */} - + - {/* The