from rest_framework import viewsets, status from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.permissions import AllowAny from rest_framework.decorators import action from rest_framework_simplejwt.authentication import JWTAuthentication from apps.accounts.authentication import ApiKeyAuthentication, QueryParamJWTAuthentication from rest_framework.parsers import MultiPartParser, FormParser, JSONParser from drf_spectacular.utils import extend_schema, OpenApiParameter, inline_serializer from drf_spectacular.types import OpenApiTypes from rest_framework import serializers from django.shortcuts import get_object_or_404, get_list_or_404 from django.db import connection, transaction from django.db.models import Count, F, Prefetch from django.db.models import Q import os, json, requests, logging, mimetypes, threading, time from urllib.parse import urlencode from datetime import timedelta from django.utils.http import http_date from apps.accounts.permissions import ( Authenticated, IsAdmin, IsOwnerOfObject, permission_classes_by_action, permission_classes_by_method, ) from core.models import UserAgent, CoreSettings from core.utils import RedisClient, safe_upload_path from .models import ( Stream, Channel, ChannelGroup, ChannelStream, Logo, ChannelProfile, ChannelProfileMembership, Recording, RecurringRecordingRule, ) from .serializers import ( StreamSerializer, ChannelSerializer, ChannelGroupSerializer, LogoSerializer, ChannelProfileMembershipSerializer, BulkChannelProfileMembershipSerializer, ChannelProfileSerializer, RecordingSerializer, RecurringRecordingRuleSerializer, ) from .tasks import ( match_epg_channels, evaluate_series_rules_impl, match_single_channel_epg, match_selected_channels_epg, sync_recurring_rule_impl, purge_recurring_rule_impl, ) import django_filters from django_filters.rest_framework import DjangoFilterBackend from rest_framework.filters import SearchFilter, OrderingFilter from apps.epg.models import EPGData from apps.vod.models import Movie, Series from django.db.models import Q from django.http import HttpResponse, StreamingHttpResponse, FileResponse, Http404, JsonResponse, HttpResponseRedirect from django.utils import timezone import mimetypes from django.conf import settings from rest_framework.pagination import PageNumberPagination from dispatcharr.utils import network_access_allowed logger = logging.getLogger(__name__) # Negative cache for remote logo URLs that failed to fetch. # Prevents repeated blocking requests to unreachable hosts (e.g., dead CDNs) # from exhausting Daphne workers. Keyed by URL, value is expiry timestamp. _logo_fetch_failures = {} _LOGO_FAIL_TTL = 300 # seconds class OrInFilter(django_filters.Filter): """ Custom filter that handles the OR condition instead of AND. """ def filter(self, queryset, value): if value: # Create a Q object for each value and combine them with OR query = Q() for val in value.split(","): query |= Q(**{self.field_name: val}) return queryset.filter(query) return queryset class StreamPagination(PageNumberPagination): page_size = 50 # Default page size to match frontend default page_size_query_param = "page_size" # Allow clients to specify page size max_page_size = 10000 # Prevent excessive page sizes class StreamFilter(django_filters.FilterSet): name = django_filters.CharFilter(lookup_expr="icontains") channel_group_name = OrInFilter( field_name="channel_group__name", lookup_expr="icontains" ) m3u_account = django_filters.BaseInFilter(field_name="m3u_account__id") m3u_account_name = django_filters.CharFilter( field_name="m3u_account__name", lookup_expr="icontains" ) m3u_account_is_active = django_filters.BooleanFilter( field_name="m3u_account__is_active" ) tvg_id = django_filters.CharFilter(lookup_expr="icontains") class Meta: model = Stream fields = [ "name", "channel_group_name", "m3u_account", "m3u_account_name", "m3u_account_is_active", "tvg_id", ] # ───────────────────────────────────────────────────────── # 1) Stream API (CRUD) # ───────────────────────────────────────────────────────── class StreamViewSet(viewsets.ModelViewSet): queryset = Stream.objects.all() serializer_class = StreamSerializer pagination_class = StreamPagination filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter] filterset_class = StreamFilter search_fields = ["name", "channel_group__name"] ordering_fields = ["name", "channel_group__name", "m3u_account__name", "tvg_id"] ordering = ["-name"] def get_permissions(self): if self.action == "duplicate": return [IsAdmin()] try: return [perm() for perm in permission_classes_by_action[self.action]] except KeyError: return [Authenticated()] def get_queryset(self): qs = super().get_queryset() # Exclude streams from inactive M3U accounts qs = qs.exclude(m3u_account__is_active=False) assigned = self.request.query_params.get("assigned") if assigned is not None: qs = qs.filter(channels__id=assigned) unassigned = self.request.query_params.get("unassigned") if unassigned and str(unassigned).lower() in ("1", "true", "yes", "on"): # Use annotation with Count for better performance on large datasets qs = qs.annotate(channel_count=Count('channels')).filter(channel_count=0) channel_group = self.request.query_params.get("channel_group") if channel_group: group_names = channel_group.split(",") qs = qs.filter(channel_group__name__in=group_names) # Allow client to hide stale streams (streams marked as is_stale=True) hide_stale = self.request.query_params.get("hide_stale") if hide_stale and str(hide_stale).lower() in ("1", "true", "yes", "on"): qs = qs.filter(is_stale=False) return qs def list(self, request, *args, **kwargs): ids = request.query_params.get("ids", None) if ids: ids = ids.split(",") streams = get_list_or_404(Stream, id__in=ids) serializer = self.get_serializer(streams, many=True) return Response(serializer.data) return super().list(request, *args, **kwargs) @action(detail=False, methods=["get"], url_path="ids") def get_ids(self, request, *args, **kwargs): # Get the filtered queryset queryset = self.get_queryset() # Apply filtering, search, and ordering queryset = self.filter_queryset(queryset) # Return only the IDs from the queryset stream_ids = queryset.values_list("id", flat=True) return JsonResponse(list(stream_ids), safe=False) @extend_schema( parameters=[ OpenApiParameter( name="channel_group", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, required=True, description="Channel group name to scope the preview to.", ), OpenApiParameter( name="find", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, required=False, description=( "Find regex for the rename preview. When supplied, " "the response includes find_matches and find_match_count." ), ), OpenApiParameter( name="replace", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, required=False, description=( "Replacement string used with the find pattern. " "Defaults to empty string when omitted." ), ), OpenApiParameter( name="match", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, required=False, description=( "Filter regex for the include preview. When supplied, " "the response includes filter_matches and filter_match_count." ), ), OpenApiParameter( name="exclude", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, required=False, description=( "Filter regex for the exclude preview. When supplied, " "the response includes exclude_matches and " "exclude_match_count." ), ), OpenApiParameter( name="limit", type=OpenApiTypes.INT, location=OpenApiParameter.QUERY, required=False, description="Max preview entries per match list (default 10, capped at 50).", ), ], responses={ 200: inline_serializer( name="StreamRegexPreviewResponse", fields={ "total_in_group": serializers.IntegerField(), "total_scanned": serializers.IntegerField(), "scan_limit_hit": serializers.BooleanField(), "find_matches": serializers.ListField( child=serializers.DictField(), required=False ), "find_match_count": serializers.IntegerField(required=False), "filter_matches": serializers.ListField( child=serializers.DictField(), required=False ), "filter_match_count": serializers.IntegerField(required=False), "exclude_matches": serializers.ListField( child=serializers.DictField(), required=False ), "exclude_match_count": serializers.IntegerField(required=False), "find_error": serializers.CharField(required=False), "match_error": serializers.CharField(required=False), "exclude_error": serializers.CharField(required=False), }, ) }, description=( "Returns regex preview info for a group's streams. Used by the " "auto-sync gear modal so users can see how their find/replace " "or filter pattern affects real stream names before saving. " "Caps in-memory iteration at SCAN_CAP streams per call so the " "endpoint stays bounded even on groups with tens of thousands " "of streams; the caller surfaces total_in_group and " "scan_limit_hit so users know whether the preview is complete." ), ) @action(detail=False, methods=["get"], url_path="regex-preview") def regex_preview(self, request, *args, **kwargs): # `regex` (third-party) supports a per-call timeout that bounds # catastrophic backtracking; paired with PATTERN_MAX_LEN to keep # the endpoint safe under adversarial input. import regex as re SCAN_CAP = 5000 PATTERN_MAX_LEN = 512 REGEX_TIMEOUT = 0.1 group_name = request.query_params.get("channel_group") if not group_name: return Response( {"detail": "channel_group is required"}, status=status.HTTP_400_BAD_REQUEST, ) # Group names are not unique across M3U accounts (two providers # can both publish a "Sports" group). Scope to the calling # account so the sample reflects only the user's edits. m3u_account_id = request.query_params.get("m3u_account_id") if m3u_account_id is not None: try: m3u_account_id = int(m3u_account_id) except (TypeError, ValueError): return Response( {"detail": "m3u_account_id must be an integer"}, status=status.HTTP_400_BAD_REQUEST, ) find_pat = request.query_params.get("find") or None replace_pat = request.query_params.get("replace") or "" match_pat = request.query_params.get("match") or None exclude_pat = request.query_params.get("exclude") or None for label, value in ( ("find", find_pat), ("replace", replace_pat), ("match", match_pat), ("exclude", exclude_pat), ): if value is not None and len(value) > PATTERN_MAX_LEN: return Response( {"detail": f"{label} exceeds {PATTERN_MAX_LEN} characters"}, status=status.HTTP_400_BAD_REQUEST, ) try: limit = int(request.query_params.get("limit", 10)) except (TypeError, ValueError): limit = 10 limit = max(1, min(limit, 50)) find_re = None match_re = None exclude_re = None find_error = None match_error = None exclude_error = None if find_pat: try: find_re = re.compile(find_pat) except re.error as e: find_error = str(e) if match_pat: try: match_re = re.compile(match_pat) except re.error as e: match_error = str(e) if exclude_pat: try: exclude_re = re.compile(exclude_pat) except re.error as e: exclude_error = str(e) # Capped at SCAN_CAP to bound memory on huge groups; the # separate COUNT lets the client surface scan_limit_hit when # the preview covers only a sample. base_qs = Stream.objects.filter(channel_group__name=group_name) if m3u_account_id is not None: base_qs = base_qs.filter(m3u_account_id=m3u_account_id) names_iter = base_qs.values_list("name", flat=True)[:SCAN_CAP] total_in_group = base_qs.count() find_matches = [] filter_matches = [] exclude_matches = [] find_match_count = 0 filter_match_count = 0 exclude_match_count = 0 total_scanned = 0 # Abort a pattern on timeout to bound CPU; partial counts and # an `*_error` field still flow back to the client. for name in names_iter: total_scanned += 1 if find_re is not None: try: new_name = find_re.sub(replace_pat, name, timeout=REGEX_TIMEOUT) except (TimeoutError, re.error) as e: find_error = find_error or f"Pattern timed out: {e}" find_re = None continue if new_name != name: find_match_count += 1 if len(find_matches) < limit: find_matches.append({"before": name, "after": new_name}) if match_re is not None: try: matched = match_re.search(name, timeout=REGEX_TIMEOUT) except (TimeoutError, re.error) as e: match_error = match_error or f"Pattern timed out: {e}" match_re = None continue if matched: filter_match_count += 1 if len(filter_matches) < limit: filter_matches.append({"name": name, "matches": True}) if exclude_re is not None: try: matched = exclude_re.search(name, timeout=REGEX_TIMEOUT) except (TimeoutError, re.error) as e: exclude_error = exclude_error or f"Pattern timed out: {e}" exclude_re = None continue if matched: exclude_match_count += 1 if len(exclude_matches) < limit: exclude_matches.append({"name": name, "matches": True}) response_payload = { "total_in_group": total_in_group, "total_scanned": total_scanned, "scan_limit_hit": total_in_group > SCAN_CAP, } if find_pat: response_payload["find_matches"] = find_matches response_payload["find_match_count"] = find_match_count if find_error: response_payload["find_error"] = find_error if match_pat: response_payload["filter_matches"] = filter_matches response_payload["filter_match_count"] = filter_match_count if match_error: response_payload["match_error"] = match_error if exclude_pat: response_payload["exclude_matches"] = exclude_matches response_payload["exclude_match_count"] = exclude_match_count if exclude_error: response_payload["exclude_error"] = exclude_error return Response(response_payload) @action(detail=False, methods=["get"], url_path="groups") def get_groups(self, request, *args, **kwargs): # Get unique ChannelGroup names that are linked to streams group_names = ( ChannelGroup.objects.filter(streams__isnull=False) .order_by("name") .values_list("name", flat=True) .distinct() ) # Return the response with the list of unique group names return Response(list(group_names)) @action(detail=False, methods=["get"], url_path="filter-options") def get_filter_options(self, request, *args, **kwargs): """ Get available filter options based on current filter state. Uses a hierarchical approach: M3U is the parent filter, Group filters based on M3U. """ # Fast path: no filters supplied - skip DISTINCT over the full streams # table and answer from parent tables via EXISTS semi-joins instead. _group_filter_params = ( "name", "m3u_account", "m3u_account_name", "m3u_account_is_active", "tvg_id", ) _m3u_filter_params = ( "name", "m3u_account_name", "m3u_account_is_active", "tvg_id", ) _has_group_filters = any(request.GET.get(p) for p in _group_filter_params) _has_m3u_filters = any(request.GET.get(p) for p in _m3u_filter_params) if not _has_group_filters and not _has_m3u_filters: base_qs = Stream.objects.exclude(m3u_account__is_active=False) group_names = list( base_qs.exclude(channel_group__isnull=True) .order_by("channel_group__name") .values_list("channel_group__name", flat=True) .distinct() ) m3u_data = list( base_qs.exclude(m3u_account__isnull=True) .order_by("m3u_account__name") .values("m3u_account__id", "m3u_account__name") .distinct() ) return Response({ "groups": group_names, "m3u_accounts": [ {"id": m["m3u_account__id"], "name": m["m3u_account__name"]} for m in m3u_data ], }) # For group options: we need to bypass the channel_group custom queryset filtering # Store original request params original_params = request.query_params # Create modified params without channel_group for getting group options params_without_group = request.GET.copy() params_without_group.pop('channel_group', None) params_without_group.pop('channel_group_name', None) # Temporarily modify request to exclude channel_group request._request.GET = params_without_group base_queryset_for_groups = self.get_queryset() # Apply filterset (which will apply M3U filters) group_filterset = self.filterset_class( params_without_group, queryset=base_queryset_for_groups ) group_queryset = group_filterset.qs group_names = ( group_queryset.exclude(channel_group__isnull=True) .order_by("channel_group__name") .values_list("channel_group__name", flat=True) .distinct() ) # For M3U options: show ALL M3Us (don't filter by anything except name search) params_for_m3u = request.GET.copy() params_for_m3u.pop('m3u_account', None) params_for_m3u.pop('channel_group', None) params_for_m3u.pop('channel_group_name', None) # Temporarily modify request to exclude filters for M3U options request._request.GET = params_for_m3u base_queryset_for_m3u = self.get_queryset() m3u_filterset = self.filterset_class( params_for_m3u, queryset=base_queryset_for_m3u ) m3u_queryset = m3u_filterset.qs m3u_accounts = ( m3u_queryset.exclude(m3u_account__isnull=True) .order_by("m3u_account__name") .values("m3u_account__id", "m3u_account__name") .distinct() ) # Restore original params request._request.GET = original_params return Response({ "groups": list(group_names), "m3u_accounts": [ {"id": m3u["m3u_account__id"], "name": m3u["m3u_account__name"]} for m3u in m3u_accounts ] }) @extend_schema( methods=["POST"], description="Retrieve streams by a list of IDs using POST to avoid URL length limitations", request=inline_serializer( name="StreamByIdsRequest", fields={ "ids": serializers.ListField( child=serializers.IntegerField(), help_text="List of stream IDs to retrieve" ), }, ), responses={200: StreamSerializer(many=True)}, ) @action(detail=False, methods=["post"], url_path="by-ids") def get_by_ids(self, request, *args, **kwargs): ids = request.data.get("ids", []) if not isinstance(ids, list): return Response( {"error": "ids must be a list of integers"}, status=status.HTTP_400_BAD_REQUEST, ) streams = Stream.objects.filter(id__in=ids) serializer = self.get_serializer(streams, many=True) return Response(serializer.data) # ───────────────────────────────────────────────────────── # 2) Channel Group Management (CRUD) # ───────────────────────────────────────────────────────── class ChannelGroupViewSet(viewsets.ModelViewSet): queryset = ChannelGroup.objects.all() serializer_class = ChannelGroupSerializer def get_permissions(self): try: return [perm() for perm in permission_classes_by_action[self.action]] except KeyError: return [Authenticated()] def get_queryset(self): # Annotate both counts at the SQL level so the serializer methods # can read them from the object rather than issuing a COUNT per row. # `distinct=True` is required when multiple reverse-FK annotations # share the same queryset to avoid row-multiplication artifacts. # m3u_accounts is still prefetched for the nested serializer data. return ( ChannelGroup.objects .annotate( channel_count=Count('channels', distinct=True), m3u_account_count=Count('m3u_accounts', distinct=True), ) .prefetch_related('m3u_accounts') .all() ) def list(self, request, *args, **kwargs): queryset = self.filter_queryset(self.get_queryset()) # Evaluate the queryset once so the annotations and prefetch cache are # populated together, then extract IDs from the in-memory objects. # A second .values_list() call would fire a separate SQL query. groups = list(queryset) group_ids = [g.id for g in groups] # Pre-aggregate stream counts for all (account, group) pairs in a # single query so the nested ChannelGroupM3UAccountSerializer never # fires a COUNT per row. counts_qs = ( Stream.objects.filter(channel_group_id__in=group_ids) .values('m3u_account_id', 'channel_group_id') .annotate(c=Count('id')) ) stream_counts = { (row['m3u_account_id'], row['channel_group_id']): row['c'] for row in counts_qs } page = self.paginate_queryset(groups) if page is not None: serializer = self.get_serializer( page, many=True, context={**self.get_serializer_context(), 'stream_counts': stream_counts}, ) return self.get_paginated_response(serializer.data) serializer = self.get_serializer( groups, many=True, context={**self.get_serializer_context(), 'stream_counts': stream_counts}, ) return Response(serializer.data) def update(self, request, *args, **kwargs): """Override update to check M3U associations""" instance = self.get_object() # Check if group has M3U account associations if hasattr(instance, 'm3u_account') and instance.m3u_account.exists(): return Response( {"error": "Cannot edit group with M3U account associations"}, status=status.HTTP_400_BAD_REQUEST ) return super().update(request, *args, **kwargs) def partial_update(self, request, *args, **kwargs): """Override partial_update to check M3U associations""" instance = self.get_object() # Check if group has M3U account associations if hasattr(instance, 'm3u_account') and instance.m3u_account.exists(): return Response( {"error": "Cannot edit group with M3U account associations"}, status=status.HTTP_400_BAD_REQUEST ) return super().partial_update(request, *args, **kwargs) @extend_schema( methods=["POST"], description="Delete all channel groups that have no associations (no channels or M3U accounts)", ) @action(detail=False, methods=["post"], url_path="cleanup") def cleanup_unused_groups(self, request): """Delete all channel groups with no channels or M3U account associations""" from django.db.models import Q, Exists, OuterRef # Find groups with no channels and no M3U account associations using Exists subqueries from .models import Channel, ChannelGroupM3UAccount has_channels = Channel.objects.filter(channel_group_id=OuterRef('pk')) has_accounts = ChannelGroupM3UAccount.objects.filter(channel_group_id=OuterRef('pk')) unused_groups = ChannelGroup.objects.annotate( has_channels=Exists(has_channels), has_accounts=Exists(has_accounts) ).filter( has_channels=False, has_accounts=False ) deleted_count = unused_groups.count() group_names = list(unused_groups.values_list('name', flat=True)) # Delete the unused groups unused_groups.delete() return Response({ "message": f"Successfully deleted {deleted_count} unused channel groups", "deleted_count": deleted_count, "deleted_groups": group_names }) def destroy(self, request, *args, **kwargs): """Override destroy to check for associations before deletion""" instance = self.get_object() # Check if group has associated channels if instance.channels.exists(): return Response( {"error": "Cannot delete group with associated channels"}, status=status.HTTP_400_BAD_REQUEST ) # Check if group has M3U account associations if hasattr(instance, 'm3u_account') and instance.m3u_account.exists(): return Response( {"error": "Cannot delete group with M3U account associations"}, status=status.HTTP_400_BAD_REQUEST ) return super().destroy(request, *args, **kwargs) # ───────────────────────────────────────────────────────── # 3) Channel Management (CRUD) # ───────────────────────────────────────────────────────── class ChannelPagination(PageNumberPagination): page_size = 50 # Default page size to match frontend default page_size_query_param = "page_size" # Allow clients to specify page size max_page_size = 10000 # Prevent excessive page sizes def paginate_queryset(self, queryset, request, view=None): if not request.query_params.get(self.page_query_param): return None # disables pagination, returns full queryset return super().paginate_queryset(queryset, request, view) def get_paginated_response(self, data): from django.db.models import Exists, OuterRef has_unassigned = Channel.objects.filter(epg_data__isnull=True).exists() response = super().get_paginated_response(data) response.data['has_unassigned_epg_channels'] = has_unassigned return response class EPGFilter(django_filters.Filter): """ Filter channels by EPG source name or null (unlinked). """ def filter(self, queryset, value): if not value: return queryset # Split comma-separated values values = [v.strip() for v in value.split(',')] query = Q() for val in values: if val == 'null': # Filter for channels with no EPG data query |= Q(epg_data__isnull=True) else: # Filter for channels with specific EPG source name query |= Q(epg_data__epg_source__name__icontains=val) return queryset.filter(query) class ChannelFilter(django_filters.FilterSet): name = django_filters.CharFilter(lookup_expr="icontains") channel_group = OrInFilter( field_name="channel_group__name", lookup_expr="icontains" ) epg = EPGFilter() class Meta: model = Channel fields = [ "name", "channel_group", "epg", ] class ChannelViewSet(viewsets.ModelViewSet): queryset = Channel.objects.all() serializer_class = ChannelSerializer pagination_class = ChannelPagination filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter] filterset_class = ChannelFilter search_fields = ["name", "channel_group__name"] ordering_fields = ["channel_number", "name", "channel_group__name", "epg_data__name"] ordering = ["-channel_number"] def create(self, request, *args, **kwargs): """Override create to handle channel profile membership""" serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) with transaction.atomic(): channel = serializer.save() # Handle channel profile membership # Semantics: # - Omitted (None): add to ALL profiles (backward compatible default) # - Empty array []: add to NO profiles # - Sentinel [0] or 0: add to ALL profiles (explicit) # - [1,2,...]: add to specified profile IDs only channel_profile_ids = request.data.get("channel_profile_ids") if channel_profile_ids is not None: # Normalize single ID to array if not isinstance(channel_profile_ids, list): channel_profile_ids = [channel_profile_ids] # Determine action based on semantics if channel_profile_ids is None: # Omitted -> add to all profiles (backward compatible) profiles = ChannelProfile.objects.all() ChannelProfileMembership.objects.bulk_create([ ChannelProfileMembership(channel_profile=profile, channel=channel, enabled=True) for profile in profiles ]) elif isinstance(channel_profile_ids, list) and len(channel_profile_ids) == 0: # Empty array -> add to no profiles pass elif isinstance(channel_profile_ids, list) and 0 in channel_profile_ids: # Sentinel 0 -> add to all profiles (explicit) profiles = ChannelProfile.objects.all() ChannelProfileMembership.objects.bulk_create([ ChannelProfileMembership(channel_profile=profile, channel=channel, enabled=True) for profile in profiles ]) else: # Specific profile IDs try: channel_profiles = ChannelProfile.objects.filter(id__in=channel_profile_ids) if len(channel_profiles) != len(channel_profile_ids): missing_ids = set(channel_profile_ids) - set(channel_profiles.values_list('id', flat=True)) return Response( {"error": f"Channel profiles with IDs {list(missing_ids)} not found"}, status=status.HTTP_400_BAD_REQUEST, ) ChannelProfileMembership.objects.bulk_create([ ChannelProfileMembership( channel_profile=profile, channel=channel, enabled=True ) for profile in channel_profiles ]) except Exception as e: return Response( {"error": f"Error creating profile memberships: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST, ) headers = self.get_success_headers(serializer.data) return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers) def get_permissions(self): if self.action in [ "edit_bulk", "assign", "from_stream", "from_stream_bulk", "match_epg", "set_epg", "batch_set_epg", ]: return [IsAdmin()] try: return [perm() for perm in permission_classes_by_action[self.action]] except KeyError: return [Authenticated()] def get_queryset(self): # get_ids and summary only need the filter conditions, not the full # object graph. Skipping the 5 select_related joins and 2 prefetch # queries for those actions cuts their DB cost significantly. action = getattr(self, "action", None) qs = super().get_queryset() if action not in ("get_ids", "summary"): qs = qs.select_related( "channel_group", "logo", "epg_data", "stream_profile", "override", "auto_created_by", ).prefetch_related( "streams", # Default-attr prefetch shares the cache with M2M writes; # a named `to_attr` would isolate it and trigger N+1. Prefetch( "channelstream_set", queryset=ChannelStream.objects.select_related( "stream__m3u_account" ).order_by("order"), ), ) channel_group = self.request.query_params.get("channel_group") if channel_group: group_names = channel_group.split(",") qs = qs.filter(channel_group__name__in=group_names) filters = {} q_filters = Q() channel_profile_id = self.request.query_params.get("channel_profile_id") show_disabled_param = self.request.query_params.get("show_disabled", None) only_streamless = self.request.query_params.get("only_streamless", None) only_stale = self.request.query_params.get("only_stale", None) only_has_overrides = self.request.query_params.get("only_has_overrides", None) visibility_filter = self.request.query_params.get("visibility_filter", "active") if channel_profile_id: try: profile_id_int = int(channel_profile_id) if show_disabled_param is None: # Show only enabled channels: channels that have a membership # record for this profile with enabled=True # Default is DISABLED (channels without membership are hidden) filters["channelprofilemembership__channel_profile_id"] = profile_id_int filters["channelprofilemembership__enabled"] = True # If show_disabled is True, show all channels (no filtering needed) except (ValueError, TypeError): # Ignore invalid profile id values pass if only_streamless: q_filters &= Q(streams__isnull=True) if only_stale: # Filter channels that have at least one related stream marked as stale q_filters &= Q(streams__is_stale=True) if only_has_overrides: q_filters &= Q(override__isnull=False) # Visibility filter applies to list-style reads only; retrieve / # update / delete must still reach a hidden channel by id so the # frontend can unhide. Summary powers the TV Guide and follows # the same hidden semantic as downstream clients. if self.action in ("list", "get_ids", "summary"): if visibility_filter == "hidden": q_filters &= Q(hidden_from_output=True) elif visibility_filter != "all": q_filters &= Q(hidden_from_output=False) if self.request.user.user_level < 10: filters["user_level__lte"] = self.request.user.user_level # Hide adult content if user preference is set custom_props = self.request.user.custom_properties or {} if custom_props.get('hide_adult_content', False): filters["is_adult"] = False if filters: qs = qs.filter(**filters) if q_filters: qs = qs.filter(q_filters) # DISTINCT is only needed when a filter joins to a one-to-many table # and can produce duplicate channel rows. channel_profile_id joins # channelprofilemembership; only_stale joins streams. All other # filters use FK or one-to-one joins that cannot produce duplicates. if channel_profile_id or only_stale: return qs.distinct() return qs def get_serializer_context(self): context = super().get_serializer_context() include_streams = ( self.request.query_params.get("include_streams", "false") == "true" ) context["include_streams"] = include_streams # source_stream is only needed by the channel edit form. For get_ids # and summary the channelstream_set prefetch is skipped entirely, so # source_stream cannot be computed without hitting the DB per channel. # For list and write/retrieve paths the prefetch is present, so we # can populate it from memory without extra queries. context["include_source_stream"] = action not in ( "get_ids", "summary" ) if (action := getattr(self, "action", None)) else False return context @extend_schema( methods=["PATCH"], description=( "Bulk edit multiple channels in a single request. " "Accepts a JSON array of channel update objects. Each object must include `id` (the channel's primary key). " "All other fields are optional and support partial updates. " "The `streams` field accepts a list of stream IDs and will replace the channel's current stream assignments. " "All updates are validated before any changes are applied and executed in a single database transaction." ), request=inline_serializer( name="ChannelBulkEditRequest", fields={ "id": serializers.IntegerField(help_text="ID of the channel to update (required)."), "name": serializers.CharField(required=False), "channel_number": serializers.FloatField(required=False), "channel_group_id": serializers.IntegerField(required=False, allow_null=True), "streams": serializers.ListField( child=serializers.IntegerField(), required=False, help_text="List of stream IDs to assign to this channel (replaces existing assignments).", ), "stream_profile_id": serializers.IntegerField(required=False, allow_null=True), "logo_id": serializers.IntegerField(required=False, allow_null=True), "tvg_id": serializers.CharField(required=False, allow_blank=True), "tvc_guide_stationid": serializers.CharField(required=False, allow_blank=True), "epg_data_id": serializers.IntegerField(required=False, allow_null=True), "user_level": serializers.IntegerField(required=False), "is_adult": serializers.BooleanField(required=False), }, many=True, ), responses={ 200: inline_serializer( name="ChannelBulkEditResponse", fields={ "message": serializers.CharField(), "channels": ChannelSerializer(many=True), }, ), 400: inline_serializer( name="ChannelBulkEditErrorResponse", fields={ "errors": serializers.ListField(child=serializers.DictField()), }, ), }, ) @action(detail=False, methods=["patch"], url_path="edit/bulk") def edit_bulk(self, request): """ Bulk edit channels efficiently. Validates all updates first, then applies in a single transaction. """ data = request.data if not isinstance(data, list): return Response( {"error": "Expected a list of channel updates"}, status=status.HTTP_400_BAD_REQUEST, ) # Extract IDs and validate presence channel_updates = {} missing_ids = [] for i, channel_data in enumerate(data): channel_id = channel_data.get("id") if not channel_id: missing_ids.append(f"Item {i}: Channel ID is required") else: channel_updates[channel_id] = channel_data if missing_ids: return Response( {"errors": missing_ids}, status=status.HTTP_400_BAD_REQUEST, ) # Fetch all channels at once (one query) channels_dict = { c.id: c for c in Channel.objects.filter(id__in=channel_updates.keys()) } # Validate and prepare updates validated_updates = [] errors = [] for channel_id, channel_data in channel_updates.items(): channel = channels_dict.get(channel_id) if not channel: errors.append({ "channel_id": channel_id, "error": "Channel not found" }) continue # Handle channel_group_id conversion if 'channel_group_id' in channel_data: group_id = channel_data['channel_group_id'] if group_id is not None: try: channel_data['channel_group_id'] = int(group_id) except (ValueError, TypeError): channel_data['channel_group_id'] = None # Validate with serializer serializer = ChannelSerializer( channel, data=channel_data, partial=True ) if serializer.is_valid(): validated_updates.append((channel, serializer.validated_data)) else: errors.append({ "channel_id": channel_id, "errors": serializer.errors }) if errors: return Response( {"errors": errors, "updated_count": len(validated_updates)}, status=status.HTTP_400_BAD_REQUEST, ) # Capture override intent from the raw payload: presence of the # key distinguishes "no change" from explicit null ("clear"), # which `validated_data` would collapse. override_intents = {} for channel_data in data: if "override" in channel_data: cid = channel_data.get("id") override_intents[cid] = channel_data["override"] # Capture hide / unhide transitions before setattr overwrites # the in-memory channels. bulk_update skips post_save, so the # compact-mode assign / release runs explicitly below. unhide_transition_ids = [ channel.id for channel, validated_data in validated_updates if validated_data.get("hidden_from_output") is False and channel.hidden_from_output is True ] hide_transition_candidates = [ channel for channel, validated_data in validated_updates if validated_data.get("hidden_from_output") is True and channel.hidden_from_output is False and channel.channel_number is not None and channel.auto_created and channel.auto_created_by_id ] # Apply all updates in a transaction with transaction.atomic(): streams_updates = [] for channel, validated_data in validated_updates: # Streams (M2M) and override (reverse OneToOne) cannot # ride the setattr loop; handle each separately below. streams = validated_data.pop("streams", None) if streams is not None: streams_updates.append((channel, streams)) validated_data.pop("override", None) for key, value in validated_data.items(): setattr(channel, key, value) # Single bulk_update query instead of individual saves channels_to_update = [channel for channel, _ in validated_updates] if channels_to_update: # Collect all unique field names from all updates (streams already popped) all_fields = set() for _, validated_data in validated_updates: all_fields.update(validated_data.keys()) # Only call bulk_update if there are non-M2M fields to update if all_fields: Channel.objects.bulk_update( channels_to_update, fields=list(all_fields), batch_size=100 ) # On unhide under compact mode, assign a number so the # channel is immediately addressable by clients. if unhide_transition_ids: from .compact_numbering import ( assign_compact_numbers_for_channels, ) assign_compact_numbers_for_channels(unhide_transition_ids) # On hide under compact mode, release the channel_number # so the slot is reused. The bulk path bypasses the # post_save signal that handles single-row hides. if hide_transition_candidates: from .compact_numbering import ( get_group_relation_for_channel, is_compact_group, ) ids_to_release = [] for ch in hide_transition_candidates: relation = get_group_relation_for_channel(ch) if relation and is_compact_group(relation): ids_to_release.append(ch.id) if ids_to_release: Channel.objects.filter(id__in=ids_to_release).update( channel_number=None ) # Refresh in-memory copies so the response shape # reflects the cleared numbers. for ch in channels_to_update: if ch.id in ids_to_release: ch.channel_number = None # Override (reverse OneToOne) needs a separate write path. if override_intents: from apps.channels.models import ChannelOverride from apps.channels.managers import OVERRIDABLE_FIELDS override_fields = OVERRIDABLE_FIELDS # Block override mutations on manual channels. Mixed # selections with override:null are tolerated because # clearing a non-existent row is a no-op. manual_with_override = [] for channel, _ in validated_updates: if channel.id not in override_intents: continue intent = override_intents[channel.id] if ( intent is not None and intent != {} and not channel.auto_created ): manual_with_override.append(channel.id) if manual_with_override: return Response( { "errors": [ { "channel_id": cid, "error": ( "Cannot set override on a manual channel; " "overrides only apply to auto-created channels." ), } for cid in manual_with_override ] }, status=status.HTTP_400_BAD_REQUEST, ) channels_to_clear = [] overrides_to_upsert = [] for channel, _ in validated_updates: if channel.id not in override_intents: continue intent = override_intents[channel.id] if intent is None: channels_to_clear.append(channel.id) elif intent == {}: # Empty dict means no override intent; treat as no-op. continue else: defaults = { f: intent.get(f) for f in override_fields if f in intent } # Coerce FK aliases (logo, channel_group, ...) to # the *_id columns ChannelOverride actually stores. for raw, mapped in ( ("logo", "logo_id"), ("channel_group", "channel_group_id"), ("epg_data", "epg_data_id"), ("stream_profile", "stream_profile_id"), ): if raw in intent and mapped not in defaults: val = intent[raw] defaults[mapped] = ( val.id if hasattr(val, "id") else val ) overrides_to_upsert.append((channel.id, defaults)) if channels_to_clear: ChannelOverride.objects.filter( channel_id__in=channels_to_clear ).delete() # Bulk upsert keeps a 1000-channel batch to two # statements (one INSERT, one UPDATE) instead of the # per-row SELECT + INSERT-or-UPDATE that update_or_create # would generate. if overrides_to_upsert: existing_overrides = { o.channel_id: o for o in ChannelOverride.objects.filter( channel_id__in=[ cid for cid, _ in overrides_to_upsert ] ) } to_create = [] to_update = [] update_field_set = set() for channel_id, defaults in overrides_to_upsert: existing = existing_overrides.get(channel_id) if existing: for f, v in defaults.items(): setattr(existing, f, v) update_field_set.add(f) to_update.append(existing) else: to_create.append( ChannelOverride( channel_id=channel_id, **defaults ) ) if to_update: ChannelOverride.objects.bulk_update( to_update, fields=list(update_field_set), batch_size=200, ) if to_create: ChannelOverride.objects.bulk_create( to_create, batch_size=200 ) # Drop override rows that ended up all-null; an empty # override would falsely surface as active in the UI. touched_ids = [cid for cid, _ in overrides_to_upsert] empty_overrides = [ o for o in ChannelOverride.objects.filter( channel_id__in=touched_ids ) if not o.has_any_override() ] if empty_overrides: ChannelOverride.objects.filter( id__in=[o.id for o in empty_overrides] ).delete() # Queryset writes leave the reverse-OneToOne cache stale; # clear it so the serializer reads the new override state. touched_channel_ids = {cid for cid, _ in overrides_to_upsert} touched_channel_ids.update(channels_to_clear) if touched_channel_ids: for channel, _ in validated_updates: if channel.id not in touched_channel_ids: continue try: channel._state.fields_cache.pop("override", None) except AttributeError: pass if hasattr(channel, "_channel_override_cache"): delattr(channel, "_channel_override_cache") # Handle streams M2M updates separately for channel, streams in streams_updates: normalized_ids = [ stream.id if hasattr(stream, "id") else stream for stream in streams ] current_links = { cs.stream_id: cs for cs in channel.channelstream_set.all() } existing_ids = set(current_links.keys()) new_ids = set(normalized_ids) to_remove = existing_ids - new_ids if to_remove: channel.channelstream_set.filter(stream_id__in=to_remove).delete() to_update = [] for order, stream_id in enumerate(normalized_ids): if stream_id in current_links: cs = current_links[stream_id] if cs.order != order: cs.order = order to_update.append(cs) else: ChannelStream.objects.create( channel=channel, stream_id=stream_id, order=order ) if to_update: ChannelStream.objects.bulk_update(to_update, ["order"]) # Return the updated objects (already in memory) serialized_channels = ChannelSerializer( [channel for channel, _ in validated_updates], many=True, context=self.get_serializer_context() ).data return Response({ "message": f"Successfully updated {len(validated_updates)} channels", "channels": serialized_channels }) @extend_schema( methods=["POST"], description=( "Bulk rename channel names using a regex find/replace executed server-side. " "Accepts JavaScript-style named groups (e.g., (?...)) and converts them to Python syntax. " "Supports flags: 'i' (IGNORECASE). Replacement tokens like $1, $& and $ are translated to Python." ), request=inline_serializer( name="BulkRegexRenameRequest", fields={ "channel_ids": serializers.ListField(child=serializers.IntegerField()), "find": serializers.CharField(), "replace": serializers.CharField(required=False, allow_blank=True), "flags": serializers.CharField(required=False, allow_blank=True), }, ), ) @action(detail=False, methods=["post"], url_path="edit/bulk-regex") def bulk_regex_rename(self, request): """ Efficiently apply a regex find/replace to the `name` field of multiple channels. """ import regex as re channel_ids = request.data.get("channel_ids", []) pattern = request.data.get("find", "") replace = request.data.get("replace", "") flags_str = request.data.get("flags", "") or "" if not isinstance(channel_ids, list) or len(channel_ids) == 0: return Response({"error": "channel_ids must be a non-empty list"}, status=status.HTTP_400_BAD_REQUEST) if not isinstance(pattern, str) or pattern.strip() == "": return Response({"error": "find (regex pattern) is required"}, status=status.HTTP_400_BAD_REQUEST) if not isinstance(replace, str): return Response({"error": "replace must be a string"}, status=status.HTTP_400_BAD_REQUEST) # Convert JS-style named groups to Python (?...) -> (?P...) try: converted_pattern = re.sub(r"\(\?<([^>]+)>", r"(?P<\1>", pattern) except Exception as e: return Response({"error": f"Failed to normalize pattern: {e}"}, status=status.HTTP_400_BAD_REQUEST) # Compile flags re_flags = 0 if "i" in flags_str: re_flags |= re.IGNORECASE # Note: 'g' (global) is the default behavior of re.sub; no action needed. # Translate common JS replacement tokens to Python def translate_js_replacement(rep: str) -> str: # $$ -> $ rep = rep.replace("$$", "$") # $& -> \g<0> rep = rep.replace("$&", r"\g<0>") # $ -> \g rep = re.sub(r"\$<([A-Za-z_][A-Za-z0-9_]*)>", r"\\g<\1>", rep) # $1 -> \g<1>, $2 -> \g<2>, etc. rep = re.sub(r"\$(\d+)", r"\\g<\1>", rep) return rep try: replacement_py = translate_js_replacement(replace) compiled = re.compile(converted_pattern, flags=re_flags) except Exception as e: return Response({"error": f"Invalid regex pattern: {e}"}, status=status.HTTP_400_BAD_REQUEST) # Fetch channels in one query channels = list(Channel.objects.filter(id__in=channel_ids)) if not channels: return Response({"error": "No matching channels found for provided IDs"}, status=status.HTTP_404_NOT_FOUND) changed = [] for ch in channels: current = ch.name or "" try: new_name = compiled.sub(replacement_py, current) except Exception as e: # Skip problematic replacements but continue processing others logger.warning(f"Regex replacement failed for channel {ch.id}: {e}") continue # Only update if name actually changes and remains non-empty if new_name != current and new_name.strip(): ch.name = new_name changed.append(ch) # Apply updates in bulk updated_count = 0 if changed: with transaction.atomic(): Channel.objects.bulk_update(changed, fields=["name"], batch_size=100) updated_count = len(changed) return Response({ "success": True, "updated_count": updated_count, }, status=status.HTTP_200_OK) @action(detail=False, methods=["post"], url_path="set-names-from-epg") def set_names_from_epg(self, request): """ Trigger a Celery task to set channel names from EPG data """ from .tasks import set_channels_names_from_epg data = request.data channel_ids = data.get("channel_ids", []) if not channel_ids: return Response( {"error": "channel_ids is required"}, status=status.HTTP_400_BAD_REQUEST, ) if not isinstance(channel_ids, list): return Response( {"error": "channel_ids must be a list"}, status=status.HTTP_400_BAD_REQUEST, ) # Start the Celery task task = set_channels_names_from_epg.delay(channel_ids) return Response({ "message": f"Started EPG name setting task for {len(channel_ids)} channels", "task_id": task.id, "channel_count": len(channel_ids) }) @action(detail=False, methods=["post"], url_path="set-logos-from-epg") def set_logos_from_epg(self, request): """ Trigger a Celery task to set channel logos from EPG data. Provide channel_ids or epg_source_id (not both). """ from .tasks import set_channels_logos_from_epg data = request.data channel_ids = data.get("channel_ids") epg_source_id = data.get("epg_source_id") if channel_ids and epg_source_id: return Response( {"error": "Provide either channel_ids or epg_source_id, not both"}, status=status.HTTP_400_BAD_REQUEST, ) if not channel_ids and not epg_source_id: return Response( {"error": "channel_ids or epg_source_id is required"}, status=status.HTTP_400_BAD_REQUEST, ) if channel_ids is not None: if not isinstance(channel_ids, list): return Response( {"error": "channel_ids must be a list"}, status=status.HTTP_400_BAD_REQUEST, ) if not channel_ids: return Response( {"error": "channel_ids cannot be empty"}, status=status.HTTP_400_BAD_REQUEST, ) task = set_channels_logos_from_epg.delay(channel_ids=channel_ids) channel_count = len(channel_ids) else: from .utils import channels_with_epg_icon_queryset task = set_channels_logos_from_epg.delay(epg_source_id=epg_source_id) channel_count = channels_with_epg_icon_queryset( epg_source_id=epg_source_id, ).count() return Response({ "message": f"Started EPG logo setting task for {channel_count} channels", "task_id": task.id, "channel_count": channel_count, }) @action(detail=False, methods=["post"], url_path="set-tvg-ids-from-epg") def set_tvg_ids_from_epg(self, request): """ Trigger a Celery task to set channel TVG-IDs from EPG data """ from .tasks import set_channels_tvg_ids_from_epg data = request.data channel_ids = data.get("channel_ids", []) if not channel_ids: return Response( {"error": "channel_ids is required"}, status=status.HTTP_400_BAD_REQUEST, ) if not isinstance(channel_ids, list): return Response( {"error": "channel_ids must be a list"}, status=status.HTTP_400_BAD_REQUEST, ) # Start the Celery task task = set_channels_tvg_ids_from_epg.delay(channel_ids) return Response({ "message": f"Started EPG TVG-ID setting task for {len(channel_ids)} channels", "task_id": task.id, "channel_count": len(channel_ids) }) @action(detail=False, methods=["get"], url_path="ids") def get_ids(self, request, *args, **kwargs): # Get the filtered queryset queryset = self.get_queryset() # Apply filtering, search, and ordering queryset = self.filter_queryset(queryset) # Return only the IDs from the queryset channel_ids = queryset.values_list("id", flat=True) # JsonResponse skips DRF's renderer pipeline for a flat int list. return JsonResponse(list(channel_ids), safe=False) @action(detail=False, methods=["get"], url_path="summary") def summary(self, request, *args, **kwargs): """Return a lightweight list of channels with only the fields needed by the TV Guide. The TV Guide is a downstream output surface like HDHR / M3U / EPG / XC and must reflect the user's overrides. Effective values are coalesced at the SQL layer; the annotated columns are renamed back to the raw field names on the way out so the response shape stays unchanged for the frontend. """ from .managers import with_effective_values queryset = with_effective_values( self.filter_queryset(self.get_queryset()) ) return JsonResponse( [ { "id": row["id"], "uuid": row["uuid"], "name": row["effective_name"], "logo_id": row["effective_logo_id"], "channel_number": row["effective_channel_number"], "epg_data_id": row["effective_epg_data_id"], "channel_group_id": row["effective_channel_group_id"], } for row in queryset.values( "id", "uuid", "effective_name", "effective_logo_id", "effective_channel_number", "effective_epg_data_id", "effective_channel_group_id", ) ], safe=False, ) @extend_schema( parameters=[ OpenApiParameter( name="start", type=OpenApiTypes.NUMBER, location=OpenApiParameter.QUERY, required=True, description="Inclusive lower bound of the range to scan.", ), OpenApiParameter( name="end", type=OpenApiTypes.NUMBER, location=OpenApiParameter.QUERY, required=False, description=( "Inclusive upper bound. If omitted or equal to start, " "behaves as a single-number lookup." ), ), ], responses={ 200: inline_serializer( name="ChannelsInRangeResponse", fields={ "occupants": serializers.ListField( child=serializers.DictField() ) }, ) }, description=( "Returns the channels (including those whose effective number is " "set via override) currently occupying numbers within the given " "range. Used by the group settings form to surface inline range " "conflict warnings. Capped at 50 entries to bound the response " "payload; the frontend only needs to know whether any conflicts " "exist after filtering, not the entire list." ), ) @action(detail=False, methods=["get"], url_path="numbers-in-range") def numbers_in_range(self, request, *args, **kwargs): from .managers import with_effective_values raw_start = request.query_params.get("start") raw_end = request.query_params.get("end") if raw_start is None or raw_start == "": return Response({"occupants": []}) try: start = float(raw_start) except (TypeError, ValueError): return Response( {"detail": "Invalid start value"}, status=status.HTTP_400_BAD_REQUEST, ) try: end = float(raw_end) if raw_end not in (None, "") else start except (TypeError, ValueError): return Response( {"detail": "Invalid end value"}, status=status.HTTP_400_BAD_REQUEST, ) if end < start: start, end = end, start queryset = ( with_effective_values( Channel.objects.all(), select_related_fks=True ) .filter( effective_channel_number__gte=start, effective_channel_number__lte=end, ) .order_by("effective_channel_number")[:50] ) occupants = [] for occupant in queryset: effective_group = getattr( occupant, "effective_channel_group_obj", None ) group_name = ( getattr(effective_group, "name", None) if effective_group is not None else None ) effective_group_id = getattr( occupant, "effective_channel_group_id", None ) override = getattr(occupant, "override", None) override_sets_number = bool( override is not None and override.channel_number is not None ) occupants.append( { "id": occupant.id, "name": getattr( occupant, "effective_name", occupant.name ), "channel_number": getattr( occupant, "effective_channel_number", occupant.channel_number, ), "channel_group": group_name, "channel_group_id": effective_group_id, "auto_created": bool(occupant.auto_created), "auto_created_by_account_id": ( occupant.auto_created_by_id if occupant.auto_created_by_id else None ), "has_channel_number_override": override_sets_number, } ) return Response({"occupants": occupants}) @extend_schema( methods=["POST"], description="Retrieve channels by a list of UUIDs using POST to avoid URL length limitations", request=inline_serializer( name="ChannelByUUIDsRequest", fields={ "uuids": serializers.ListField( child=serializers.CharField(), help_text="List of channel UUIDs to retrieve", ) }, ), responses={200: ChannelSerializer(many=True)}, ) @action(detail=False, methods=["post"], url_path="by-uuids") def get_by_uuids(self, request, *args, **kwargs): uuids = request.data.get("uuids", []) if not isinstance(uuids, list): return Response( {"error": "uuids must be a list of strings"}, status=status.HTTP_400_BAD_REQUEST, ) channels = Channel.objects.filter(uuid__in=uuids) serializer = self.get_serializer(channels, many=True) return Response(serializer.data) @extend_schema( methods=["POST"], description="Auto-assign channel_number in bulk by an ordered list of channel IDs.", request=inline_serializer( name="AssignChannelsRequest", fields={ "starting_number": serializers.FloatField( help_text="Starting channel number to assign (can be decimal)", required=False, ), "channel_ids": serializers.ListField( child=serializers.IntegerField(), help_text="Channel IDs to assign", ), }, ), ) @action(detail=False, methods=["post"], url_path="assign") def assign(self, request): with transaction.atomic(): channel_ids = request.data.get("channel_ids", []) # Ensure starting_number is processed as a float try: channel_num = float(request.data.get("starting_number", 1)) except (ValueError, TypeError): channel_num = 1.0 for channel_id in channel_ids: Channel.objects.filter(id=channel_id).update(channel_number=channel_num) channel_num = channel_num + 1 return Response( {"message": "Channels have been auto-assigned!"}, status=status.HTTP_200_OK ) @extend_schema( methods=["POST"], description=( "Create a new channel from an existing stream. " "If 'channel_number' is provided, it will be used (if available); " "otherwise, the next available channel number is assigned. " "If 'channel_profile_ids' is provided, the channel will only be added to those profiles. " "Accepts either a single ID or an array of IDs." ), request=inline_serializer( name="FromStreamRequest", fields={ "stream_id": serializers.IntegerField(help_text="ID of the stream to link"), "channel_number": serializers.FloatField( help_text="(Optional) Desired channel number. Must not be in use.", required=False, ), "name": serializers.CharField(help_text="Desired channel name", required=False), "channel_profile_ids": serializers.ListField( child=serializers.IntegerField(), help_text="(Optional) Channel profile ID(s). Behavior: omitted = add to ALL profiles (default); empty array [] = add to NO profiles; [0] = add to ALL profiles (explicit); [1,2,...] = add only to specified profiles.", required=False, ), }, ), responses={201: ChannelSerializer()}, ) @action(detail=False, methods=["post"], url_path="from-stream") def from_stream(self, request): stream_id = request.data.get("stream_id") if not stream_id: return Response( {"error": "Missing stream_id"}, status=status.HTTP_400_BAD_REQUEST ) stream = get_object_or_404(Stream, pk=stream_id) channel_group = stream.channel_group name = request.data.get("name") if name is None: name = stream.name # Check if client provided a channel_number; if not, use stream_chno or auto-assign channel_number = request.data.get("channel_number") if channel_number is None: # Channel number not provided by client, check stream's channel number or auto-assign if stream.stream_chno is not None: channel_number = stream.stream_chno elif channel_number == 0: # Special case: 0 means ignore provider numbers and auto-assign channel_number = None elif channel_number == -1: # Special case: -1 means assign the number after the current highest highest = Channel.objects.order_by('-channel_number').values_list('channel_number', flat=True).first() channel_number = (int(highest) + 1) if highest is not None else 1 if channel_number is None: # Still None, auto-assign the next available channel number channel_number = Channel.get_next_available_channel_number() try: channel_number = float(channel_number) except ValueError: return Response( {"error": "channel_number must be an integer."}, status=status.HTTP_400_BAD_REQUEST, ) # If the provided number is already used, return an error. if Channel.objects.filter(channel_number=channel_number).exists(): channel_number = Channel.get_next_available_channel_number(channel_number) # Get the tvc_guide_stationid from custom properties if it exists stream_custom_props = stream.custom_properties or {} tvc_guide_stationid = stream_custom_props.get("tvc-guide-stationid") channel_data = { "channel_number": channel_number, "name": name, "tvg_id": stream.tvg_id, "tvc_guide_stationid": tvc_guide_stationid, "streams": [stream_id], "is_adult": stream.is_adult, } # Only add channel_group_id if the stream has a channel group if channel_group: channel_data["channel_group_id"] = channel_group.id if stream.logo_url: # Import validation function from apps.channels.tasks import validate_logo_url validated_logo_url = validate_logo_url(stream.logo_url) if validated_logo_url: logo, _ = Logo.objects.get_or_create( url=validated_logo_url, defaults={"name": stream.name or stream.tvg_id} ) channel_data["logo_id"] = logo.id # Attempt to find existing EPGs with the same tvg-id epgs = EPGData.objects.filter(tvg_id=stream.tvg_id) if epgs: channel_data["epg_data_id"] = epgs.first().id serializer = self.get_serializer(data=channel_data) serializer.is_valid(raise_exception=True) with transaction.atomic(): channel = serializer.save() channel.streams.add(stream) # Handle channel profile membership # Semantics: # - Omitted (None): add to ALL profiles (backward compatible default) # - Empty array []: add to NO profiles # - Sentinel [0] or 0: add to ALL profiles (explicit) # - [1,2,...]: add to specified profile IDs only channel_profile_ids = request.data.get("channel_profile_ids") if channel_profile_ids is not None: # Normalize single ID to array if not isinstance(channel_profile_ids, list): channel_profile_ids = [channel_profile_ids] # Determine action based on semantics if channel_profile_ids is None: # Omitted -> add to all profiles (backward compatible) profiles = ChannelProfile.objects.all() ChannelProfileMembership.objects.bulk_create([ ChannelProfileMembership(channel_profile=profile, channel=channel, enabled=True) for profile in profiles ]) elif isinstance(channel_profile_ids, list) and len(channel_profile_ids) == 0: # Empty array -> add to no profiles pass elif isinstance(channel_profile_ids, list) and 0 in channel_profile_ids: # Sentinel 0 -> add to all profiles (explicit) profiles = ChannelProfile.objects.all() ChannelProfileMembership.objects.bulk_create([ ChannelProfileMembership(channel_profile=profile, channel=channel, enabled=True) for profile in profiles ]) else: # Specific profile IDs try: channel_profiles = ChannelProfile.objects.filter(id__in=channel_profile_ids) if len(channel_profiles) != len(channel_profile_ids): missing_ids = set(channel_profile_ids) - set(channel_profiles.values_list('id', flat=True)) return Response( {"error": f"Channel profiles with IDs {list(missing_ids)} not found"}, status=status.HTTP_400_BAD_REQUEST, ) ChannelProfileMembership.objects.bulk_create([ ChannelProfileMembership( channel_profile=profile, channel=channel, enabled=True ) for profile in channel_profiles ]) except Exception as e: return Response( {"error": f"Error creating profile memberships: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST, ) # Send WebSocket notification for single channel creation from core.utils import send_websocket_update send_websocket_update('updates', 'update', { 'type': 'channels_created', 'count': 1, 'channel_id': channel.id, 'channel_name': channel.name, 'channel_number': channel.channel_number }) return Response(serializer.data, status=status.HTTP_201_CREATED) @extend_schema( methods=["POST"], description=( "Asynchronously bulk create channels from stream IDs. " "Returns a task ID to track progress via WebSocket. " "This is the recommended approach for large bulk operations." ), request=inline_serializer( name="FromStreamBulkRequest", fields={ "stream_ids": serializers.ListField( child=serializers.IntegerField(), help_text="List of stream IDs to create channels from" ), "channel_profile_ids": serializers.ListField( child=serializers.IntegerField(), help_text="(Optional) Channel profile ID(s). Behavior: omitted = add to ALL profiles (default); empty array [] = add to NO profiles; [0] = add to ALL profiles (explicit); [1,2,...] = add only to specified profiles.", required=False, ), "starting_channel_number": serializers.IntegerField( help_text="(Optional) Starting channel number mode: null=use provider numbers, 0=lowest available, other=start from specified number", required=False, ), }, ), ) @action(detail=False, methods=["post"], url_path="from-stream/bulk") def from_stream_bulk(self, request): from .tasks import bulk_create_channels_from_streams stream_ids = request.data.get("stream_ids", []) channel_profile_ids = request.data.get("channel_profile_ids") starting_channel_number = request.data.get("starting_channel_number") if not stream_ids: return Response( {"error": "stream_ids is required and cannot be empty"}, status=status.HTTP_400_BAD_REQUEST, ) if not isinstance(stream_ids, list): return Response( {"error": "stream_ids must be a list of integers"}, status=status.HTTP_400_BAD_REQUEST, ) # Normalize channel_profile_ids to array if single ID provided if channel_profile_ids is not None: if not isinstance(channel_profile_ids, list): channel_profile_ids = [channel_profile_ids] # Start the async task task = bulk_create_channels_from_streams.delay(stream_ids, channel_profile_ids, starting_channel_number) return Response({ "task_id": task.id, "message": f"Bulk channel creation task started for {len(stream_ids)} streams", "stream_count": len(stream_ids), "status": "started" }, status=status.HTTP_202_ACCEPTED) # ───────────────────────────────────────────────────────── # 6) EPG Fuzzy Matching # ───────────────────────────────────────────────────────── @extend_schema( methods=["POST"], description="Kick off a Celery task that tries to fuzzy-match channels with EPG data. If channel_ids are provided, only those channels will be processed.", request=inline_serializer( name="MatchEpgRequest", fields={ 'channel_ids': serializers.ListField( child=serializers.IntegerField(), help_text='List of channel IDs to process (includes channels that already have EPG). If empty or not provided, only channels without EPG are processed.', required=False, ) } ), ) @action(detail=False, methods=["post"], url_path="match-epg") def match_epg(self, request): # Get channel IDs from request body if provided channel_ids = request.data.get('channel_ids', []) if channel_ids: # Process only selected channels from .tasks import match_selected_channels_epg match_selected_channels_epg.delay(channel_ids) message = f"EPG matching task initiated for {len(channel_ids)} selected channel(s)." else: # Process all channels without EPG (original behavior) match_epg_channels.delay() message = "EPG matching task initiated for all channels without EPG." return Response( {"message": message}, status=status.HTTP_202_ACCEPTED ) @extend_schema( methods=["POST"], description="Try to auto-match this specific channel with EPG data.", ) @action(detail=True, methods=["post"], url_path="match-epg") def match_channel_epg(self, request, pk=None): channel = self.get_object() match_single_channel_epg.delay(channel.id) return Response( { "message": f"EPG matching started for channel '{channel.name}'", "accepted": True, "channel_id": channel.id, }, status=status.HTTP_202_ACCEPTED, ) # ───────────────────────────────────────────────────────── # 7) Set EPG and Refresh # ───────────────────────────────────────────────────────── @extend_schema( methods=["POST"], description="Set EPG data for a channel and refresh program data", request=inline_serializer( name="SetEpgRequest", fields={ "epg_data_id": serializers.IntegerField(help_text="EPG data ID to link") }, ), responses={200: "EPG data linked and refresh triggered"}, ) @action(detail=True, methods=["post"], url_path="set-epg") def set_epg(self, request, pk=None): channel = self.get_object() epg_data_id = request.data.get("epg_data_id") # Handle removing EPG link if epg_data_id in (None, "", "0", 0): channel.epg_data = None channel.save(update_fields=["epg_data"]) return Response( {"message": f"EPG data removed from channel {channel.name}"} ) try: # Get the EPG data object from apps.epg.models import EPGData epg_data = EPGData.objects.get(pk=epg_data_id) # Set the EPG data and save. refresh_epg_programs (post_save) queues # parse_programs_for_tvg_id for non-dummy sources — no second dispatch here. channel.epg_data = epg_data channel.save(update_fields=["epg_data"]) status_message = None if epg_data.epg_source.source_type != 'dummy': status_message = "EPG refresh queued" # Build response message message = f"EPG data set to {epg_data.tvg_id} for channel {channel.name}" if status_message: message += f". {status_message}" return Response( { "message": message, "channel": self.get_serializer(channel).data, "task_status": status_message, } ) except Exception as e: return Response({"error": str(e)}, status=400) @extend_schema( description=( "Reorder a channel by moving it after another channel (or to the start if insert_after_id is null). " "The channel will receive the next whole number after the target channel, and all subsequent " "channels will be renumbered accordingly." ), request=inline_serializer( name="ReorderChannelRequest", fields={ "insert_after_id": serializers.IntegerField( help_text="ID of the channel to insert after. Use null to move to the beginning.", required=False, allow_null=True, ), }, ), ) @action(detail=True, methods=["post"], url_path="reorder") def reorder(self, request, pk=None): """ Reorder a channel by moving it after another channel (or to the start if insert_after_id is null). Shifts other channels as needed to maintain contiguous ordering. """ channel = self.get_object() insert_after_id = request.data.get("insert_after_id") old_channel_number = channel.channel_number with transaction.atomic(): if insert_after_id is None: # Move to the beginning (channel_number = 1) target_number = 0 desired_number = 1 else: try: target_channel = Channel.objects.get(id=insert_after_id) target_number = target_channel.channel_number or 0 desired_number = int(target_number) + 1 except Channel.DoesNotExist: return Response( {"error": "Target channel not found"}, status=status.HTTP_404_NOT_FOUND, ) if desired_number == old_channel_number: # No change needed return Response( { "message": f"Channel {channel.name} already at position {desired_number}", "channel": self.get_serializer(channel).data, }, status=status.HTTP_200_OK, ) if desired_number < old_channel_number: # Moving up: increment all channels between desired_number and old_channel_number-1 Channel.objects.filter( channel_number__gte=desired_number, channel_number__lt=old_channel_number ).update(channel_number=F('channel_number') + 1) channel.channel_number = desired_number channel.save(update_fields=['channel_number']) elif desired_number > old_channel_number: # Moving down: shift down channels between old+1 and desired-1, then set to desired-1 if desired_number > old_channel_number + 1: Channel.objects.filter( channel_number__gt=old_channel_number, channel_number__lt=desired_number ).update(channel_number=F('channel_number') - 1) channel.channel_number = desired_number - 1 channel.save(update_fields=['channel_number']) else: # No move or same position channel.channel_number = desired_number channel.save(update_fields=['channel_number']) return Response( { "message": f"Channel {channel.name} moved to position {desired_number}", "channel": self.get_serializer(channel).data, }, status=status.HTTP_200_OK, ) @extend_schema( methods=["POST"], description="Associate multiple channels with EPG data without triggering a full refresh", request=inline_serializer( name="BatchSetEpgRequest", fields={ "associations": serializers.ListField( child=inline_serializer( name="EpgAssociation", fields={ "channel_id": serializers.IntegerField(), "epg_data_id": serializers.IntegerField( required=False, allow_null=True, help_text="EPG data ID to link. Pass null to remove EPG linkage.", ), }, ), ) }, ), ) @action(detail=False, methods=["post"], url_path="batch-set-epg") def batch_set_epg(self, request): """Efficiently associate multiple channels with EPG data at once.""" associations = request.data.get("associations", []) if not associations: return Response( {"error": "associations list is required"}, status=status.HTTP_400_BAD_REQUEST, ) # Extract channel IDs upfront channel_updates = {} for assoc in associations: channel_id = assoc.get("channel_id") epg_data_id = assoc.get("epg_data_id") if not channel_id: continue channel_updates[channel_id] = epg_data_id # Batch fetch all channels (single query) channels_dict = { c.id: c for c in Channel.objects.filter(id__in=channel_updates.keys()) } # Collect channels whose EPG assignment actually changes channels_to_update = [] changed_epg_ids = set() for channel_id, epg_data_id in channel_updates.items(): if channel_id not in channels_dict: logger.error(f"Channel with ID {channel_id} not found") continue channel = channels_dict[channel_id] if channel.epg_data_id == epg_data_id: continue channel.epg_data_id = epg_data_id channels_to_update.append(channel) if epg_data_id: changed_epg_ids.add(epg_data_id) # Bulk update all channels (single query) if channels_to_update: with transaction.atomic(): Channel.objects.bulk_update( channels_to_update, fields=["epg_data_id"], batch_size=100 ) channels_updated = len(channels_to_update) from apps.epg.tasks import dispatch_program_refresh_for_epg_ids programs_refreshed = dispatch_program_refresh_for_epg_ids(changed_epg_ids) return Response( { "success": True, "channels_updated": channels_updated, "programs_refreshed": programs_refreshed, } ) # ───────────────────────────────────────────────────────── # 4) Bulk Delete Streams # ───────────────────────────────────────────────────────── class BulkDeleteStreamsAPIView(APIView): def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] @extend_schema( description="Bulk delete streams by ID", request=inline_serializer( name="BulkDeleteStreamsRequest", fields={ "stream_ids": serializers.ListField( child=serializers.IntegerField(), help_text="Stream IDs to delete", ) }, ), ) def delete(self, request, *args, **kwargs): stream_ids = request.data.get("stream_ids", []) Stream.objects.filter(id__in=stream_ids).delete() return Response( {"message": "Streams deleted successfully!"}, status=status.HTTP_204_NO_CONTENT, ) # ───────────────────────────────────────────────────────── # 5) Bulk Delete Channels # ───────────────────────────────────────────────────────── class BulkDeleteChannelsAPIView(APIView): def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] @extend_schema( description="Bulk delete channels by ID", request=inline_serializer( name="BulkDeleteChannelsRequest", fields={ "channel_ids": serializers.ListField( child=serializers.IntegerField(), help_text="Channel IDs to delete", ) }, ), ) def delete(self, request): channel_ids = request.data.get("channel_ids", []) Channel.objects.filter(id__in=channel_ids).delete() return Response( {"message": "Channels deleted"}, status=status.HTTP_204_NO_CONTENT ) # ───────────────────────────────────────────────────────── # 6) Bulk Delete Logos # ───────────────────────────────────────────────────────── class BulkDeleteLogosAPIView(APIView): def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] @extend_schema( description="Bulk delete logos by ID", request=inline_serializer( name="BulkDeleteLogosRequest", fields={ "logo_ids": serializers.ListField( child=serializers.IntegerField(), help_text="Logo IDs to delete", ), "delete_files": serializers.BooleanField( required=False, default=False, help_text="Whether to also delete local logo files from disk.", ), }, ), ) def delete(self, request): logo_ids = request.data.get("logo_ids", []) delete_files = request.data.get("delete_files", False) # Get logos and their usage info before deletion logos_to_delete = Logo.objects.filter(id__in=logo_ids) total_channels_affected = 0 local_files_deleted = 0 for logo in logos_to_delete: # Handle file deletion for local files if delete_files and logo.url and logo.url.startswith('/data/logos'): try: if os.path.exists(logo.url): os.remove(logo.url) local_files_deleted += 1 logger.info(f"Deleted local logo file: {logo.url}") except Exception as e: logger.error(f"Failed to delete logo file {logo.url}: {str(e)}") return Response( {"error": f"Failed to delete logo file {logo.url}: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) if logo.channels.exists(): channel_count = logo.channels.count() total_channels_affected += channel_count # Remove logo from channels logo.channels.update(logo=None) logger.info(f"Removed logo {logo.name} from {channel_count} channels before deletion") # Delete logos deleted_count = logos_to_delete.delete()[0] message = f"Successfully deleted {deleted_count} logos" if total_channels_affected > 0: message += f" and removed them from {total_channels_affected} channels" if local_files_deleted > 0: message += f" and deleted {local_files_deleted} local files" return Response( {"message": message}, status=status.HTTP_204_NO_CONTENT ) class CleanupUnusedLogosAPIView(APIView): def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] @extend_schema( description="Delete all channel logos that are not used by any channels", request=inline_serializer( name="CleanupUnusedLogosRequest", fields={ "delete_files": serializers.BooleanField( help_text="Whether to delete local logo files from disk", default=False, required=False, ) }, ), ) def post(self, request): """Delete all channel logos with no channel associations""" delete_files = request.data.get("delete_files", False) # Find logos that are not used by any channels unused_logos = Logo.objects.filter(channels__isnull=True) deleted_count = unused_logos.count() logo_names = list(unused_logos.values_list('name', flat=True)) local_files_deleted = 0 # Handle file deletion for local files if requested if delete_files: for logo in unused_logos: if logo.url and logo.url.startswith('/data/logos'): try: if os.path.exists(logo.url): os.remove(logo.url) local_files_deleted += 1 logger.info(f"Deleted local logo file: {logo.url}") except Exception as e: logger.error(f"Failed to delete logo file {logo.url}: {str(e)}") return Response( {"error": f"Failed to delete logo file {logo.url}: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) # Delete the unused logos unused_logos.delete() message = f"Successfully deleted {deleted_count} unused logos" if local_files_deleted > 0: message += f" and deleted {local_files_deleted} local files" return Response({ "message": message, "deleted_count": deleted_count, "deleted_logos": logo_names, "local_files_deleted": local_files_deleted }) class LogoPagination(PageNumberPagination): page_size = 50 # Default page size to match frontend default page_size_query_param = "page_size" # Allow clients to specify page size max_page_size = 1000 # Prevent excessive page sizes def paginate_queryset(self, queryset, request, view=None): # Check if pagination should be disabled for specific requests if request.query_params.get('no_pagination') == 'true': return None # disables pagination, returns full queryset return super().paginate_queryset(queryset, request, view) class LogoViewSet(viewsets.ModelViewSet): queryset = Logo.objects.all() serializer_class = LogoSerializer pagination_class = LogoPagination parser_classes = (MultiPartParser, FormParser, JSONParser) def get_permissions(self): if self.action in ["upload"]: return [IsAdmin()] if self.action in ["cache"]: return [AllowAny()] try: return [perm() for perm in permission_classes_by_action[self.action]] except KeyError: return [Authenticated()] def get_queryset(self): """Optimize queryset with prefetch and add filtering""" # Annotate channel_count and prefetch channels to avoid N+1 in LogoSerializer. queryset = ( Logo.objects .annotate(channel_count=Count('channels')) .prefetch_related('channels') .order_by('name') ) # Filter by specific IDs ids = self.request.query_params.getlist('ids') if ids: try: # Convert string IDs to integers and filter id_list = [int(id_str) for id_str in ids if id_str.isdigit()] if id_list: queryset = queryset.filter(id__in=id_list) except (ValueError, TypeError): pass # Invalid IDs, return empty queryset queryset = Logo.objects.none() # Filter by usage used_filter = self.request.query_params.get('used', None) if used_filter == 'true': queryset = queryset.filter(channel_count__gt=0) elif used_filter == 'false': queryset = queryset.filter(channel_count=0) # Filter by name name_filter = self.request.query_params.get('name', None) if name_filter: queryset = queryset.filter(name__icontains=name_filter) return queryset def create(self, request, *args, **kwargs): """Create a new logo entry""" serializer = self.get_serializer(data=request.data) if serializer.is_valid(): logo = serializer.save() return Response(self.get_serializer(logo).data, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def update(self, request, *args, **kwargs): """Update an existing logo""" partial = kwargs.pop('partial', False) instance = self.get_object() serializer = self.get_serializer(instance, data=request.data, partial=partial) if serializer.is_valid(): logo = serializer.save() return Response(self.get_serializer(logo).data) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def destroy(self, request, *args, **kwargs): """Delete a logo and remove it from any channels using it""" logo = self.get_object() delete_file = request.query_params.get('delete_file', 'false').lower() == 'true' # Check if it's a local file that should be deleted if delete_file and logo.url and logo.url.startswith('/data/logos'): try: if os.path.exists(logo.url): os.remove(logo.url) logger.info(f"Deleted local logo file: {logo.url}") except Exception as e: logger.error(f"Failed to delete logo file {logo.url}: {str(e)}") return Response( {"error": f"Failed to delete logo file: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) # Instead of preventing deletion, remove the logo from channels if logo.channels.exists(): channel_count = logo.channels.count() logo.channels.update(logo=None) logger.info(f"Removed logo {logo.name} from {channel_count} channels before deletion") return super().destroy(request, *args, **kwargs) @action(detail=False, methods=["post"]) def upload(self, request): if "file" not in request.FILES: return Response( {"error": "No file uploaded"}, status=status.HTTP_400_BAD_REQUEST ) file = request.FILES["file"] # Validate file try: from dispatcharr.utils import validate_logo_file validate_logo_file(file) except Exception as e: return Response( {"error": str(e)}, status=status.HTTP_400_BAD_REQUEST ) # Sanitize filename: strip directory components to prevent path traversal try: file_path = safe_upload_path(file.name, "/data/logos") except ValueError: return Response({"error": "Invalid filename."}, status=status.HTTP_400_BAD_REQUEST) os.makedirs("/data/logos", exist_ok=True) with open(file_path, "wb+") as destination: for chunk in file.chunks(): destination.write(chunk) # Mark file as processed in Redis to prevent file scanner notifications try: redis_client = RedisClient.get_client() if redis_client: # Use the same key format as the file scanner redis_key = f"processed_file:{file_path}" # Store the actual file modification time to match the file scanner's expectation file_mtime = os.path.getmtime(file_path) redis_client.setex(redis_key, 60 * 60 * 24 * 3, str(file_mtime)) # 3 day TTL logger.debug(f"Marked uploaded logo file as processed in Redis: {file_path} (mtime: {file_mtime})") except Exception as e: logger.warning(f"Failed to mark logo file as processed in Redis: {e}") # Get custom name from request data, fallback to filename custom_name = request.data.get('name', '').strip() logo_name = custom_name if custom_name else os.path.basename(file_path) logo, _ = Logo.objects.get_or_create( url=file_path, defaults={ "name": logo_name, }, ) # Use get_serializer to ensure proper context serializer = self.get_serializer(logo) return Response( serializer.data, status=status.HTTP_201_CREATED, ) @action(detail=True, methods=["get"], permission_classes=[AllowAny]) def cache(self, request, pk=None): """Streams the logo file, whether it's local or remote.""" logo = self.get_object() logo_url = logo.url if logo_url.startswith("/data"): # Local file if not os.path.exists(logo_url): raise Http404("Image not found") stat = os.stat(logo_url) # Get proper mime type (first item of the tuple) content_type, _ = mimetypes.guess_type(logo_url) if not content_type: content_type = "image/jpeg" # Default to a common image type # Use context manager and set Content-Disposition to inline response = StreamingHttpResponse( open(logo_url, "rb"), content_type=content_type ) response["Cache-Control"] = "public, max-age=14400" # Cache in browser for 4 hours response["Last-Modified"] = http_date(stat.st_mtime) response["Content-Disposition"] = 'inline; filename="{}"'.format( os.path.basename(logo_url) ) return response else: # Remote image # Skip URLs that recently failed to avoid blocking workers # on unreachable hosts (e.g., dead CDNs referenced by old recordings). fail_expiry = _logo_fetch_failures.get(logo_url) if fail_expiry and time.monotonic() < fail_expiry: raise Http404("Remote image temporarily unavailable") try: # Get the default user agent try: default_user_agent_id = CoreSettings.get_default_user_agent_id() user_agent_obj = UserAgent.objects.get(id=int(default_user_agent_id)) user_agent = user_agent_obj.user_agent except (CoreSettings.DoesNotExist, UserAgent.DoesNotExist, ValueError): # Fallback if default not found from core.utils import dispatcharr_user_agent user_agent = dispatcharr_user_agent() # Hard total timeout (connect + full download) prevents a slow # server dripping bytes from holding a greenlet indefinitely. _LOGO_TOTAL_TIMEOUT = 10 # seconds _LOGO_MAX_BYTES = 5 * 1024 * 1024 # 5 MB remote_response = requests.get( logo_url, stream=True, timeout=(3, 5), # (connect_timeout, read_timeout per chunk) headers={'User-Agent': user_agent} ) if remote_response.status_code == 200: # Eagerly read the full image with a total time + size cap # so the greenlet is released quickly. chunks = [] total = 0 deadline = time.monotonic() + _LOGO_TOTAL_TIMEOUT for chunk in remote_response.iter_content(chunk_size=8192): total += len(chunk) if total > _LOGO_MAX_BYTES: remote_response.close() raise Http404("Remote image too large") if time.monotonic() > deadline: remote_response.close() now = time.monotonic() _logo_fetch_failures[logo_url] = now + _LOGO_FAIL_TTL raise Http404("Remote image fetch timed out") chunks.append(chunk) body = b"".join(chunks) # Full read succeeded, clear any previous failure entry _logo_fetch_failures.pop(logo_url, None) # Try to get content type from response headers first content_type = remote_response.headers.get("Content-Type") # If no content type in headers or it's empty, guess based on URL if not content_type: content_type, _ = mimetypes.guess_type(logo_url) # If still no content type, default to common image type if not content_type: content_type = "image/jpeg" response = HttpResponse( body, content_type=content_type, ) response["Content-Length"] = str(len(body)) if remote_response.headers.get("Cache-Control"): response["Cache-Control"] = remote_response.headers.get("Cache-Control") if remote_response.headers.get("Last-Modified"): response["Last-Modified"] = remote_response.headers.get("Last-Modified") response["Content-Disposition"] = 'inline; filename="{}"'.format( os.path.basename(logo_url) ) return response # Non-200 response — cache the failure and evict stale entries now = time.monotonic() _logo_fetch_failures[logo_url] = now + _LOGO_FAIL_TTL if len(_logo_fetch_failures) > 256: for k in [k for k, v in _logo_fetch_failures.items() if v <= now]: _logo_fetch_failures.pop(k, None) raise Http404("Remote image not found") except requests.RequestException as e: now = time.monotonic() _logo_fetch_failures[logo_url] = now + _LOGO_FAIL_TTL if len(_logo_fetch_failures) > 256: for k in [k for k, v in _logo_fetch_failures.items() if v <= now]: _logo_fetch_failures.pop(k, None) logger.warning(f"Error fetching logo from {logo_url}: {e}") raise Http404("Error fetching remote image") class ChannelProfileViewSet(viewsets.ModelViewSet): queryset = ChannelProfile.objects.all() serializer_class = ChannelProfileSerializer def get_queryset(self): from django.db.models import Prefetch enabled_memberships_prefetch = Prefetch( 'channelprofilemembership_set', queryset=ChannelProfileMembership.objects.filter(enabled=True), to_attr='enabled_memberships', ) user = self.request.user if hasattr(user, "user_level") and user.user_level == 10: return ChannelProfile.objects.prefetch_related(enabled_memberships_prefetch) return self.request.user.channel_profiles.prefetch_related(enabled_memberships_prefetch) def get_permissions(self): if self.action == "duplicate": return [IsAdmin()] try: return [perm() for perm in permission_classes_by_action[self.action]] except KeyError: return [Authenticated()] @action(detail=True, methods=["post"], url_path="duplicate", permission_classes=[IsAdmin]) def duplicate(self, request, pk=None): requested_name = str(request.data.get("name", "")).strip() if not requested_name: return Response( {"detail": "Name is required to duplicate a profile."}, status=status.HTTP_400_BAD_REQUEST, ) if ChannelProfile.objects.filter(name=requested_name).exists(): return Response( {"detail": "A channel profile with this name already exists."}, status=status.HTTP_400_BAD_REQUEST, ) source_profile = self.get_object() with transaction.atomic(): new_profile = ChannelProfile.objects.create(name=requested_name) source_memberships = ChannelProfileMembership.objects.filter( channel_profile=source_profile ) source_enabled_map = { membership.channel_id: membership.enabled for membership in source_memberships } new_memberships = list( ChannelProfileMembership.objects.filter(channel_profile=new_profile) ) for membership in new_memberships: membership.enabled = source_enabled_map.get( membership.channel_id, False ) if new_memberships: ChannelProfileMembership.objects.bulk_update( new_memberships, ["enabled"] ) serializer = self.get_serializer(new_profile) return Response(serializer.data, status=status.HTTP_201_CREATED) class GetChannelStreamsAPIView(APIView): def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] def get(self, request, channel_id): channel = get_object_or_404(Channel, id=channel_id) # Order the streams by channelstream__order to match the order in the channel view streams = channel.streams.all().order_by("channelstream__order") serializer = StreamSerializer(streams, many=True) return Response(serializer.data) class GetChannelStreamStatsAPIView(APIView): """Returns a stats delta for a channel's streams (id, stream_stats, stream_stats_updated_at). Supports `since` (ISO 8601) and `ids` (comma-separated) query params.""" def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] @extend_schema( description=( "Return a minimal stats delta for the streams attached to a " "channel. Used by the channel table to refresh `stream_stats` " "on row expand and after a preview closes without re-pulling " "full stream rows." ), parameters=[ OpenApiParameter( name="since", type=OpenApiTypes.DATETIME, location=OpenApiParameter.QUERY, required=False, description=( "ISO 8601 timestamp. Returns only streams whose " "`stream_stats_updated_at` is strictly newer than this " "value. Omit to return all streams for the channel." ), ), OpenApiParameter( name="ids", type=OpenApiTypes.STR, location=OpenApiParameter.QUERY, required=False, description=( "Comma-separated stream IDs to restrict the response " "to. Combined with `since` via AND." ), ), ], responses={ 200: inline_serializer( name="ChannelStreamStatsDelta", fields={ "id": serializers.IntegerField(), "stream_stats": serializers.JSONField(allow_null=True), "stream_stats_updated_at": serializers.DateTimeField(allow_null=True), }, many=True, ), 400: inline_serializer( name="ChannelStreamStatsErrorResponse", fields={"detail": serializers.CharField()}, ), }, ) def get(self, request, channel_id): from django.utils.dateparse import parse_datetime get_object_or_404(Channel, id=channel_id) qs = Stream.objects.filter(channels=channel_id) since_raw = request.query_params.get("since") if since_raw: since_dt = parse_datetime(since_raw) if since_dt is None: return Response( {"detail": "Invalid 'since' value. Expected ISO 8601."}, status=status.HTTP_400_BAD_REQUEST, ) qs = qs.filter(stream_stats_updated_at__gt=since_dt) ids_raw = request.query_params.get("ids") if ids_raw: try: ids = [int(x) for x in ids_raw.split(",") if x.strip()] except ValueError: return Response( {"detail": "Invalid 'ids' value. Expected comma-separated integers."}, status=status.HTTP_400_BAD_REQUEST, ) qs = qs.filter(id__in=ids) data = list( qs.values("id", "stream_stats", "stream_stats_updated_at") ) return Response(data) class UpdateChannelMembershipAPIView(APIView): permission_classes = [IsOwnerOfObject] def patch(self, request, profile_id, channel_id): """Enable or disable a channel for a specific group""" channel_profile = get_object_or_404(ChannelProfile, id=profile_id) channel = get_object_or_404(Channel, id=channel_id) try: membership = ChannelProfileMembership.objects.get( channel_profile=channel_profile, channel=channel ) except ChannelProfileMembership.DoesNotExist: # Create the membership if it does not exist (for custom channels) membership = ChannelProfileMembership.objects.create( channel_profile=channel_profile, channel=channel, enabled=False # Default to False, will be updated below ) serializer = ChannelProfileMembershipSerializer( membership, data=request.data, partial=True ) if serializer.is_valid(): serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class BulkUpdateChannelMembershipAPIView(APIView): def get_permissions(self): try: return [ perm() for perm in permission_classes_by_method[self.request.method] ] except KeyError: return [Authenticated()] @extend_schema( description="Bulk enable or disable channels for a specific profile. Creates membership records if they don't exist.", request=BulkChannelProfileMembershipSerializer, ) def patch(self, request, profile_id): """Bulk enable or disable channels for a specific profile""" # Get the channel profile channel_profile = get_object_or_404(ChannelProfile, id=profile_id) # Validate the incoming data using the serializer serializer = BulkChannelProfileMembershipSerializer(data=request.data) if serializer.is_valid(): updates = serializer.validated_data["channels"] channel_ids = [entry["channel_id"] for entry in updates] # Validate that all channels exist existing_channels = set( Channel.objects.filter(id__in=channel_ids).values_list("id", flat=True) ) invalid_channels = [cid for cid in channel_ids if cid not in existing_channels] if invalid_channels: return Response( { "error": "Some channels do not exist", "invalid_channels": invalid_channels, }, status=status.HTTP_400_BAD_REQUEST, ) # Get existing memberships existing_memberships = ChannelProfileMembership.objects.filter( channel_profile=channel_profile, channel_id__in=channel_ids ) membership_dict = {m.channel_id: m for m in existing_memberships} # Prepare lists for bulk operations memberships_to_update = [] memberships_to_create = [] for entry in updates: channel_id = entry["channel_id"] enabled_status = entry["enabled"] if channel_id in membership_dict: # Update existing membership membership_dict[channel_id].enabled = enabled_status memberships_to_update.append(membership_dict[channel_id]) else: # Create new membership memberships_to_create.append( ChannelProfileMembership( channel_profile=channel_profile, channel_id=channel_id, enabled=enabled_status, ) ) # Perform bulk operations with transaction.atomic(): if memberships_to_update: ChannelProfileMembership.objects.bulk_update( memberships_to_update, ["enabled"] ) if memberships_to_create: ChannelProfileMembership.objects.bulk_create(memberships_to_create) return Response( { "status": "success", "updated": len(memberships_to_update), "created": len(memberships_to_create), "invalid_channels": [], }, status=status.HTTP_200_OK, ) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class RecurringRecordingRuleViewSet(viewsets.ModelViewSet): queryset = RecurringRecordingRule.objects.all().select_related("channel") serializer_class = RecurringRecordingRuleSerializer def get_permissions(self): return [IsAdmin()] def perform_create(self, serializer): rule = serializer.save() try: sync_recurring_rule_impl(rule.id, drop_existing=True) except Exception as err: logger.warning(f"Failed to initialize recurring rule {rule.id}: {err}") return rule def perform_update(self, serializer): rule = serializer.save() try: if rule.enabled: sync_recurring_rule_impl(rule.id, drop_existing=True) else: purge_recurring_rule_impl(rule.id) except Exception as err: logger.warning(f"Failed to resync recurring rule {rule.id}: {err}") return rule def perform_destroy(self, instance): rule_id = instance.id super().perform_destroy(instance) try: purge_recurring_rule_impl(rule_id) except Exception as err: logger.warning(f"Failed to purge recordings for rule {rule_id}: {err}") def _stop_dvr_clients(channel_uuid, recording_id=None): """Stop DVR recording clients for a channel. If recording_id is provided, only the client whose User-Agent contains that recording ID is stopped (safe for simultaneous recordings on the same channel). If recording_id is None, all Dispatcharr-DVR clients for the channel are stopped (used by destroy() when deleting a recording whose task_id is unknown). Returns the number of DVR clients stopped. """ from core.utils import RedisClient from apps.proxy.live_proxy.redis_keys import RedisKeys from apps.proxy.live_proxy.services.channel_service import ChannelService r = RedisClient.get_client() if not r: return 0 client_set_key = RedisKeys.clients(channel_uuid) client_ids = r.smembers(client_set_key) or [] stopped = 0 for raw_id in client_ids: try: cid = raw_id.decode("utf-8") if isinstance(raw_id, (bytes, bytearray)) else str(raw_id) meta_key = RedisKeys.client_metadata(channel_uuid, cid) ua = r.hget(meta_key, "user_agent") ua_s = ua.decode("utf-8") if isinstance(ua, (bytes, bytearray)) else (ua or "") if not (ua_s and "Dispatcharr-DVR" in ua_s): continue # When a recording_id is specified, only stop the client for that recording. # Each run_recording task connects with User-Agent "Dispatcharr-DVR/recording-{id}", # so we can safely target just this recording without affecting others on the channel. if recording_id is not None and f"recording-{recording_id}" not in ua_s: continue try: ChannelService.stop_client(channel_uuid, cid) stopped += 1 except Exception as inner_e: logger.debug(f"Failed to stop DVR client {cid} for channel {channel_uuid}: {inner_e}") except Exception as inner: logger.debug(f"Error while checking client metadata: {inner}") # Do not call ChannelService.stop_channel() here. # Stopping the channel proxy would terminate the source connection which may # be shared with other recordings on the same channel. The TS proxy server # already detects when client count reaches zero and tears down the channel # cleanly on its own (with the configured shutdown delay). return stopped # QueryParamJWTAuthentication supports native