diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d8f4a3a7..bf5b6040 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ab8110f..f077d845 100644 --- a/CHANGELOG.md +++ b/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 ` header or the `X-API-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 diff --git a/apps/accounts/api_urls.py b/apps/accounts/api_urls.py index dda3832c..27de8eb7 100644 --- a/apps/accounts/api_urls.py +++ b/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"}) diff --git a/apps/accounts/api_views.py b/apps/accounts/api_views.py index 82e914e0..cf2d9225 100644 --- a/apps/accounts/api_views.py +++ b/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", diff --git a/apps/accounts/authentication.py b/apps/accounts/authentication.py new file mode 100644 index 00000000..5380af49 --- /dev/null +++ b/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 ` or `X-API-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 diff --git a/apps/accounts/migrations/0004_user_api_key.py b/apps/accounts/migrations/0004_user_api_key.py new file mode 100644 index 00000000..fee7c983 --- /dev/null +++ b/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), + ), + ] diff --git a/apps/accounts/models.py b/apps/accounts/models.py index b4bbc7f6..e04d66ed 100644 --- a/apps/accounts/models.py +++ b/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 diff --git a/apps/accounts/serializers.py b/apps/accounts/serializers.py index b9980f0e..05e11ebb 100644 --- a/apps/accounts/serializers.py +++ b/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", diff --git a/apps/epg/serializers.py b/apps/epg/serializers.py index 60ce097b..07775809 100644 --- a/apps/epg/serializers.py +++ b/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) diff --git a/apps/epg/signals.py b/apps/epg/signals.py index 89deaa66..97992ef3 100644 --- a/apps/epg/signals.py +++ b/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, diff --git a/apps/epg/tasks.py b/apps/epg/tasks.py index 97552171..9dc597d3 100644 --- a/apps/epg/tasks.py +++ b/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) diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index 8bfa7635..22c4057a 100644 --- a/apps/m3u/serializers.py +++ b/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 diff --git a/apps/m3u/signals.py b/apps/m3u/signals.py index ced6f754..3de67c90 100644 --- a/apps/m3u/signals.py +++ b/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, diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index dc178317..a9413143 100644 --- a/apps/m3u/tasks.py +++ b/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 [ - ' { + 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}/`, { diff --git a/frontend/src/components/Sidebar.jsx b/frontend/src/components/Sidebar.jsx index 04453b08..49e54b45 100644 --- a/frontend/src/components/Sidebar.jsx +++ b/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: , path: '/stats' }, { label: 'Plugins', icon: , path: '/plugins' }, { - label: 'Connect', - icon: , + label: 'Integrations', + icon: , paths: [ { label: 'Connections', @@ -202,7 +200,7 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { }, { label: 'System', - icon: , + icon: , paths: [ { label: 'Users', @@ -216,7 +214,7 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { }, { label: 'Settings', - icon: , + icon: , path: '/settings', }, ], @@ -247,11 +245,6 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { }); }; - const onLogout = async () => { - await logout(); - window.location.reload(); - }; - return ( { 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 (
@@ -271,6 +340,61 @@ const User = ({ user = null, isOpen, onClose }) => { )} + + {canGenerateKey && ( + + {userAPIKey && ( + + copyToClipboard(userAPIKey, { + successTitle: 'API Key Copied!', + successMessage: + 'The API Key has been copied to your clipboard.', + }) + } + > + + + } + /> + )} + + + + + {userAPIKey && ( + + )} + + + )}