diff --git a/apps/accounts/api_views.py b/apps/accounts/api_views.py index 607ce199..188cd8cb 100644 --- a/apps/accounts/api_views.py +++ b/apps/accounts/api_views.py @@ -1,4 +1,5 @@ from django.contrib.auth import authenticate, login, logout +import logging from django.contrib.auth.models import Group, Permission from django.http import JsonResponse, HttpResponse from django.views.decorators.csrf import csrf_exempt @@ -15,6 +16,8 @@ from .models import User from .serializers import UserSerializer, GroupSerializer, PermissionSerializer from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView +logger = logging.getLogger(__name__) + class TokenObtainPairView(TokenObtainPairView): def post(self, request, *args, **kwargs): @@ -25,6 +28,7 @@ class TokenObtainPairView(TokenObtainPairView): username = request.data.get("username", 'unknown') client_ip = request.META.get('REMOTE_ADDR', 'unknown') user_agent = request.META.get('HTTP_USER_AGENT', 'unknown') + logger.info(f"Login blocked by network policy: user={username} ip={client_ip} ua={user_agent}") log_system_event( event_type='login_failed', user=username, @@ -43,6 +47,7 @@ class TokenObtainPairView(TokenObtainPairView): user_agent = request.META.get('HTTP_USER_AGENT', 'unknown') try: + logger.debug(f"Attempting JWT login for user={username}") response = super().post(request, *args, **kwargs) # If login was successful, update last_login and log success @@ -61,6 +66,7 @@ class TokenObtainPairView(TokenObtainPairView): client_ip=client_ip, user_agent=user_agent, ) + logger.info(f"Login success: user={username} ip={client_ip}") except User.DoesNotExist: pass # User doesn't exist, but login somehow succeeded else: @@ -72,6 +78,7 @@ class TokenObtainPairView(TokenObtainPairView): user_agent=user_agent, reason='Invalid credentials', ) + logger.info(f"Login failed: user={username} ip={client_ip}") return response @@ -84,6 +91,7 @@ class TokenObtainPairView(TokenObtainPairView): user_agent=user_agent, reason=f'Authentication error: {str(e)[:100]}', ) + logger.error(f"Login error for user={username}: {e}") raise # Re-raise the exception to maintain normal error flow @@ -95,6 +103,7 @@ class TokenRefreshView(TokenRefreshView): from core.utils import log_system_event client_ip = request.META.get('REMOTE_ADDR', 'unknown') user_agent = request.META.get('HTTP_USER_AGENT', 'unknown') + logger.info(f"Token refresh blocked by network policy: ip={client_ip} ua={user_agent}") log_system_event( event_type='login_failed', user='token_refresh', @@ -167,6 +176,7 @@ class AuthViewSet(viewsets.ViewSet): from core.utils import log_system_event client_ip = request.META.get('REMOTE_ADDR', 'unknown') user_agent = request.META.get('HTTP_USER_AGENT', 'unknown') + logger.debug(f"Login attempt via session: user={username} ip={client_ip}") if user: login(request, user) @@ -182,6 +192,7 @@ class AuthViewSet(viewsets.ViewSet): client_ip=client_ip, user_agent=user_agent, ) + logger.info(f"Login success via session: user={username} ip={client_ip}") return Response( { @@ -203,6 +214,7 @@ class AuthViewSet(viewsets.ViewSet): user_agent=user_agent, reason='Invalid credentials', ) + logger.info(f"Login failed via session: user={username} ip={client_ip}") return Response({"error": "Invalid credentials"}, status=400) @extend_schema( @@ -222,6 +234,7 @@ class AuthViewSet(viewsets.ViewSet): client_ip=client_ip, user_agent=user_agent, ) + logger.info(f"Logout: user={username} ip={client_ip}") logout(request) return Response({"message": "Logout successful"}) diff --git a/apps/api/urls.py b/apps/api/urls.py index 5a688778..b176bc1b 100644 --- a/apps/api/urls.py +++ b/apps/api/urls.py @@ -13,6 +13,7 @@ urlpatterns = [ path('plugins/', include(('apps.plugins.api_urls', 'plugins'), namespace='plugins')), path('vod/', include(('apps.vod.api_urls', 'vod'), namespace='vod')), path('backups/', include(('apps.backups.api_urls', 'backups'), namespace='backups')), + path('connect/', include(('apps.connect.api_urls', 'connect'), namespace='connect')), # path('output/', include(('apps.output.api_urls', 'output'), namespace='output')), #path('player/', include(('apps.player.api_urls', 'player'), namespace='player')), #path('settings/', include(('apps.settings.api_urls', 'settings'), namespace='settings')), diff --git a/apps/connect/__init__.py b/apps/connect/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/connect/api_urls.py b/apps/connect/api_urls.py new file mode 100644 index 00000000..664c437b --- /dev/null +++ b/apps/connect/api_urls.py @@ -0,0 +1,17 @@ +from django.urls import path +from rest_framework.routers import DefaultRouter +from .api_views import ( + IntegrationViewSet, + EventSubscriptionViewSet, + DeliveryLogViewSet, +) + +app_name = 'connect' + +router = DefaultRouter() +router.register(r'integrations', IntegrationViewSet, basename='integration') +router.register(r'subscriptions', EventSubscriptionViewSet, basename='subscription') +router.register(r'logs', DeliveryLogViewSet, basename='delivery-log') + +urlpatterns = [] +urlpatterns += router.urls diff --git a/apps/connect/api_views.py b/apps/connect/api_views.py new file mode 100644 index 00000000..1402de04 --- /dev/null +++ b/apps/connect/api_views.py @@ -0,0 +1,133 @@ +from rest_framework import viewsets, status +from rest_framework.pagination import PageNumberPagination +from django_filters.rest_framework import DjangoFilterBackend +from rest_framework.response import Response +from rest_framework.decorators import action +from .models import Integration, EventSubscription, DeliveryLog +from .serializers import ( + IntegrationSerializer, + EventSubscriptionSerializer, + DeliveryLogSerializer, +) +from apps.accounts.permissions import ( + Authenticated, + permission_classes_by_action, +) + + +class IntegrationViewSet(viewsets.ModelViewSet): + queryset = Integration.objects.all() + serializer_class = IntegrationSerializer + + def get_permissions(self): + try: + return [perm() for perm in permission_classes_by_action[self.action]] + except KeyError: + return [Authenticated()] + + @action(detail=True, methods=["get"], url_path="subscriptions") + def list_subscriptions(self, request, pk=None): + qs = EventSubscription.objects.filter(integration_id=pk) + serializer = EventSubscriptionSerializer(qs, many=True) + return Response(serializer.data) + + @action(detail=True, methods=["put"], url_path=r"subscriptions/set") + def set_subscriptions(self, request, pk=None): + """ + Replace the integration's subscriptions with the provided list. + Body format: [{"event": "channel_start", "enabled": true, "payload_template": "..."}, ...] + Any existing subscriptions not in the list will be deleted; missing ones will be created/updated. + """ + try: + integration = Integration.objects.get(pk=pk) + except Integration.DoesNotExist: + return Response( + {"detail": "Integration not found"}, status=status.HTTP_404_NOT_FOUND + ) + + data = request.data + if not isinstance(data, list): + return Response( + {"detail": "Expected a list of subscriptions"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + # Validate incoming items using serializer (without integration field) + # We'll attach the integration explicitly + valid_events = set(evt for evt, _ in EventSubscription.EVENT_CHOICES) + incoming = [] + for item in data: + if not isinstance(item, dict): + return Response( + {"detail": "Each subscription must be an object"}, + status=status.HTTP_400_BAD_REQUEST, + ) + event = item.get("event") + if event not in valid_events: + return Response( + {"detail": f"Invalid event: {event}"}, + status=status.HTTP_400_BAD_REQUEST, + ) + incoming.append( + { + "event": event, + "enabled": bool(item.get("enabled", True)), + "payload_template": item.get("payload_template"), + } + ) + + incoming_events = {s["event"] for s in incoming} + + # Delete subscriptions that are no longer present + EventSubscription.objects.filter(integration=integration).exclude( + event__in=incoming_events + ).delete() + + # Upsert incoming subscriptions + updated = [] + for sub in incoming: + obj, _created = EventSubscription.objects.update_or_create( + integration=integration, + event=sub["event"], + defaults={ + "enabled": sub["enabled"], + "payload_template": sub.get("payload_template"), + }, + ) + updated.append(obj) + + serializer = EventSubscriptionSerializer(updated, many=True) + return Response(serializer.data, status=status.HTTP_200_OK) + + +class EventSubscriptionViewSet(viewsets.ModelViewSet): + queryset = EventSubscription.objects.all() + serializer_class = EventSubscriptionSerializer + + +class DeliveryLogViewSet(viewsets.ReadOnlyModelViewSet): + queryset = DeliveryLog.objects.all().order_by("-created_at") + serializer_class = DeliveryLogSerializer + filter_backends = [DjangoFilterBackend] + + # Support server-side pagination with page_size query param + class ConnectLogsPagination(PageNumberPagination): + page_size = 50 + page_size_query_param = "page_size" + max_page_size = 250 + + pagination_class = ConnectLogsPagination + + def get_queryset(self): + qs = super().get_queryset() + + # Optional filters: integration id and type + integration_id = self.request.query_params.get("integration") + if integration_id: + qs = qs.filter(subscription__integration_id=integration_id) + + integration_type = self.request.query_params.get("type") + if integration_type: + qs = qs.filter(subscription__integration__type=integration_type) + + return qs diff --git a/apps/connect/apps.py b/apps/connect/apps.py new file mode 100644 index 00000000..db063fa4 --- /dev/null +++ b/apps/connect/apps.py @@ -0,0 +1,8 @@ +from django.apps import AppConfig + + +class ConnectConfig(AppConfig): + default_auto_field = 'django.db.models.BigAutoField' + name = 'apps.connect' + verbose_name = "Connect Integrations" + label = 'dispatcharr_connect' diff --git a/apps/connect/handlers/__init__.py b/apps/connect/handlers/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/connect/handlers/api.py b/apps/connect/handlers/api.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/connect/handlers/base.py b/apps/connect/handlers/base.py new file mode 100644 index 00000000..d61b7c53 --- /dev/null +++ b/apps/connect/handlers/base.py @@ -0,0 +1,12 @@ +# connect/handlers/base.py +import abc + +class IntegrationHandler(abc.ABC): + def __init__(self, integration, subscription, payload): + self.integration = integration + self.subscription = subscription + self.payload = payload + + @abc.abstractmethod + def execute(self): + pass diff --git a/apps/connect/handlers/script.py b/apps/connect/handlers/script.py new file mode 100644 index 00000000..3fa8215b --- /dev/null +++ b/apps/connect/handlers/script.py @@ -0,0 +1,28 @@ +# connect/handlers/script.py +import os +import subprocess +from .base import IntegrationHandler + +class ScriptHandler(IntegrationHandler): + def execute(self): + script_path = self.integration.config.get("path") + + # Build environment variables from payload (prefixed for clarity) + env = os.environ.copy() + for key, value in (self.payload or {}).items(): + # Convert keys to upper snake case and prefix + env_key = f"DISPATCHARR_{str(key).upper()}" + env[env_key] = str(value) if value is not None else "" + + result = subprocess.run( + [script_path], + capture_output=True, + text=True, + env=env, + ) + return { + "exit_code": result.returncode, + "stdout": result.stdout, + "stderr": result.stderr, + "success": result.returncode == 0, + } diff --git a/apps/connect/handlers/webhook.py b/apps/connect/handlers/webhook.py new file mode 100644 index 00000000..ccca530a --- /dev/null +++ b/apps/connect/handlers/webhook.py @@ -0,0 +1,10 @@ +# connect/handlers/webhook.py +import requests +from .base import IntegrationHandler + +class WebhookHandler(IntegrationHandler): + def execute(self): + url = self.integration.config.get("url") + headers = self.integration.config.get("headers", {}) + response = requests.post(url, json=self.payload, headers=headers, timeout=10) + return {"status_code": response.status_code, "body": response.text, "success": response.ok} diff --git a/apps/connect/migrations/0001_initial.py b/apps/connect/migrations/0001_initial.py new file mode 100644 index 00000000..cd8f6b42 --- /dev/null +++ b/apps/connect/migrations/0001_initial.py @@ -0,0 +1,52 @@ +# Generated by Django 5.2.9 on 2026-01-27 21:05 + +import django.db.models.deletion +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ] + + operations = [ + migrations.CreateModel( + name='EventSubscription', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('event', models.CharField(choices=[('channel_start', 'Channel Started'), ('channel_stop', 'Channel Stopped'), ('movie_added', 'Movie Added'), ('series_added', 'Series Added'), ('download_complete', 'Download Complete')], max_length=100)), + ('enabled', models.BooleanField(default=True)), + ('payload_template', models.TextField(blank=True, help_text='Optional Jinja2/Django template for customizing payload', null=True)), + ], + ), + migrations.CreateModel( + name='Integration', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=255)), + ('type', models.CharField(choices=[('webhook', 'Webhook'), ('api', 'API'), ('script', 'Custom Script')], max_length=50)), + ('config', models.JSONField(default=dict)), + ('enabled', models.BooleanField(default=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ], + ), + migrations.CreateModel( + name='DeliveryLog', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('status', models.CharField(choices=[('success', 'Success'), ('failed', 'Failed')], max_length=50)), + ('request_payload', models.JSONField(blank=True, default=dict)), + ('response_payload', models.JSONField(blank=True, default=dict)), + ('error_message', models.TextField(blank=True, null=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('subscription', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='logs', to='dispatcharr_connect.eventsubscription')), + ], + ), + migrations.AddField( + model_name='eventsubscription', + name='integration', + field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='subscriptions', to='dispatcharr_connect.integration'), + ), + ] diff --git a/apps/connect/migrations/__init__.py b/apps/connect/migrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/apps/connect/models.py b/apps/connect/models.py new file mode 100644 index 00000000..99e89ac8 --- /dev/null +++ b/apps/connect/models.py @@ -0,0 +1,36 @@ +from django.db import models + + +class Integration(models.Model): + TYPE_CHOICES = [ + ("webhook", "Webhook"), + ("api", "API"), + ("script", "Custom Script"), + ] + name = models.CharField(max_length=255) + type = models.CharField(max_length=50, choices=TYPE_CHOICES) + config = models.JSONField(default=dict) + enabled = models.BooleanField(default=True) + created_at = models.DateTimeField(auto_now_add=True) + + +class EventSubscription(models.Model): + EVENT_CHOICES = [ + ("channel_start", "Channel Started"), + ("channel_stop", "Channel Stopped"), + ("movie_added", "Movie Added"), + ("series_added", "Series Added"), + ("download_complete", "Download Complete"), + ] + event = models.CharField(max_length=100, choices=EVENT_CHOICES) + integration = models.ForeignKey(Integration, on_delete=models.CASCADE, related_name="subscriptions") + enabled = models.BooleanField(default=True) + payload_template = models.TextField(blank=True, null=True, help_text="Optional Jinja2/Django template for customizing payload") + +class DeliveryLog(models.Model): + subscription = models.ForeignKey(EventSubscription, on_delete=models.CASCADE, related_name="logs") + status = models.CharField(max_length=50, choices=[("success", "Success"), ("failed", "Failed")]) + request_payload = models.JSONField(default=dict, blank=True) + response_payload = models.JSONField(default=dict, blank=True) + error_message = models.TextField(blank=True, null=True) + created_at = models.DateTimeField(auto_now_add=True) diff --git a/apps/connect/serializers.py b/apps/connect/serializers.py new file mode 100644 index 00000000..859295f3 --- /dev/null +++ b/apps/connect/serializers.py @@ -0,0 +1,46 @@ +from rest_framework import serializers +from .models import Integration, EventSubscription, DeliveryLog + + +class EventSubscriptionSerializer(serializers.ModelSerializer): + class Meta: + model = EventSubscription + fields = [ + "id", + "event", + "enabled", + "payload_template", + "integration", + ] + + +class IntegrationSerializer(serializers.ModelSerializer): + subscriptions = EventSubscriptionSerializer(many=True, read_only=True) + + class Meta: + model = Integration + fields = [ + "id", + "name", + "type", + "config", + "enabled", + "created_at", + "subscriptions", + ] + + +class DeliveryLogSerializer(serializers.ModelSerializer): + subscription = EventSubscriptionSerializer(read_only=True) + + class Meta: + model = DeliveryLog + fields = [ + "id", + "subscription", + "status", + "request_payload", + "response_payload", + "error_message", + "created_at", + ] diff --git a/apps/connect/utils.py b/apps/connect/utils.py new file mode 100644 index 00000000..dcd44132 --- /dev/null +++ b/apps/connect/utils.py @@ -0,0 +1,73 @@ +# connect/utils.py +import logging +from django.template import Template, Context +from .models import EventSubscription, DeliveryLog +from .handlers.webhook import WebhookHandler +from .handlers.script import ScriptHandler + +logger = logging.getLogger(__name__) + +HANDLERS = { + "webhook": WebhookHandler, + "script": ScriptHandler, +} + + +def trigger_event(event_name, payload): + logger.debug(f"Triggering connect event: {event_name} payload_keys={list((payload or {}).keys())}") + subscriptions = ( + EventSubscription.objects.filter(event=event_name, enabled=True) + .select_related("integration") + ) + + count = subscriptions.count() + logger.info(f"Found {count} connect subscription(s) for event '{event_name}'") + + for sub in subscriptions: + integration = sub.integration + if not integration.enabled: + logger.debug(f"Skipping disabled integration id={integration.id} name={integration.name}") + continue + + # apply optional payload template + final_payload = payload + if sub.payload_template: + try: + template = Template(sub.payload_template) + rendered = template.render(Context(payload)) + final_payload = {"message": rendered} + except Exception as e: + logger.error(f"Payload template render failed for subscription id={sub.id}: {e}") + final_payload = payload + + handler_cls = HANDLERS.get(integration.type) + if not handler_cls: + DeliveryLog.objects.create( + subscription=sub, + status="failed", + request_payload=final_payload, + error_message=f"No handler for integration type '{integration.type}'", + ) + logger.error(f"No handler for integration type '{integration.type}' (integration id={integration.id})") + continue + + handler = handler_cls(integration, sub, final_payload) + logger.debug(f"Executing handler type={integration.type} integration_id={integration.id} subscription_id={sub.id}") + + try: + result = handler.execute() + DeliveryLog.objects.create( + subscription=sub, + status="success" if result.get("success") else "failed", + request_payload=final_payload, + response_payload=result, + ) + logger.info(f"Connect delivery succeeded for subscription id={sub.id} integration '{integration.name}'") + except Exception as e: + DeliveryLog.objects.create( + subscription=sub, + status="failed", + request_payload=final_payload, + error_message=str(e), + ) + logger.error(f"Connect delivery failed for subscription id={sub.id} integration '{integration.name}': {e}") diff --git a/core/utils.py b/core/utils.py index 5c5a0045..cdd9f0bd 100644 --- a/core/utils.py +++ b/core/utils.py @@ -415,6 +415,79 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details): details=details ) + # Trigger connect integrations for specific events + if event_type in ("channel_start", "channel_stop"): + try: + from apps.connect.utils import trigger_event + from apps.channels.models import Channel, Stream + from core.models import StreamProfile + from core.utils import RedisClient + + payload = {} + + channel_obj = None + if channel_id: + try: + channel_obj = Channel.objects.get(uuid=channel_id) + payload["channel_name"] = channel_obj.name + except Exception: + payload["channel_name"] = channel_name or None + else: + payload["channel_name"] = channel_name or None + + # Resolve current stream info + stream_id = details.get("stream_id") + stream_obj = None + if not stream_id and channel_obj: + try: + redis = RedisClient.get_client() + sid = redis.get(f"channel_stream:{channel_obj.id}") + if sid: + stream_id = int(sid) + except Exception: + stream_id = None + + if stream_id: + try: + stream_obj = Stream.objects.get(id=stream_id) + except Exception: + stream_obj = None + + # Populate stream details + payload["stream_name"] = getattr(stream_obj, "name", None) + payload["stream_url"] = getattr(stream_obj, "url", None) + + # Channel URL: use stream URL as best-effort + payload["channel_url"] = payload.get("stream_url") + + # Provider name from M3U account + provider_name = None + try: + if stream_obj and stream_obj.m3u_account: + provider_name = stream_obj.m3u_account.name + except Exception: + provider_name = None + payload["provider_name"] = provider_name + + # Profile used + profile_used = None + try: + if stream_id: + redis = RedisClient.get_client() + pid = redis.get(f"stream_profile:{stream_id}") + if pid: + profile = StreamProfile.objects.filter(id=int(pid)).first() + profile_used = profile.name if profile else None + except Exception: + profile_used = None + + payload["profile_used"] = profile_used + + trigger_event(event_type, payload) + except Exception as e: + # Don't fail main path if connect dispatch fails + pass + # Get max events from settings (default 100) try: from .models import CoreSettings @@ -509,4 +582,3 @@ def send_notification_dismissed(notification_key): ) except Exception as e: logger.error(f"Failed to send notification dismissed event: {e}") - diff --git a/dispatcharr/settings.py b/dispatcharr/settings.py index cfc499da..7afa75c5 100644 --- a/dispatcharr/settings.py +++ b/dispatcharr/settings.py @@ -31,6 +31,7 @@ INSTALLED_APPS = [ "apps.proxy.apps.ProxyConfig", "apps.proxy.ts_proxy", "apps.vod.apps.VODConfig", + "apps.connect.apps.ConnectConfig", "core", "daphne", "drf_spectacular", diff --git a/frontend/src/App.jsx b/frontend/src/App.jsx index 3869740e..86e8a7d7 100644 --- a/frontend/src/App.jsx +++ b/frontend/src/App.jsx @@ -14,6 +14,8 @@ import Stats from './pages/Stats'; import DVR from './pages/DVR'; import Settings from './pages/Settings'; import PluginsPage from './pages/Plugins'; +import ConnectPage from './pages/Connect'; +import ConnectLogsPage from './pages/ConnectLogs'; import Users from './pages/Users'; import LogosPage from './pages/Logos'; import VODsPage from './pages/VODs'; @@ -152,6 +154,11 @@ const App = () => { } /> } /> } /> + } /> + } + /> } /> } /> } /> diff --git a/frontend/src/api.js b/frontend/src/api.js index 9cc5494c..fe8bf275 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -12,6 +12,7 @@ import { notifications } from '@mantine/notifications'; import useChannelsTableStore from './store/channelsTable'; import useStreamsTableStore from './store/streamsTable'; import useUsersStore from './store/users'; +import useConnectStore from './store/connect'; // If needed, you can set a base host or keep it empty if relative requests const host = import.meta.env.DEV @@ -171,7 +172,7 @@ export default class API { static async logout() { return await request(`${host}/api/accounts/auth/logout/`, { - auth: true, // Send JWT token so backend can identify the user + auth: true, // Send JWT token so backend can identify the user method: 'POST', }); } @@ -261,15 +262,11 @@ export default class API { API.lastQueryParams = newParams; const [response, ids] = await Promise.all([ - request( - `${host}/api/channels/channels/?${newParams.toString()}` - ), + request(`${host}/api/channels/channels/?${newParams.toString()}`), API.getAllChannelIds(newParams), ]); - useChannelsTableStore - .getState() - .queryChannels(response, newParams); + useChannelsTableStore.getState().queryChannels(response, newParams); useChannelsTableStore.getState().setAllQueryIds(ids); return response; @@ -390,7 +387,8 @@ export default class API { channelData.channel_number === '' || channelData.channel_number === null || channelData.channel_number === undefined || - (typeof channelData.channel_number === 'string' && channelData.channel_number.trim() === '') + (typeof channelData.channel_number === 'string' && + channelData.channel_number.trim() === '') ) { delete channelData.channel_number; } @@ -719,7 +717,11 @@ export default class API { } } - static async createChannelsFromStreamsAsync(streamIds, channelProfileIds = null, startingChannelNumber = null) { + static async createChannelsFromStreamsAsync( + streamIds, + channelProfileIds = null, + startingChannelNumber = null + ) { try { const requestBody = { stream_ids: streamIds, @@ -815,15 +817,11 @@ export default class API { try { const [response, ids] = await Promise.all([ - request( - `${host}/api/channels/streams/?${params.toString()}` - ), + request(`${host}/api/channels/streams/?${params.toString()}`), API.getAllStreamIds(params), ]); - useStreamsTableStore - .getState() - .queryStreams(response, params); + useStreamsTableStore.getState().queryStreams(response, params); useStreamsTableStore.getState().setAllQueryIds(ids); return response; @@ -1179,13 +1177,10 @@ export default class API { static async getCurrentPrograms(channelIds = null) { try { - const response = await request( - `${host}/api/epg/current-programs/`, - { - method: 'POST', - body: { channel_ids: channelIds }, - } - ); + const response = await request(`${host}/api/epg/current-programs/`, { + method: 'POST', + body: { channel_ids: channelIds }, + }); return response; } catch (e) { @@ -1318,9 +1313,15 @@ export default class API { errorNotification('Failed to retrieve timezones', e); // Return fallback data instead of throwing return { - timezones: ['UTC', 'US/Eastern', 'US/Central', 'US/Mountain', 'US/Pacific'], + timezones: [ + 'UTC', + 'US/Eastern', + 'US/Central', + 'US/Mountain', + 'US/Pacific', + ], grouped: {}, - count: 5 + count: 5, }; } } @@ -1444,16 +1445,22 @@ export default class API { static async refreshAccountInfo(profileId) { try { - const response = await request(`${host}/api/m3u/refresh-account-info/${profileId}/`, { - method: 'POST', - }); + const response = await request( + `${host}/api/m3u/refresh-account-info/${profileId}/`, + { + method: 'POST', + } + ); return response; } catch (e) { // If it's a structured error response, return it instead of throwing if (e.body && typeof e.body === 'object') { return e.body; } - errorNotification(`Failed to refresh account info for profile ${profileId}`, e); + errorNotification( + `Failed to refresh account info for profile ${profileId}`, + e + ); throw e; } } @@ -1580,7 +1587,11 @@ export default class API { }); // Wait for the task to complete using token for auth - const result = await API.waitForBackupTask(response.task_id, onProgress, response.task_token); + const result = await API.waitForBackupTask( + response.task_id, + onProgress, + response.task_token + ); return result; } catch (e) { errorNotification('Failed to create backup', e); @@ -1593,13 +1604,10 @@ export default class API { const formData = new FormData(); formData.append('file', file); - const response = await request( - `${host}/api/backups/upload/`, - { - method: 'POST', - body: formData, - } - ); + const response = await request(`${host}/api/backups/upload/`, { + method: 'POST', + body: formData, + }); return response; } catch (e) { errorNotification('Failed to upload backup', e); @@ -1622,7 +1630,9 @@ export default class API { static async getDownloadToken(filename) { // Get a download token from the server try { - const response = await request(`${host}/api/backups/${encodeURIComponent(filename)}/download-token/`); + const response = await request( + `${host}/api/backups/${encodeURIComponent(filename)}/download-token/` + ); return response.token; } catch (e) { throw e; @@ -1666,7 +1676,11 @@ export default class API { // Wait for the task to complete using token for auth // Token-based auth allows status polling even after DB restore invalidates user sessions - const result = await API.waitForBackupTask(response.task_id, onProgress, response.task_token); + const result = await API.waitForBackupTask( + response.task_id, + onProgress, + response.task_token + ); return result; } catch (e) { errorNotification('Failed to restore backup', e); @@ -1741,17 +1755,27 @@ export default class API { return response; } catch (e) { // Show only the concise error message for plugin import - const msg = (e?.body && (e.body.error || e.body.detail)) || e?.message || 'Failed to import plugin'; - notifications.show({ title: 'Import failed', message: msg, color: 'red' }); + const msg = + (e?.body && (e.body.error || e.body.detail)) || + e?.message || + 'Failed to import plugin'; + notifications.show({ + title: 'Import failed', + message: msg, + color: 'red', + }); throw e; } } static async deletePlugin(key) { try { - const response = await request(`${host}/api/plugins/plugins/${key}/delete/`, { - method: 'DELETE', - }); + const response = await request( + `${host}/api/plugins/plugins/${key}/delete/`, + { + method: 'DELETE', + } + ); return response; } catch (e) { errorNotification('Failed to delete plugin', e); @@ -1776,10 +1800,13 @@ export default class API { static async runPluginAction(key, action, params = {}) { try { - const response = await request(`${host}/api/plugins/plugins/${key}/run/`, { - method: 'POST', - body: { action, params }, - }); + const response = await request( + `${host}/api/plugins/plugins/${key}/run/`, + { + method: 'POST', + body: { action, params }, + } + ); return response; } catch (e) { errorNotification('Failed to run plugin action', e); @@ -1788,10 +1815,13 @@ export default class API { static async setPluginEnabled(key, enabled) { try { - const response = await request(`${host}/api/plugins/plugins/${key}/enabled/`, { - method: 'POST', - body: { enabled }, - }); + const response = await request( + `${host}/api/plugins/plugins/${key}/enabled/`, + { + method: 'POST', + body: { enabled }, + } + ); return response; } catch (e) { errorNotification('Failed to update plugin enabled state', e); @@ -1973,7 +2003,7 @@ export default class API { if (!logoIds || logoIds.length === 0) return []; const params = new URLSearchParams(); - logoIds.forEach(id => params.append('ids', id)); + logoIds.forEach((id) => params.append('ids', id)); // Disable pagination for ID-based queries to get all matching logos params.append('no_pagination', 'true'); @@ -2440,10 +2470,13 @@ export default class API { static async updateRecurringRule(ruleId, payload) { try { - const response = await request(`${host}/api/channels/recurring-rules/${ruleId}/`, { - method: 'PATCH', - body: payload, - }); + const response = await request( + `${host}/api/channels/recurring-rules/${ruleId}/`, + { + method: 'PATCH', + body: payload, + } + ); return response; } catch (e) { errorNotification(`Failed to update recurring rule ${ruleId}`, e); @@ -2462,9 +2495,13 @@ export default class API { static async deleteRecording(id) { try { - await request(`${host}/api/channels/recordings/${id}/`, { method: 'DELETE' }); + await request(`${host}/api/channels/recordings/${id}/`, { + method: 'DELETE', + }); // Optimistically remove locally for instant UI update - try { useChannelsStore.getState().removeRecording(id); } catch {} + try { + useChannelsStore.getState().removeRecording(id); + } catch {} } catch (e) { errorNotification(`Failed to delete recording ${id}`, e); } @@ -2472,9 +2509,12 @@ export default class API { static async runComskip(recordingId) { try { - const resp = await request(`${host}/api/channels/recordings/${recordingId}/comskip/`, { - method: 'POST', - }); + const resp = await request( + `${host}/api/channels/recordings/${recordingId}/comskip/`, + { + method: 'POST', + } + ); // Refresh recordings list to reflect comskip status when done later // This endpoint just queues the task; the websocket/refresh will update eventually return resp; @@ -2512,7 +2552,9 @@ export default class API { static async deleteSeriesRule(tvgId) { try { const encodedTvgId = encodeURIComponent(tvgId); - await request(`${host}/api/channels/series-rules/${encodedTvgId}/`, { method: 'DELETE' }); + await request(`${host}/api/channels/series-rules/${encodedTvgId}/`, { + method: 'DELETE', + }); notifications.show({ title: 'Series rule removed' }); } catch (e) { errorNotification('Failed to remove series rule', e); @@ -2522,9 +2564,12 @@ export default class API { static async deleteAllUpcomingRecordings() { try { - const resp = await request(`${host}/api/channels/recordings/bulk-delete-upcoming/`, { - method: 'POST', - }); + const resp = await request( + `${host}/api/channels/recordings/bulk-delete-upcoming/`, + { + method: 'POST', + } + ); notifications.show({ title: `Removed ${resp.removed || 0} upcoming` }); useChannelsStore.getState().fetchRecordings(); return resp; @@ -2545,12 +2590,19 @@ export default class API { } } - static async bulkRemoveSeriesRecordings({ tvg_id, title = null, scope = 'title' }) { + static async bulkRemoveSeriesRecordings({ + tvg_id, + title = null, + scope = 'title', + }) { try { - const resp = await request(`${host}/api/channels/series-rules/bulk-remove/`, { - method: 'POST', - body: { tvg_id, title, scope }, - }); + const resp = await request( + `${host}/api/channels/series-rules/bulk-remove/`, + { + method: 'POST', + body: { tvg_id, title, scope }, + } + ); notifications.show({ title: `Removed ${resp.removed || 0} scheduled` }); return resp; } catch (e) { @@ -2712,13 +2764,10 @@ export default class API { try { // Use POST for large ID lists to avoid URL length limitations if (ids.length > 50) { - const response = await request( - `${host}/api/channels/streams/by-ids/`, - { - method: 'POST', - body: { ids }, - } - ); + const response = await request(`${host}/api/channels/streams/by-ids/`, { + method: 'POST', + body: { ids }, + }); return response; } else { // Use GET for small ID lists for backward compatibility @@ -2744,8 +2793,9 @@ export default class API { return response; } catch (e) { // Don't show error notification for "Invalid page" errors as they're handled gracefully - const isInvalidPage = e.body?.detail?.includes('Invalid page') || - e.message?.includes('Invalid page'); + const isInvalidPage = + e.body?.detail?.includes('Invalid page') || + e.message?.includes('Invalid page'); if (!isInvalidPage) { errorNotification('Failed to retrieve movies', e); @@ -2762,8 +2812,9 @@ export default class API { return response; } catch (e) { // Don't show error notification for "Invalid page" errors as they're handled gracefully - const isInvalidPage = e.body?.detail?.includes('Invalid page') || - e.message?.includes('Invalid page'); + const isInvalidPage = + e.body?.detail?.includes('Invalid page') || + e.message?.includes('Invalid page'); if (!isInvalidPage) { errorNotification('Failed to retrieve series', e); @@ -2774,7 +2825,10 @@ export default class API { static async getAllContent(params = new URLSearchParams()) { try { - console.log('Calling getAllContent with URL:', `${host}/api/vod/all/?${params.toString()}`); + console.log( + 'Calling getAllContent with URL:', + `${host}/api/vod/all/?${params.toString()}` + ); const response = await request( `${host}/api/vod/all/?${params.toString()}` ); @@ -2787,8 +2841,9 @@ export default class API { console.error('Error message:', e.message); // Don't show error notification for "Invalid page" errors as they're handled gracefully - const isInvalidPage = e.body?.detail?.includes('Invalid page') || - e.message?.includes('Invalid page'); + const isInvalidPage = + e.body?.detail?.includes('Invalid page') || + e.message?.includes('Invalid page'); if (!isInvalidPage) { errorNotification('Failed to retrieve content', e); @@ -2911,9 +2966,8 @@ export default class API { ); // Update the store with fetched notifications - const { default: useNotificationsStore } = await import( - './store/notifications' - ); + const { default: useNotificationsStore } = + await import('./store/notifications'); useNotificationsStore.getState().setNotifications(response.notifications); return response; @@ -2928,9 +2982,8 @@ export default class API { const response = await request(`${host}/api/core/notifications/count/`); // Update the store with the count - const { default: useNotificationsStore } = await import( - './store/notifications' - ); + const { default: useNotificationsStore } = + await import('./store/notifications'); useNotificationsStore.getState().setUnreadCount(response.unread_count); return response; @@ -2962,10 +3015,11 @@ export default class API { ); // Update the store - const { default: useNotificationsStore } = await import( - './store/notifications' - ); - useNotificationsStore.getState().dismissNotification(response.notification_key); + const { default: useNotificationsStore } = + await import('./store/notifications'); + useNotificationsStore + .getState() + .dismissNotification(response.notification_key); return response; } catch (e) { @@ -2984,9 +3038,8 @@ export default class API { ); // Update the store - const { default: useNotificationsStore } = await import( - './store/notifications' - ); + const { default: useNotificationsStore } = + await import('./store/notifications'); useNotificationsStore.getState().dismissAllNotifications(); return response; @@ -2994,4 +3047,124 @@ export default class API { errorNotification('Failed to dismiss all notifications', e); } } + + static async getConnectIntegrations() { + try { + return await request(`${host}/api/connect/integrations/`); + } catch (e) { + errorNotification('Failed to fetch connect integrations', e); + } + } + + static async createConnectIntegration(values) { + try { + const response = await request(`${host}/api/connect/integrations/`, { + method: 'POST', + body: values, + }); + + useConnectStore.getState().addIntegration(response); + + return response; + } catch (e) { + errorNotification('Failed to create integration', e); + } + } + + static async updateConnectIntegration(id, values) { + try { + const response = await request( + `${host}/api/connect/integrations/${id}/`, + { + method: 'PUT', + body: values, + } + ); + + if (response.id) { + useConnectStore.getState().updateIntegration(response); + } + + return response; + } catch (e) { + errorNotification('Failed to update integration', e); + } + } + + static async deleteConnectIntegration(id) { + try { + await request(`${host}/api/connect/integrations/${id}/`, { + method: 'DELETE', + }); + + useConnectStore.getState().removeIntegration(id); + + return true; + } catch (e) { + errorNotification('Failed to delete integration', e); + throw e; + } + } + + static async createConnectSubscription(values) { + try { + await request(`${host}/api/connect/subscriptions/`, { + method: 'POST', + body: values, + }); + + return true; + } catch (e) { + errorNotification('Failed to create subscription', e); + } + } + + static async listConnectSubscriptions(integrationId) { + try { + return await request( + `${host}/api/connect/integrations/${integrationId}/subscriptions/` + ); + } catch (e) { + errorNotification('Failed to fetch subscriptions', e); + } + } + + static async setConnectSubscriptions(integrationId, subscriptions) { + // subscriptions: [{ event, enabled, payload_template }] + console.log(subscriptions); + try { + const response = await request( + `${host}/api/connect/integrations/${integrationId}/subscriptions/set/`, + { + method: 'PUT', + body: subscriptions, + } + ); + + useConnectStore + .getState() + .updateIntegrationSubscriptions(integrationId, response); + + return true; + } catch (e) { + errorNotification('Failed to set subscriptions', e); + throw e; + } + } + + static async getConnectLogs(params = {}) { + try { + const search = new URLSearchParams(); + if (params.page) search.set('page', params.page); + if (params.page_size) search.set('page_size', params.page_size); + if (params.type) search.set('type', params.type); + if (params.integration) search.set('integration', params.integration); + + return await request( + `${host}/api/connect/logs/${search.toString() ? `?${search.toString()}` : ''}` + ); + } catch (e) { + errorNotification('Failed to fetch connect logs', e); + } + } } diff --git a/frontend/src/components/Sidebar.jsx b/frontend/src/components/Sidebar.jsx index 6c997cfe..4598427d 100644 --- a/frontend/src/components/Sidebar.jsx +++ b/frontend/src/components/Sidebar.jsx @@ -15,6 +15,11 @@ import { LogOut, User, FileImage, + Webhook, + Logs, + ChevronDown, + ChevronRight, + MonitorCog, } from 'lucide-react'; import { Avatar, @@ -27,6 +32,7 @@ import { TextInput, ActionIcon, Menu, + ScrollArea, } from '@mantine/core'; import { notifications } from '@mantine/notifications'; import logo from '../images/logo.png'; @@ -70,6 +76,74 @@ const NavLink = ({ item, isActive, collapsed }) => { ); }; +function NavGroup({ label, icon, paths, location, collapsed }) { + const [open, setOpen] = useState(() => + location.pathname.startsWith('/connect') + ); + + const parentActive = paths + .map((path) => path.path) + .includes(location.pathname); + + return ( + + setOpen((o) => !o)} + className={`navlink ${parentActive ? 'navlink-parent-active' : ''} ${open ? 'navlink-collapsed' : ''}`} + style={{ width: '100%' }} + > + {icon} + {!collapsed && ( + + + {label} + + + + {open ? : } + + + )} + + + {open && ( + + + {paths.map((child) => { + const active = location.pathname === child.path; + return ( + + + + ); + })} + + + )} + + ); +} + const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { const location = useLocation(); @@ -111,19 +185,41 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { { label: 'Stats', icon: , path: '/stats' }, { label: 'Plugins', icon: , path: '/plugins' }, { - label: 'Users', - icon: , - path: '/users', - }, - { - label: 'Logo Manager', - icon: , - path: '/logos', + label: 'Connect', + icon: , + paths: [ + { + label: 'Connections', + icon: , + path: '/connect', + }, + { + label: 'Logs', + icon: , + path: '/connect/logs', + }, + ], }, { label: 'Settings', icon: , - path: '/settings', + paths: [ + { + label: 'Users', + icon: , + path: '/users', + }, + { + label: 'Logo Manager', + icon: , + path: '/logos', + }, + { + label: 'System', + icon: , + path: '/settings', + }, + ], }, ] : [ @@ -205,20 +301,44 @@ const Sidebar = ({ collapsed, toggleDrawer, drawerWidth, miniDrawerWidth }) => { {/* Navigation Links */} - - {navItems.map((item) => { - const isActive = location.pathname === item.path; + + + {navItems.map((item) => { + if (item.paths) { + return ( + + ); + } - return ( - - ); - })} - + const isActive = location.pathname === item.path; + + return ( + + ); + })} + + {/* Profile Section */} ({ + value, + label, + }) +); + +const ConnectionForm = ({ connection = null, isOpen, onClose }) => { + const [submitting, setSubmitting] = useState(false); + const [selectedEvents, setSelectedEvents] = useState([]); + + // One-time form + const form = useForm({ + mode: 'controlled', + initialValues: { + name: connection?.name || '', + type: connection?.type || 'webhook', + url: connection?.config?.url || '', + script_path: connection?.config?.path || '', + enabled: connection?.enabled ?? true, + }, + validate: { + name: isNotEmpty('Provide a name'), + type: isNotEmpty('Select a type'), + url: (value, values) => { + if (values.type === 'webhook' && !value.trim()) { + return 'Provide a webhook URL'; + } + return null; + }, + script_path: (value, values) => { + if (values.type === 'script' && !value.trim()) { + return 'Provide a script path'; + } + return null; + }, + }, + }); + + useEffect(() => { + if (connection) { + const values = { + name: connection.name, + type: connection.type, + url: connection.config?.url, + script_path: connection.config?.path, + enabled: connection.enabled, + }; + form.setValues(values); + setSelectedEvents( + connection.subscriptions.reduce((acc, sub) => { + if (sub.enabled) acc.push(sub.event); + return acc; + }, []) + ); + } else { + form.reset(); + } + }, [connection]); + + const handleClose = () => { + onClose?.(); + }; + + const onSubmit = async (values) => { + console.log(values); + try { + setSubmitting(true); + const config = + values.type === 'webhook' + ? { url: values.url } + : { path: values.script_path }; + + if (connection) { + await API.updateConnectIntegration(connection.id, { + name: values.name, + type: values.type, + config, + enabled: values.enabled, + }); + } else { + connection = await API.createConnectIntegration({ + name: values.name, + type: values.type, + config, + enabled: values.enabled, + }); + } + + await API.setConnectSubscriptions( + connection.id, + Object.keys(SUBSCRIPTION_EVENTS).map((event) => ({ + event, + enabled: selectedEvents.includes(event), + })) + ); + handleClose(); + } catch (error) { + console.error('Failed to create connection', error); + } finally { + setSubmitting(false); + } + }; + + const toggleEvent = (event) => { + setSelectedEvents((prev) => + prev.includes(event) ? prev.filter((e) => e !== event) : [...prev, event] + ); + }; + + if (!isOpen) return null; + + return ( + + +
+ + + setFilters((prev) => ({ ...prev, type: value })) + } + style={{ width: 150 }} + /> + Integration +