Merge branch 'dev' into Feature/954-1004-user-account-management

.github/workflows/ci.yml

@@ -201,13 +201,15 @@ jobs:
 
           echo "Creating multi-arch manifest for ${OWNER}/${REPO}"
 
-          # branch tag (e.g. latest or dev)
+          # ghcr.io: both the branch tag (e.g. dev) and the version+timestamp tag
+          # point to the same manifest by using multiple --tag flags in one call,
+          # which prevents orphaned untagged images on each new build
           docker buildx imagetools create \
             --annotation "index:org.opencontainers.image.title=${{ needs.prepare.outputs.repo_name }}" \
             --annotation "index:org.opencontainers.image.description=Your ultimate IPTV & stream Management companion." \
             --annotation "index:org.opencontainers.image.url=https://github.com/${{ github.repository }}" \
             --annotation "index:org.opencontainers.image.source=https://github.com/${{ github.repository }}" \
-            --annotation "index:org.opencontainers.image.version=${BRANCH_TAG}" \
+            --annotation "index:org.opencontainers.image.version=${VERSION}-${TIMESTAMP}" \
             --annotation "index:org.opencontainers.image.created=${TIMESTAMP}" \
             --annotation "index:org.opencontainers.image.revision=${{ github.sha }}" \
             --annotation "index:org.opencontainers.image.licenses=See repository" \
@@ -217,9 +219,10 @@ jobs:
             --annotation "index:maintainer=${{ github.actor }}" \
             --annotation "index:build_version=Dispatcharr version: ${VERSION}-${TIMESTAMP}" \
             --tag ghcr.io/${OWNER}/${REPO}:${BRANCH_TAG} \
+            --tag ghcr.io/${OWNER}/${REPO}:${VERSION}-${TIMESTAMP} \
             ghcr.io/${OWNER}/${REPO}:${BRANCH_TAG}-amd64 ghcr.io/${OWNER}/${REPO}:${BRANCH_TAG}-arm64
 
-          # version + timestamp tag
+          # Docker Hub: same single call with both tags
           docker buildx imagetools create \
             --annotation "index:org.opencontainers.image.title=${{ needs.prepare.outputs.repo_name }}" \
             --annotation "index:org.opencontainers.image.description=Your ultimate IPTV & stream Management companion." \
@@ -234,40 +237,6 @@ jobs:
             --annotation "index:org.opencontainers.image.authors=${{ github.actor }}" \
             --annotation "index:maintainer=${{ github.actor }}" \
             --annotation "index:build_version=Dispatcharr version: ${VERSION}-${TIMESTAMP}" \
-            --tag ghcr.io/${OWNER}/${REPO}:${VERSION}-${TIMESTAMP} \
-            ghcr.io/${OWNER}/${REPO}:${VERSION}-${TIMESTAMP}-amd64 ghcr.io/${OWNER}/${REPO}:${VERSION}-${TIMESTAMP}-arm64
-
-          # also create Docker Hub manifests using the same username
-          docker buildx imagetools create \
-            --annotation "index:org.opencontainers.image.title=${{ needs.prepare.outputs.repo_name }}" \
-            --annotation "index:org.opencontainers.image.description=Your ultimate IPTV & stream Management companion." \
-            --annotation "index:org.opencontainers.image.url=https://github.com/${{ github.repository }}" \
-            --annotation "index:org.opencontainers.image.source=https://github.com/${{ github.repository }}" \
-            --annotation "index:org.opencontainers.image.version=${BRANCH_TAG}" \
-            --annotation "index:org.opencontainers.image.created=${TIMESTAMP}" \
-            --annotation "index:org.opencontainers.image.revision=${{ github.sha }}" \
-            --annotation "index:org.opencontainers.image.licenses=See repository" \
-            --annotation "index:org.opencontainers.image.documentation=https://dispatcharr.github.io/Dispatcharr-Docs/" \
-            --annotation "index:org.opencontainers.image.vendor=${OWNER}" \
-            --annotation "index:org.opencontainers.image.authors=${{ github.actor }}" \
-            --annotation "index:maintainer=${{ github.actor }}" \
-            --annotation "index:build_version=Dispatcharr version: ${VERSION}-${TIMESTAMP}" \
             --tag docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${BRANCH_TAG} \
-            docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${BRANCH_TAG}-amd64 docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${BRANCH_TAG}-arm64
-
-          docker buildx imagetools create \
-            --annotation "index:org.opencontainers.image.title=${{ needs.prepare.outputs.repo_name }}" \
-            --annotation "index:org.opencontainers.image.description=Your ultimate IPTV & stream Management companion." \
-            --annotation "index:org.opencontainers.image.url=https://github.com/${{ github.repository }}" \
-            --annotation "index:org.opencontainers.image.source=https://github.com/${{ github.repository }}" \
-            --annotation "index:org.opencontainers.image.version=${VERSION}-${TIMESTAMP}" \
-            --annotation "index:org.opencontainers.image.created=${TIMESTAMP}" \
-            --annotation "index:org.opencontainers.image.revision=${{ github.sha }}" \
-            --annotation "index:org.opencontainers.image.licenses=See repository" \
-            --annotation "index:org.opencontainers.image.documentation=https://dispatcharr.github.io/Dispatcharr-Docs/" \
-            --annotation "index:org.opencontainers.image.vendor=${OWNER}" \
-            --annotation "index:org.opencontainers.image.authors=${{ github.actor }}" \
-            --annotation "index:maintainer=${{ github.actor }}" \
-            --annotation "index:build_version=Dispatcharr version: ${VERSION}-${TIMESTAMP}" \
             --tag docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${VERSION}-${TIMESTAMP} \
-            docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${VERSION}-${TIMESTAMP}-amd64 docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${VERSION}-${TIMESTAMP}-arm64
+            docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${BRANCH_TAG}-amd64 docker.io/${{ secrets.DOCKERHUB_ORGANIZATION }}/${REPO}:${BRANCH_TAG}-arm64

CHANGELOG.md

@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- API key authentication: Added support for API key-based authentication as an alternative to JWT tokens. Users can generate and revoke their own personal API key from their profile page, enabling programmatic access for scripts, automations, and third-party integrations without exposing account credentials. Keys authenticate via the `Authorization: ApiKey <key>` header or the `X-API-Key: <key>` header. Admin users can additionally generate and revoke keys on behalf of any user.
 - Lightweight channel summary API endpoint: Added a new `/api/channels/summary/` endpoint that returns only the minimal channel data needed for TV Guide and DVR scheduling (id, name, logo), avoiding the overhead of serializing full channel objects for high-frequency UI operations.
 - Custom Dummy EPG subtitle template support: Added optional subtitle template field to custom dummy EPG configuration. Users can now define subtitle patterns using extracted regex groups and time/date placeholders (e.g., `{starttime} - {endtime}`). (Closes #942)
 - Event-driven webhooks and script execution (Connect): Added new Connect feature that enables event-driven execution of custom scripts and webhooks in response to system events. Supports multiple event types including channel lifecycle (start, stop, reconnect, error, failover), stream operations (switch), recording events (start, end), data refreshes (EPG, M3U), and client activity (connect, disconnect). Event data is available as environment variables in scripts (prefixed with `DISPATCHARR_`), POST payloads for webhooks, and plugin execution payloads. Plugins can now subscribe to events by specifying an `events` array in their action definitions. Includes connection testing endpoint with dummy payloads for validation. (Closes #203)
@@ -149,6 +150,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Plugin loader now supports `plugin.py` without `__init__.py`, including folders with non-identifier names, by loading modules directly from file paths.
 - Plugin action handling stabilized: avoids registry race conditions and only shows loading on the active action.
 - Plugin enable/disable toggles now update immediately without requiring a full page refresh.
+- M3U/EPG tasks downloading endlessly for large files: Fixed the root cause where the Redis task lock (300s TTL) expired during long downloads, allowing Celery Beat to start competing duplicate tasks that never completed. Added a `TaskLockRenewer` daemon thread that periodically extends the lock TTL while a task is actively working, applied to all long-running task paths (M3U refresh, M3U group refresh, EPG refresh, EPG program parsing). Also adds an HTTP timeout to M3U download requests, streams M3U downloads directly to a temp file on disk instead of accumulating the entire file in memory, and adds Celery task time limits as a safety net against runaway tasks. (Fixes #861) - Thanks [@CodeBormen](https://github.com/CodeBormen)
 
 ## [0.18.1] - 2026-01-27
 

apps/accounts/api_urls.py

@@ -4,6 +4,7 @@ from .api_views import (
     AuthViewSet,
     UserViewSet,
     GroupViewSet,
+    APIKeyViewSet,
     TokenObtainPairView,
     TokenRefreshView,
     list_permissions,
@@ -17,6 +18,7 @@ app_name = "accounts"
 router = DefaultRouter()
 router.register(r"users", UserViewSet, basename="user")
 router.register(r"groups", GroupViewSet, basename="group")
+router.register(r"api-keys", APIKeyViewSet, basename="api-key")
 
 # 🔹 Custom Authentication Endpoints
 auth_view = AuthViewSet.as_view({"post": "login"})

apps/accounts/api_views.py

@@ -9,6 +9,7 @@ from rest_framework import viewsets, status, serializers
 from drf_spectacular.utils import extend_schema, OpenApiParameter, inline_serializer
 from drf_spectacular.types import OpenApiTypes
 import json
+import secrets
 from .permissions import IsAdmin, Authenticated
 from dispatcharr.utils import network_access_allowed
 
@@ -318,6 +319,59 @@ class GroupViewSet(viewsets.ModelViewSet):
         return super().destroy(request, *args, **kwargs)
 
 
+# API Key management
+class APIKeyViewSet(viewsets.ViewSet):
+    permission_classes = [Authenticated]
+
+    def list(self, request):
+        user = request.user
+        return Response({"key": user.api_key})
+
+    @action(detail=False, methods=["post"], url_path="generate")
+    def generate(self, request):
+        target_user = request.user
+        user_id = request.data.get("user_id")
+
+        if user_id:
+            from .permissions import IsAdmin
+
+            if not IsAdmin().has_permission(request, self):
+                return Response({"detail": "Not allowed to create keys for other users."}, status=status.HTTP_403_FORBIDDEN)
+
+            try:
+                target_user = User.objects.get(id=int(user_id))
+            except Exception:
+                return Response({"detail": "User not found."}, status=status.HTTP_404_NOT_FOUND)
+
+        raw = secrets.token_urlsafe(40)
+        target_user.api_key = raw
+        target_user.save(update_fields=["api_key"])
+
+        user_data = UserSerializer(target_user).data
+        return Response({"key": raw, "user": user_data}, status=status.HTTP_201_CREATED)
+
+    @action(detail=False, methods=["post"], url_path="revoke")
+    def revoke(self, request):
+        target_user = request.user
+        user_id = request.data.get("user_id")
+
+        if user_id:
+            from .permissions import IsAdmin
+
+            if not IsAdmin().has_permission(request, self):
+                return Response({"detail": "Not allowed to revoke keys for other users."}, status=status.HTTP_403_FORBIDDEN)
+
+            try:
+                target_user = User.objects.get(id=int(user_id))
+            except Exception:
+                return Response({"detail": "User not found."}, status=status.HTTP_404_NOT_FOUND)
+
+        target_user.api_key = None
+        target_user.save(update_fields=["api_key"])
+
+        return Response({"success": True})
+
+
 # 🔹 4) Permissions List API
 @extend_schema(
     description="Retrieve a list of all permissions",

apps/accounts/authentication.py

@@ -0,0 +1,49 @@
+from rest_framework import authentication
+from rest_framework import exceptions
+from django.conf import settings
+from .models import User
+
+
+class ApiKeyAuthentication(authentication.BaseAuthentication):
+    """
+    Accepts header `Authorization: ApiKey <key>` or `X-API-Key: <key>`.
+    """
+
+    keyword = "ApiKey"
+
+    def authenticate(self, request):
+        # Check X-API-Key header first
+        raw_key = request.META.get("HTTP_X_API_KEY")
+
+        if not raw_key:
+            auth = authentication.get_authorization_header(request).split()
+            if not auth:
+                return None
+
+            if len(auth) != 2:
+                return None
+
+            scheme = auth[0].decode().lower()
+            if scheme != self.keyword.lower():
+                return None
+
+            raw_key = auth[1].decode()
+
+        if not raw_key:
+            return None
+
+        if not raw_key:
+            return None
+
+        try:
+            user = User.objects.get(api_key=raw_key)
+        except User.DoesNotExist:
+            raise exceptions.AuthenticationFailed("Invalid API key")
+
+        if not user.is_active:
+            raise exceptions.AuthenticationFailed("User inactive")
+
+        return (user, None)
+
+    def authenticate_header(self, request):
+        return self.keyword

apps/accounts/migrations/0004_user_api_key.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.2.11 on 2026-02-21 18:14
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('accounts', '0003_alter_user_custom_properties'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='user',
+            name='api_key',
+            field=models.CharField(blank=True, db_index=True, max_length=200, null=True),
+        ),
+    ]

apps/accounts/models.py

@@ -29,6 +29,7 @@ class User(AbstractUser):
     )
     user_level = models.IntegerField(default=UserLevel.STREAMER)
     custom_properties = models.JSONField(default=dict, blank=True, null=True)
+    api_key = models.CharField(max_length=200, blank=True, null=True, db_index=True)
 
     def __str__(self):
         return self.username

apps/accounts/serializers.py

@@ -28,12 +28,14 @@ class UserSerializer(serializers.ModelSerializer):
     channel_profiles = serializers.PrimaryKeyRelatedField(
         queryset=ChannelProfile.objects.all(), many=True, required=False
     )
+    api_key = serializers.CharField(read_only=True, allow_null=True)
 
     class Meta:
         model = User
         fields = [
             "id",
             "username",
+            "api_key",
             "email",
             "user_level",
             "password",

apps/epg/serializers.py

@@ -53,7 +53,15 @@ class EPGSourceSerializer(serializers.ModelSerializer):
         return data
 
     def update(self, instance, validated_data):
-        cron_expr = validated_data.pop('cron_expression', '')
+        # Pop cron_expression before it reaches model fields
+        # If not present (partial update), preserve the existing cron from the PeriodicTask
+        if 'cron_expression' in validated_data:
+            cron_expr = validated_data.pop('cron_expression')
+        else:
+            cron_expr = ''
+            if instance.refresh_task_id and instance.refresh_task and instance.refresh_task.crontab:
+                ct = instance.refresh_task.crontab
+                cron_expr = f'{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}'
         instance._cron_expression = cron_expr
         for attr, value in validated_data.items():
             setattr(instance, attr, value)

apps/epg/signals.py

@@ -70,7 +70,7 @@ def create_dummy_epg_data(sender, instance, created, **kwargs):
             logger.debug(f"EPGData already exists for dummy EPG source: {instance.name} (ID: {instance.id})")
 
 @receiver(post_save, sender=EPGSource)
-def create_or_update_refresh_task(sender, instance, **kwargs):
+def create_or_update_refresh_task(sender, instance, created, update_fields=None, **kwargs):
     """
     Create or update a Celery Beat periodic task when an EPGSource is created/updated.
     Skip creating tasks for dummy EPG sources as they don't need refreshing.
@@ -84,11 +84,35 @@ def create_or_update_refresh_task(sender, instance, **kwargs):
             instance.refresh_task.save(update_fields=['enabled'])
         return
 
+    # Skip rescheduling when only non-schedule fields were saved (e.g. status/last_message
+    # updates from the refresh task itself). We only need to reschedule when schedule-relevant
+    # fields change or when _cron_expression was explicitly set by the serializer.
+    SCHEDULE_FIELDS = {'refresh_interval', 'is_active', 'refresh_task'}
+    if (
+        not created
+        and update_fields is not None
+        and not (set(update_fields) & SCHEDULE_FIELDS)
+        and not hasattr(instance, '_cron_expression')
+    ):
+        return
+
     task_name = f"epg_source-refresh-{instance.id}"
     should_be_enabled = instance.is_active
 
-    # Read cron_expression from transient attribute set by the serializer
-    cron_expr = getattr(instance, "_cron_expression", "")
+    # Read cron_expression from transient attribute set by the serializer.
+    # If not set (e.g. save came from a task updating status/last_message),
+    # preserve the existing crontab so we don't accidentally revert to interval.
+    if hasattr(instance, "_cron_expression"):
+        cron_expr = instance._cron_expression
+    else:
+        cron_expr = ""
+        try:
+            existing_task = instance.refresh_task
+            if existing_task and existing_task.crontab:
+                ct = existing_task.crontab
+                cron_expr = f"{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}"
+        except Exception:
+            pass
 
     task = create_or_update_periodic_task(
         task_name=task_name,

apps/epg/tasks.py

@@ -24,7 +24,7 @@ from asgiref.sync import async_to_sync
 from channels.layers import get_channel_layer
 
 from .models import EPGSource, EPGData, ProgramData
-from core.utils import acquire_task_lock, release_task_lock, send_websocket_update, cleanup_memory, log_system_event
+from core.utils import acquire_task_lock, release_task_lock, TaskLockRenewer, send_websocket_update, cleanup_memory, log_system_event
 
 logger = logging.getLogger(__name__)
 
@@ -146,12 +146,15 @@ def refresh_all_epg_data():
     return "EPG data refreshed."
 
 
-@shared_task
+@shared_task(time_limit=1800, soft_time_limit=1700)
 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
 
+    lock_renewer = TaskLockRenewer('refresh_epg_data', source_id)
+    lock_renewer.start()
+
     source = None
     try:
         # Try to get the EPG source
@@ -168,6 +171,7 @@ def refresh_epg_data(source_id):
                 logger.info(f"No orphaned task found for EPG source {source_id}")
 
             # Release the lock and exit
+            lock_renewer.stop()
             release_task_lock('refresh_epg_data', source_id)
             # Force garbage collection before exit
             gc.collect()
@@ -176,6 +180,7 @@ def refresh_epg_data(source_id):
         # The source exists but is not active, just skip processing
         if not source.is_active:
             logger.info(f"EPG source {source_id} is not active. Skipping.")
+            lock_renewer.stop()
             release_task_lock('refresh_epg_data', source_id)
             # Force garbage collection before exit
             gc.collect()
@@ -184,6 +189,7 @@ def refresh_epg_data(source_id):
         # Skip refresh for dummy EPG sources - they don't need refreshing
         if source.source_type == 'dummy':
             logger.info(f"Skipping refresh for dummy EPG source {source.name} (ID: {source_id})")
+            lock_renewer.stop()
             release_task_lock('refresh_epg_data', source_id)
             gc.collect()
             return
@@ -194,6 +200,7 @@ def refresh_epg_data(source_id):
             fetch_success = fetch_xmltv(source)
             if not fetch_success:
                 logger.error(f"Failed to fetch XMLTV for source {source.name}")
+                lock_renewer.stop()
                 release_task_lock('refresh_epg_data', source_id)
                 # Force garbage collection before exit
                 gc.collect()
@@ -202,6 +209,7 @@ def refresh_epg_data(source_id):
             parse_channels_success = parse_channels_only(source)
             if not parse_channels_success:
                 logger.error(f"Failed to parse channels for source {source.name}")
+                lock_renewer.stop()
                 release_task_lock('refresh_epg_data', source_id)
                 # Force garbage collection before exit
                 gc.collect()
@@ -234,6 +242,7 @@ def refresh_epg_data(source_id):
         source = None
         # Force garbage collection before releasing the lock
         gc.collect()
+        lock_renewer.stop()
         release_task_lock('refresh_epg_data', source_id)
 
 
@@ -1126,12 +1135,15 @@ def parse_channels_only(source):
 
 
 
-@shared_task
+@shared_task(time_limit=3600, soft_time_limit=3500)
 def parse_programs_for_tvg_id(epg_id):
     if not acquire_task_lock('parse_epg_programs', epg_id):
         logger.info(f"Program parse for {epg_id} already in progress, skipping duplicate task")
         return "Task already running"
 
+    lock_renewer = TaskLockRenewer('parse_epg_programs', epg_id)
+    lock_renewer.start()
+
     source_file = None
     program_parser = None
     programs_to_create = []
@@ -1161,11 +1173,13 @@ def parse_programs_for_tvg_id(epg_id):
         # Skip program parsing for dummy EPG sources - they don't have program data files
         if epg_source.source_type == 'dummy':
             logger.info(f"Skipping program parsing for dummy EPG source {epg_source.name} (ID: {epg_id})")
+            lock_renewer.stop()
             release_task_lock('parse_epg_programs', epg_id)
             return
 
         if not Channel.objects.filter(epg_data=epg).exists():
             logger.info(f"No channels matched to EPG {epg.tvg_id}")
+            lock_renewer.stop()
             release_task_lock('parse_epg_programs', epg_id)
             return
 
@@ -1207,6 +1221,7 @@ def parse_programs_for_tvg_id(epg_id):
                     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")
+                    lock_renewer.stop()
                     release_task_lock('parse_epg_programs', epg_id)
                     return
 
@@ -1217,6 +1232,7 @@ def parse_programs_for_tvg_id(epg_id):
                     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")
+                    lock_renewer.stop()
                     release_task_lock('parse_epg_programs', epg_id)
                     return
 
@@ -1232,6 +1248,7 @@ def parse_programs_for_tvg_id(epg_id):
                 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")
+                lock_renewer.stop()
                 release_task_lock('parse_epg_programs', epg_id)
                 return
 
@@ -1379,7 +1396,7 @@ def parse_programs_for_tvg_id(epg_id):
         epg_source = None
         # Add comprehensive cleanup before releasing lock
         cleanup_memory(log_usage=should_log_memory, force_collection=True)
-         # Memory tracking after processing
+        # Memory tracking after processing
         if process:
             try:
                 mem_after = process.memory_info().rss / 1024 / 1024
@@ -1389,6 +1406,7 @@ def parse_programs_for_tvg_id(epg_id):
             process = None
         epg = None
         programs_processed = None
+        lock_renewer.stop()
         release_task_lock('parse_epg_programs', epg_id)
 
 

apps/m3u/serializers.py

@@ -204,7 +204,14 @@ class M3UAccountSerializer(serializers.ModelSerializer):
 
     def update(self, instance, validated_data):
         # Pop cron_expression before it reaches model fields
-        cron_expr = validated_data.pop("cron_expression", "")
+        # If not present (partial update), preserve the existing cron from the PeriodicTask
+        if "cron_expression" in validated_data:
+            cron_expr = validated_data.pop("cron_expression")
+        else:
+            cron_expr = ""
+            if instance.refresh_task_id and instance.refresh_task and instance.refresh_task.crontab:
+                ct = instance.refresh_task.crontab
+                cron_expr = f"{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}"
         instance._cron_expression = cron_expr
 
         # Handle enable_vod preference and auto_enable_new_groups settings

apps/m3u/signals.py

@@ -20,16 +20,40 @@ def refresh_account_on_save(sender, instance, created, **kwargs):
         refresh_m3u_groups.delay(instance.id)
 
 @receiver(post_save, sender=M3UAccount)
-def create_or_update_refresh_task(sender, instance, **kwargs):
+def create_or_update_refresh_task(sender, instance, created, update_fields=None, **kwargs):
     """
     Create or update a Celery Beat periodic task when an M3UAccount is created/updated.
     Supports both interval-based and cron-based scheduling via the shared utility.
     """
+    # Skip rescheduling when only non-schedule fields were saved (e.g. status/last_message
+    # updates from the refresh task itself). We only need to reschedule when schedule-relevant
+    # fields change or when _cron_expression was explicitly set by the serializer.
+    SCHEDULE_FIELDS = {'refresh_interval', 'is_active', 'refresh_task'}
+    if (
+        not created
+        and update_fields is not None
+        and not (set(update_fields) & SCHEDULE_FIELDS)
+        and not hasattr(instance, '_cron_expression')
+    ):
+        return
+
     task_name = f"m3u_account-refresh-{instance.id}"
     should_be_enabled = instance.is_active
 
-    # Read cron_expression from transient attribute set by the serializer
-    cron_expr = getattr(instance, "_cron_expression", "")
+    # Read cron_expression from transient attribute set by the serializer.
+    # If not set (e.g. save came from a task updating status/last_message),
+    # preserve the existing crontab so we don't accidentally revert to interval.
+    if hasattr(instance, "_cron_expression"):
+        cron_expr = instance._cron_expression
+    else:
+        cron_expr = ""
+        try:
+            existing_task = instance.refresh_task
+            if existing_task and existing_task.crontab:
+                ct = existing_task.crontab
+                cron_expr = f"{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}"
+        except Exception:
+            pass
 
     task = create_or_update_periodic_task(
         task_name=task_name,

apps/m3u/tasks.py

@@ -23,6 +23,7 @@ from core.utils import (
     RedisClient,
     acquire_task_lock,
     release_task_lock,
+    TaskLockRenewer,
     natural_sort_key,
     log_system_event,
 )
@@ -66,7 +67,8 @@ def fetch_m3u_lines(account, use_cache=False):
                 account.save(update_fields=["status", "last_message"])
 
                 response = requests.get(
-                    account.server_url, headers=headers, stream=True
+                    account.server_url, headers=headers, stream=True,
+                    timeout=(30, 60),  # 30s connect, 60s read between chunks
                 )
 
                 # Log the actual response details for debugging
@@ -126,119 +128,60 @@ def fetch_m3u_lines(account, use_cache=False):
                 start_time = time.time()
                 last_update_time = start_time
                 progress = 0
-                temp_content = b""  # Store content temporarily to validate before saving
                 has_content = False
 
-                # First, let's collect the content and validate it
-                send_m3u_update(account.id, "downloading", 0)
-                for chunk in response.iter_content(chunk_size=8192):
-                    if chunk:
-                        temp_content += chunk
-                        has_content = True
+                # Stream directly to a temp file to avoid holding the entire
+                # M3U in memory (large files can be 100MB+, which would use
+                # ~3x that in RAM in certain approaches).
+                temp_path = file_path + ".tmp"
+                try:
+                    send_m3u_update(account.id, "downloading", 0)
+                    with open(temp_path, "wb") as tmp_file:
+                        for chunk in response.iter_content(chunk_size=8192):
+                            if chunk:
+                                tmp_file.write(chunk)
+                                has_content = True
 
-                        downloaded += len(chunk)
-                        elapsed_time = time.time() - start_time
+                                downloaded += len(chunk)
+                                elapsed_time = time.time() - start_time
 
-                        # Calculate download speed in KB/s
-                        speed = downloaded / elapsed_time / 1024  # in KB/s
+                                # Calculate download speed in KB/s
+                                speed = downloaded / elapsed_time / 1024  # in KB/s
 
-                        # Calculate progress percentage
-                        if total_size and total_size > 0:
-                            progress = (downloaded / total_size) * 100
+                                # Calculate progress percentage
+                                if total_size and total_size > 0:
+                                    progress = (downloaded / total_size) * 100
 
-                        # Time remaining (in seconds)
-                        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:
-                                # 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",
-                                    progress,
-                                    speed=speed,
-                                    elapsed_time=elapsed_time,
-                                    time_remaining=time_remaining,
-                                    message=progress_msg,
+                                # Time remaining (in seconds)
+                                time_remaining = (
+                                    (total_size - downloaded) / (speed * 1024)
+                                    if speed > 0
+                                    else 0
                                 )
 
-                # Check if we actually received any content
-                logger.info(f"Download completed. Has content: {has_content}, Content length: {len(temp_content)} bytes")
-                if not has_content or len(temp_content) == 0:
-                    error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
-                    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
+                                current_time = time.time()
+                                if current_time - last_update_time >= 0.5:
+                                    last_update_time = current_time
+                                    if progress > 0:
+                                        # 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"])
 
-                # Basic validation: check if content looks like an M3U file
-                try:
-                    content_str = temp_content.decode('utf-8', errors='ignore')
-                    content_lines = content_str.strip().split('\n')
+                                        send_m3u_update(
+                                            account.id,
+                                            "downloading",
+                                            progress,
+                                            speed=speed,
+                                            elapsed_time=elapsed_time,
+                                            time_remaining=time_remaining,
+                                            message=progress_msg,
+                                        )
 
-                    # Log first few lines for debugging (be careful not to log too much)
-                    preview_lines = content_lines[:5]
-                    logger.info(f"Content preview (first 5 lines): {preview_lines}")
-                    logger.info(f"Total lines in content: {len(content_lines)}")
-
-                    # Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
-                    is_valid_m3u = False
-
-                    # First, check if this looks like an error response disguised as 200 OK
-                    content_lower = content_str.lower()
-                    if any(error_indicator in content_lower for error_indicator in [
-                        '<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
-                        'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
-                    ]):
-                        logger.warning(f"Content appears to be an error response disguised as HTTP 200: {content_str[:200]!r}")
-                        # Continue with M3U validation, but this gives us a clue
-
-                    if content_lines and content_lines[0].strip().upper().startswith('#EXTM3U'):
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: starts with #EXTM3U")
-                    elif any(line.strip().startswith('#EXTINF:') for line in content_lines):
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: contains #EXTINF entries")
-                    elif any(line.strip().startswith('http') for line in content_lines):
-                        # Has HTTP URLs, might be a simple M3U without headers
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: contains HTTP URLs")
-                    elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in content_lines):
-                        # Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
-                        is_valid_m3u = True
-                        logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")
-
-                    if not is_valid_m3u:
-                        # Log what we actually received for debugging
-                        logger.error(f"Invalid M3U content received. First 200 characters: {content_str[:200]!r}")
-
-                        # Try to provide more specific error messages based on content
-                        if '<html' in content_lower or '<!doctype html' in content_lower:
-                            error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
-                        elif 'error' in content_lower or 'not found' in content_lower:
-                            error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {content_str[:100]}"
-                        elif len(content_str.strip()) == 0:
-                            error_msg = f"Server returned completely empty response from URL: {account.server_url}"
-                        else:
-                            error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U file."
+                    # Check if we actually received any content
+                    logger.info(f"Download completed. Has content: {has_content}, Content length: {downloaded} bytes")
+                    if not has_content or downloaded == 0:
+                        error_msg = f"Server responded successfully (HTTP {response.status_code}) but provided empty M3U file from URL: {account.server_url}"
                         logger.error(error_msg)
                         account.status = M3UAccount.Status.ERROR
                         account.last_message = error_msg
@@ -252,31 +195,113 @@ def fetch_m3u_lines(account, use_cache=False):
                         )
                         return [], False
 
-                except UnicodeDecodeError:
-                    logger.error(f"Non-text content received. First 200 bytes: {temp_content[:200]!r}")
-                    error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U 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
+                    # Validate the file by reading only the first portion from
+                    # disk — no need to load the entire file into memory just
+                    # to check the header.
+                    VALIDATION_READ_SIZE = 32768  # 32KB covers headers comfortably
+                    try:
+                        with open(temp_path, "rb") as vf:
+                            head_bytes = vf.read(VALIDATION_READ_SIZE)
+                        head_str = head_bytes.decode('utf-8', errors='ignore')
+                        head_lines = head_str.strip().split('\n')
 
-                # Content is valid, save it to file
-                with open(file_path, "wb") as file:
-                    file.write(temp_content)
+                        # Count total lines efficiently without loading full file
+                        with open(temp_path, "rb") as vf:
+                            total_lines = sum(1 for _ in vf)
 
-                # 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)
+                        # Log first few lines for debugging (be careful not to log too much)
+                        preview_lines = head_lines[:5]
+                        logger.info(f"Content preview (first 5 lines): {preview_lines}")
+                        logger.info(f"Total lines in content: {total_lines}")
+
+                        # Check if it's a valid M3U file (should start with #EXTM3U or contain M3U-like content)
+                        is_valid_m3u = False
+
+                        # First, check if this looks like an error response disguised as 200 OK
+                        head_lower = head_str.lower()
+                        if any(error_indicator in head_lower for error_indicator in [
+                            '<html', '<!doctype html', 'error', 'not found', '404', '403', '500',
+                            'access denied', 'unauthorized', 'forbidden', 'invalid', 'expired'
+                        ]):
+                            logger.warning(f"Content appears to be an error response disguised as HTTP 200: {head_str[:200]!r}")
+                            # Continue with M3U validation, but this gives us a clue
+
+                        if head_lines and head_lines[0].strip().upper().startswith('#EXTM3U'):
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: starts with #EXTM3U")
+                        elif any(line.strip().startswith('#EXTINF:') for line in head_lines):
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: contains #EXTINF entries")
+                        elif any(line.strip().startswith('http') for line in head_lines):
+                            # Has HTTP URLs, might be a simple M3U without headers
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: contains HTTP URLs")
+                        elif any(line.strip().startswith(('rtsp', 'rtp', 'udp')) for line in head_lines):
+                            # Has RTSP/RTP/UDP URLs, might be a simple M3U without headers
+                            is_valid_m3u = True
+                            logger.info("Content validated as M3U: contains RTSP/RTP/UDP URLs")
+
+                        if not is_valid_m3u:
+                            # Log what we actually received for debugging
+                            logger.error(f"Invalid M3U content received. First 200 characters: {head_str[:200]!r}")
+
+                            # Try to provide more specific error messages based on content
+                            if '<html' in head_lower or '<!doctype html' in head_lower:
+                                error_msg = f"Server returned HTML page instead of M3U file from URL: {account.server_url}. This usually indicates an error or authentication issue."
+                            elif 'error' in head_lower or 'not found' in head_lower:
+                                error_msg = f"Server returned an error message instead of M3U file from URL: {account.server_url}. Content: {head_str[:100]}"
+                            elif len(head_str.strip()) == 0:
+                                error_msg = f"Server returned completely empty response from URL: {account.server_url}"
+                            else:
+                                error_msg = f"Server provided invalid M3U content from URL: {account.server_url}. Content does not appear to be a valid M3U 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
+
+                    except UnicodeDecodeError:
+                        with open(temp_path, "rb") as vf:
+                            first_bytes = vf.read(200)
+                        logger.error(f"Non-text content received. First 200 bytes: {first_bytes!r}")
+                        error_msg = f"Server provided non-text content from URL: {account.server_url}. Unable to process as M3U 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
+
+                    # Validation passed — promote temp file to final path
+                    os.replace(temp_path, file_path)
+
+                    # Final update with 100% progress
+                    dl_size = downloaded / 1024 / 1024
+                    final_msg = f"Download complete. Size: {dl_size:.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)
+
+                finally:
+                    # Clean up temp file on any failure path
+                    if os.path.exists(temp_path):
+                        try:
+                            os.remove(temp_path)
+                        except OSError:
+                            pass
             except requests.exceptions.HTTPError as e:
                 # Handle HTTP errors specifically with more context
                 status_code = e.response.status_code if e.response else "unknown"
@@ -1210,9 +1235,13 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
     if not acquire_task_lock("refresh_m3u_account_groups", account_id):
         return f"Task already running for account_id={account_id}.", None
 
+    lock_renewer = TaskLockRenewer("refresh_m3u_account_groups", account_id)
+    lock_renewer.start()
+
     try:
         account = M3UAccount.objects.get(id=account_id, is_active=True)
     except M3UAccount.DoesNotExist:
+        lock_renewer.stop()
         release_task_lock("refresh_m3u_account_groups", account_id)
         return f"M3UAccount with ID={account_id} not found or inactive.", None
 
@@ -1238,6 +1267,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             send_m3u_update(
                 account_id, "processing_groups", 100, status="error", error=error_msg
             )
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
 
@@ -1250,6 +1280,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             send_m3u_update(
                 account_id, "processing_groups", 100, status="error", error=error_msg
             )
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
 
@@ -1359,6 +1390,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
                                 status="error",
                                 error=error_msg,
                             )
+                            lock_renewer.stop()
                             release_task_lock("refresh_m3u_account_groups", account_id)
                             return error_msg, None
 
@@ -1397,6 +1429,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
                             status="error",
                             error=error_msg,
                         )
+                        lock_renewer.stop()
                         release_task_lock("refresh_m3u_account_groups", account_id)
                         return error_msg, None
 
@@ -1413,6 +1446,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
                     status="error",
                     error=error_msg,
                 )
+                lock_renewer.stop()
                 release_task_lock("refresh_m3u_account_groups", account_id)
                 return error_msg, None
         except Exception as e:
@@ -1424,6 +1458,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
             send_m3u_update(
                 account_id, "processing_groups", 100, status="error", error=error_msg
             )
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return error_msg, None
     else:
@@ -1431,6 +1466,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
         lines, success = fetch_m3u_lines(account, use_cache)
         if not success:
             # If fetch failed, don't continue processing
+            lock_renewer.stop()
             release_task_lock("refresh_m3u_account_groups", account_id)
             return f"Failed to fetch M3U data for account_id={account_id}.", None
 
@@ -1525,6 +1561,7 @@ def refresh_m3u_groups(account_id, use_cache=False, full_refresh=False, scan_sta
 
     process_groups(account, groups, scan_start_time)
 
+    lock_renewer.stop()
     release_task_lock("refresh_m3u_account_groups", account_id)
 
     if not full_refresh:
@@ -2593,12 +2630,18 @@ def refresh_account_info(profile_id):
 
         release_task_lock("refresh_account_info", profile_id)
         return error_msg
-@shared_task
+@shared_task(time_limit=3600, soft_time_limit=3500)
 def refresh_single_m3u_account(account_id):
     """Splits M3U processing into chunks and dispatches them as parallel tasks."""
     if not acquire_task_lock("refresh_single_m3u_account", account_id):
         return f"Task already running for account_id={account_id}."
 
+    # Keep the lock alive while this long-running task is working.
+    # Without renewal, the 300s lock TTL can expire during large
+    # downloads/parses, allowing duplicate tasks to start.
+    lock_renewer = TaskLockRenewer("refresh_single_m3u_account", account_id)
+    lock_renewer.start()
+
     # Record start time
     refresh_start_timestamp = timezone.now()  # For the cleanup function
     start_time = time.time()  # For tracking elapsed time as float
@@ -2610,6 +2653,7 @@ def refresh_single_m3u_account(account_id):
         account = M3UAccount.objects.get(id=account_id, is_active=True)
         if not account.is_active:
             logger.debug(f"Account {account_id} is not active, skipping.")
+            lock_renewer.stop()
             release_task_lock("refresh_single_m3u_account", account_id)
             return
 
@@ -2639,6 +2683,7 @@ def refresh_single_m3u_account(account_id):
         else:
             logger.debug(f"No orphaned task found for M3U account {account_id}")
 
+        lock_renewer.stop()
         release_task_lock("refresh_single_m3u_account", account_id)
         return f"M3UAccount with ID={account_id} not found or inactive, task cleaned up"
 
@@ -2689,6 +2734,7 @@ def refresh_single_m3u_account(account_id):
                 logger.error(
                     f"Failed to refresh M3U groups for account {account_id}: {result}"
                 )
+                lock_renewer.stop()
                 release_task_lock("refresh_single_m3u_account", account_id)
                 return "Failed to update m3u account - download failed or other error"
 
@@ -2722,6 +2768,7 @@ def refresh_single_m3u_account(account_id):
                 status="error",
                 error=f"Error refreshing M3U groups: {str(e)}",
             )
+            lock_renewer.stop()
             release_task_lock("refresh_single_m3u_account", account_id)
             return "Failed to update m3u account"
 
@@ -2745,6 +2792,7 @@ def refresh_single_m3u_account(account_id):
             status="error",
             error="No data available for processing",
         )
+        lock_renewer.stop()
         release_task_lock("refresh_single_m3u_account", account_id)
         return "Failed to update m3u account, no data available"
 
@@ -3055,8 +3103,9 @@ def refresh_single_m3u_account(account_id):
         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)
+    finally:
+        lock_renewer.stop()
+        release_task_lock("refresh_single_m3u_account", account_id)
 
     # Aggressive garbage collection
     # Only delete variables if they exist

core/utils.py

@@ -222,6 +222,75 @@ def release_task_lock(task_name, id):
     # Remove the lock
     redis_client.delete(lock_id)
 
+
+class TaskLockRenewer:
+    """Periodically renews a Redis task lock to prevent expiry during long-running tasks.
+
+    Use as a context manager after acquiring a lock:
+
+        if acquire_task_lock("my_task", task_id):
+            with TaskLockRenewer("my_task", task_id):
+                # ... long-running work ...
+            release_task_lock("my_task", task_id)
+
+    A daemon thread extends the lock TTL at regular intervals so that
+    slow downloads or large parsing jobs don't lose their lock mid-operation.
+    """
+
+    def __init__(self, task_name, id, ttl=300, renewal_interval=120):
+        self.task_name = task_name
+        self.id = id
+        self.ttl = ttl
+        self.renewal_interval = renewal_interval
+        self.lock_id = f"task_lock_{task_name}_{id}"
+        self._stop_event = threading.Event()
+        self._thread = None
+
+    def _renew_loop(self):
+        """Background loop that extends the lock TTL until stopped."""
+        while not self._stop_event.wait(self.renewal_interval):
+            try:
+                redis_client = RedisClient.get_client()
+                if redis_client.exists(self.lock_id):
+                    redis_client.expire(self.lock_id, self.ttl)
+                    logger.debug(
+                        f"Renewed lock {self.lock_id} TTL to {self.ttl}s"
+                    )
+                else:
+                    # Lock was deleted externally (e.g. manual release) — stop renewing
+                    logger.warning(
+                        f"Lock {self.lock_id} no longer exists, stopping renewal"
+                    )
+                    break
+            except Exception as e:
+                logger.error(f"Error renewing lock {self.lock_id}: {e}")
+
+    def start(self):
+        """Start the background renewal thread."""
+        self._stop_event.clear()
+        self._thread = threading.Thread(
+            target=self._renew_loop, daemon=True,
+            name=f"lock-renew-{self.task_name}-{self.id}"
+        )
+        self._thread.start()
+        return self
+
+    def stop(self):
+        """Stop the renewal thread."""
+        self._stop_event.set()
+        if self._thread and self._thread.is_alive():
+            self._thread.join(timeout=5)
+        self._thread = None
+
+    def __enter__(self):
+        self.start()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.stop()
+        return False
+
+
 def send_websocket_update(group_name, event_type, data, collect_garbage=False):
     """
     Standardized function to send WebSocket updates with proper memory management.
@@ -404,7 +473,7 @@ def dispatch_event_system(event_type, channel_id=None, channel_name=None, **deta
         from core.models import StreamProfile
         from core.utils import RedisClient
 
-        payload = {}
+        payload = dict(details)
 
         channel_obj = None
         if channel_id:
@@ -464,6 +533,11 @@ def dispatch_event_system(event_type, channel_id=None, channel_name=None, **deta
 
         payload["profile_used"] = profile_used
 
+        # remove empty keys
+        for k in list(payload.keys()):
+            if not payload[k]:
+                del payload[k]
+
         trigger_event(event_type, payload)
 
     except Exception as e:

dispatcharr/settings.py

@@ -168,6 +168,7 @@ REST_FRAMEWORK = {
     ],
     "DEFAULT_AUTHENTICATION_CLASSES": [
         "rest_framework_simplejwt.authentication.JWTAuthentication",
+        "apps.accounts.authentication.ApiKeyAuthentication",
     ],
     "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
 }
@@ -415,7 +416,7 @@ LOGGING = {
 
 # Connect script execution safety settings
 # Allowed base directories for custom scripts; real paths must be inside
-_allowed_dirs_env = os.environ.get("DISPATCHARR_ALLOWED_SCRIPT_DIRS", "/data/plugins")
+_allowed_dirs_env = os.environ.get("DISPATCHARR_ALLOWED_SCRIPT_DIRS", "/data/scripts")
 CONNECT_ALLOWED_SCRIPT_DIRS = [p for p in _allowed_dirs_env.split(":") if p]
 
 # Max execution time (seconds) for scripts

frontend/src/api.js

@@ -13,6 +13,7 @@ import useChannelsTableStore from './store/channelsTable';
 import useStreamsTableStore from './store/streamsTable';
 import useUsersStore from './store/users';
 import useConnectStore from './store/connect';
+import Limiter from './utils';
 
 // If needed, you can set a base host or keep it empty if relative requests
 const host = import.meta.env.DEV
@@ -105,6 +106,41 @@ export default class API {
     return await useAuthStore.getState().getToken();
   }
 
+  /**
+   * Fetch all pages for a paginated endpoint when you already know totalCount.
+   * Builds page calls from totalCount and pageSize and aggregates all results.
+   * - endpoint: path like "/api/channels/channels/"
+   * - params: URLSearchParams for filters (will not be mutated)
+   * - totalCount: total number of matching items
+   * - pageSize: items per page
+   * Returns a flat array of results. Supports both array and {results, next} responses.
+   */
+  static async fetchAllByCount(endpoint, params, totalCount, pageSize = 200) {
+    const total = Number(totalCount) || 0;
+    const size = Number(pageSize) || 200;
+    const totalPages = Math.max(1, Math.ceil(total / size));
+
+    const requests = [];
+    for (let page = 1; page <= totalPages; page++) {
+      const q = new URLSearchParams(params || new URLSearchParams());
+      q.set('page', String(page));
+      q.set('page_size', String(size));
+      const url = `${host}${endpoint}?${q.toString()}`;
+      requests.push(request(url));
+    }
+
+    const responses = await Promise.all(requests);
+    const all = [];
+    for (const data of responses) {
+      if (Array.isArray(data)) {
+        all.push(...data);
+      } else if (Array.isArray(data?.results)) {
+        all.push(...data.results);
+      }
+    }
+    return all;
+  }
+
   static async fetchSuperUser() {
     try {
       return await request(`${host}/api/accounts/initialize-superuser/`, {
@@ -181,32 +217,39 @@ export default class API {
     try {
       // Paginate through channels to avoid heavy single response
       const pageSize = 200;
-      let page = 1;
-      let allChannels = [];
+      const allChannels = [];
 
-      while (true) {
-        const data = await request(
-          `${host}/api/channels/channels/?page=${page}&page_size=${pageSize}`
-        );
+      // Get first page to get total results count
+      const data = await request(
+        `${host}/api/channels/channels/?page=1&page_size=${pageSize}`
+      );
 
-        // Backward compatibility: if endpoint returns an array (legacy), just return it
-        if (Array.isArray(data)) {
-          allChannels = data;
-          break;
-        }
-
-        const results = Array.isArray(data?.results) ? data.results : [];
-        allChannels = allChannels.concat(results);
-
-        const hasMore = Boolean(data?.next);
-        if (!hasMore || results.length === 0) {
-          break;
-        }
-
-        page += 1;
+      // Backward compatibility: if endpoint returns an array (legacy), just return it
+      if (Array.isArray(data)) {
+        return data;
       }
 
-      return allChannels;
+      allChannels.concat(Array.isArray(data?.results) ? data.results : []);
+
+      const totalPages = Math.max(1, Math.ceil(data.count / pageSize)) - 1;
+      const apiCalls = [];
+      for (let page = 2; page <= totalPages; page++) {
+        apiCalls.push(
+          new Promise(async (resolve) => {
+            const response = await request(
+              `${host}/api/channels/channels/?page=${page}&page_size=${pageSize}`
+            );
+
+            return resolve(
+              Array.isArray(response?.results) ? response.results : []
+            );
+          })
+        );
+      }
+
+      const allResults = await Limiter.all(5, apiCalls);
+
+      return allResults;
     } catch (e) {
       errorNotification('Failed to retrieve channels', e);
     }
@@ -2832,6 +2875,62 @@ export default class API {
     }
   }
 
+  static async generateApiKey({ user_id = null, name = '' } = {}) {
+    try {
+      const body = {};
+      if (user_id) body.user_id = user_id;
+      if (name) body.name = name;
+
+      const response = await request(
+        `${host}/api/accounts/api-keys/generate/`,
+        {
+          method: 'POST',
+          body,
+        }
+      );
+
+      // If the backend returned an updated user, refresh the users store
+      try {
+        if (response && response.user) {
+          useUsersStore.getState().updateUser(response.user);
+        }
+      } catch (e) {
+        // ignore store update errors
+      }
+
+      return response;
+    } catch (e) {
+      errorNotification('Failed to generate API key', e);
+    }
+  }
+
+  static async revokeApiKey({ user_id = null } = {}) {
+    try {
+      const body = {};
+      if (user_id) {
+        body.user_id = user_id;
+      }
+
+      const response = await request(`${host}/api/accounts/api-keys/revoke/`, {
+        method: 'POST',
+        body,
+      });
+
+      // If the backend returned an updated user, refresh the users store
+      try {
+        if (response && response.user) {
+          useUsersStore.getState().updateUser(response.user);
+        }
+      } catch (e) {
+        // ignore store update errors
+      }
+
+      return response;
+    } catch (e) {
+      errorNotification('Failed to revoke API key', e);
+    }
+  }
+
   static async updateUser(id, body) {
     try {
       const response = await request(`${host}/api/accounts/users/${id}/`, {

frontend/src/components/Sidebar.jsx

@@ -5,7 +5,6 @@ import {
   ListOrdered,
   Play,
   Database,
-  SlidersHorizontal,
   LayoutGrid,
   Settings as LucideSettings,
   Copy,
@@ -20,6 +19,7 @@ import {
   ChevronDown,
   ChevronRight,
   MonitorCog,
+  Blocks,
 } from 'lucide-react';
 import {
   Avatar,
@@ -31,10 +31,8 @@ import {
   UnstyledButton,
   TextInput,
   ActionIcon,
-  Menu,
   ScrollArea,
 } from '@mantine/core';
-import { notifications } from '@mantine/notifications';
 import logo from '../images/logo.png';
 import useChannelsStore from '../store/channels';
 import './sidebar.css';
@@ -185,8 +183,8 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => {
           { label: 'Stats', icon: <ChartLine size={20} />, path: '/stats' },
           { label: 'Plugins', icon: <PlugZap size={20} />, path: '/plugins' },
           {
-            label: 'Connect',
-            icon: <Webhook size={20} />,
+            label: 'Integrations',
+            icon: <Blocks size={20} />,
             paths: [
               {
                 label: 'Connections',
@@ -202,7 +200,7 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => {
           },
           {
             label: 'System',
-            icon: <MonitorCog size={20} />,
+            icon: <LucideSettings size={20} />,
             paths: [
               {
                 label: 'Users',
@@ -216,7 +214,7 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => {
               },
               {
                 label: 'Settings',
-                icon: <LucideSettings size={20} />,
+                icon: <MonitorCog size={20} />,
                 path: '/settings',
               },
             ],
@@ -247,11 +245,6 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => {
     });
   };
 
-  const onLogout = async () => {
-    await logout();
-    window.location.reload();
-  };
-
   return (
     <AppShell.Navbar
       width={{ base: collapsed ? miniDrawerWidth : drawerWidth }}

frontend/src/components/forms/User.jsx

@@ -15,12 +15,17 @@ import {
   Switch,
   Box,
   Tooltip,
+  Grid,
+  SimpleGrid,
+  useMantineTheme,
 } from '@mantine/core';
-import { RotateCcwKey, X } from 'lucide-react';
+import { RotateCcwKey, RotateCw, X } from 'lucide-react';
+import { Copy, Key } from 'lucide-react';
 import { useForm } from '@mantine/form';
 import useChannelsStore from '../../store/channels';
 import { USER_LEVELS, USER_LEVEL_LABELS } from '../../constants';
 import useAuthStore from '../../store/auth';
+import { copyToClipboard } from '../../utils';
 
 const User = ({ user = null, isOpen, onClose }) => {
   const profiles = useChannelsStore((s) => s.profiles);
@@ -29,6 +34,11 @@ const User = ({ user = null, isOpen, onClose }) => {
 
   const [, setEnableXC] = useState(false);
   const [selectedProfiles, setSelectedProfiles] = useState(new Set());
+  const [generating, setGenerating] = useState(false);
+  const [generatedKey, setGeneratedKey] = useState(null);
+  const [userAPIKey, setUserAPIKey] = useState(user?.api_key || null);
+
+  const theme = useMantineTheme();
 
   const form = useForm({
     mode: 'uncontrolled',
@@ -115,6 +125,7 @@ const User = ({ user = null, isOpen, onClose }) => {
     }
 
     form.reset();
+    setUserAPIKey(null);
     onClose();
   };
 
@@ -139,6 +150,8 @@ const User = ({ user = null, isOpen, onClose }) => {
       if (customProps.xc_password) {
         setEnableXC(true);
       }
+
+      setUserAPIKey(user.api_key || null);
     } else {
       form.reset();
     }
@@ -158,6 +171,62 @@ const User = ({ user = null, isOpen, onClose }) => {
   const isEditingSelf = authUser.id === user?.id;
   const showPermissions = isAdmin && !isEditingSelf;
 
+  const canGenerateKey =
+    authUser.user_level == USER_LEVELS.ADMIN || authUser.id === user?.id;
+
+  const onGenerateKey = async () => {
+    if (!canGenerateKey) {
+      return;
+    }
+
+    setGenerating(true);
+    try {
+      const payload = {};
+      if (authUser.user_level == USER_LEVELS.ADMIN && user?.id) {
+        payload.user_id = user.id;
+      }
+
+      const resp = await API.generateApiKey(payload);
+      const newKey = resp && (resp.key || resp.raw_key);
+      if (newKey) {
+        setGeneratedKey(newKey);
+        setUserAPIKey(newKey);
+      }
+    } catch (e) {
+      // API shows notifications
+    } finally {
+      setGenerating(false);
+    }
+  };
+
+  const onRevokeKey = async () => {
+    if (!canGenerateKey) return;
+
+    setGenerating(true);
+    try {
+      const payload = {};
+      if (authUser.user_level == USER_LEVELS.ADMIN && user?.id) {
+        payload.user_id = user.id;
+      }
+
+      const resp = await API.revokeApiKey(payload);
+      // backend returns { success: true } - clear local state
+      if (resp && resp.success) {
+        setGeneratedKey(null);
+        setUserAPIKey(null);
+
+        // If we're revoking the current authenticated user's key, update auth store
+        if (user?.id && authUser?.id === user.id) {
+          setUser({ ...authUser, api_key: null });
+        }
+      }
+    } catch (e) {
+      // API shows notifications
+    } finally {
+      setGenerating(false);
+    }
+  };
+
   return (
     <Modal opened={isOpen} onClose={onClose} title="User" size="xl">
       <form onSubmit={form.onSubmit(onSubmit)}>
@@ -271,6 +340,61 @@ const User = ({ user = null, isOpen, onClose }) => {
                 </Tooltip>
               </Box>
             )}
+
+            {canGenerateKey && (
+              <Stack>
+                {userAPIKey && (
+                  <TextInput
+                    label="API Key"
+                    disabled={true}
+                    value={userAPIKey}
+                    rightSection={
+                      <ActionIcon
+                        variant="transparent"
+                        size="sm"
+                        color="white"
+                        onClick={() =>
+                          copyToClipboard(userAPIKey, {
+                            successTitle: 'API Key Copied!',
+                            successMessage:
+                              'The API Key has been copied to your clipboard.',
+                          })
+                        }
+                      >
+                        <Copy />
+                      </ActionIcon>
+                    }
+                  />
+                )}
+
+                <Group gap="xs" grow>
+                  <Button
+                    leftSection={<Key size={14} />}
+                    size="xs"
+                    onClick={onGenerateKey}
+                    loading={generating}
+                    variant="light"
+                    fullWidth
+                  >
+                    {userAPIKey ? 'Regenerate API Key' : 'Generate API Key'}
+                  </Button>
+
+                  {userAPIKey && (
+                    <Button
+                      leftSection={<X size={14} />}
+                      size="xs"
+                      onClick={onRevokeKey}
+                      loading={generating}
+                      color={theme.colors.red[5]}
+                      variant="light"
+                      fullWidth
+                    >
+                      Revoke API Key
+                    </Button>
+                  )}
+                </Group>
+              </Stack>
+            )}
           </Stack>
         </Group>