diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c768e1e..320cf90b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,8 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Event-driven webhooks and script execution (Connect): Added new Connect feature that enables event-driven execution of custom scripts and webhooks in response to system events. Supports multiple event types including channel lifecycle (start, stop, reconnect, error, failover), stream operations (switch), recording events (start, end), data refreshes (EPG, M3U), and client activity (connect, disconnect). Event data is available as environment variables in scripts (prefixed with `DISPATCHARR_`), POST payloads for webhooks, and plugin execution payloads. Plugins can now subscribe to events by specifying an `events` array in their action definitions. Includes connection testing endpoint with dummy payloads for validation. (Closes #203) +- Cron scheduling support for M3U and EPG refreshes: Added interactive cron expression builder with preset buttons and custom field editors, plus info popover with common cron examples. Refactored backup scheduling to use shared ScheduleInput component for consistency across all scheduling interfaces. (Closes #165) +- Channel numbering modes for auto channel sync: Added three channel numbering modes when auto-syncing channels from M3U groups: + - **Fixed Start Number** (default): Start at a specified number and increment sequentially + - **Use Provider Number**: Use channel numbers from the M3U source (tvg-chno), with configurable fallback if provider number is missing + - **Next Available**: Auto-assign starting from 1, skipping all used channel numbers + Each mode includes its own configuration options accessible via the "Channel Numbering Mode" dropdown in auto sync settings. (Closes #956, #433) + +### Changed + +- XC API M3U stream URLs: M3U generation for Xtream Codes API endpoints now use proper XC-style stream URLs (`/live/username/password/channel_id`) instead of UUID-based stream endpoints, ensuring full compatibility with XC clients. (Fixes #839) + ### Fixed +- Auto channel sync duplicate channel numbers across groups: Fixed issue where multiple auto-sync groups starting at the same number would create duplicate channel numbers. The used channel number tracking now persists across all groups in a single sync operation, ensuring each assigned channel number is globally unique. - Modular mode PostgreSQL/Redis connection checks: Replaced raw Python socket checks with native tools (`pg_isready` for PostgreSQL and `socket.create_connection` for Redis) in modular deployment mode to prevent indefinite hangs in Docker environments with non-standard networking or DNS configurations. Now properly supports IPv4 and IPv6 configurations. (Fixes #952) - Thanks [@CodeBormen](https://github.com/CodeBormen) ## [0.19.0] - 2026-02-10 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/backups/scheduler.py b/apps/backups/scheduler.py index aa7e9bcd..a427757d 100644 --- a/apps/backups/scheduler.py +++ b/apps/backups/scheduler.py @@ -1,9 +1,13 @@ import json import logging -from django_celery_beat.models import PeriodicTask, CrontabSchedule +from django_celery_beat.models import PeriodicTask from core.models import CoreSettings +from core.scheduling import ( + create_or_update_periodic_task, + delete_periodic_task, +) logger = logging.getLogger(__name__) @@ -105,98 +109,25 @@ def _sync_periodic_task() -> None: settings = get_schedule_settings() if not settings["enabled"]: - # Delete the task if it exists - task = PeriodicTask.objects.filter(name=BACKUP_SCHEDULE_TASK_NAME).first() - if task: - old_crontab = task.crontab - task.delete() - _cleanup_orphaned_crontab(old_crontab) + delete_periodic_task(BACKUP_SCHEDULE_TASK_NAME) logger.info("Backup schedule disabled, removed periodic task") return - # Get old crontab before creating new one - old_crontab = None - try: - old_task = PeriodicTask.objects.get(name=BACKUP_SCHEDULE_TASK_NAME) - old_crontab = old_task.crontab - except PeriodicTask.DoesNotExist: - pass - # Check if using cron expression (advanced mode) if settings["cron_expression"]: - # Parse cron expression: "minute hour day month weekday" - try: - parts = settings["cron_expression"].split() - if len(parts) != 5: - raise ValueError("Cron expression must have 5 parts: minute hour day month weekday") - - minute, hour, day_of_month, month_of_year, day_of_week = parts - - crontab, _ = CrontabSchedule.objects.get_or_create( - minute=minute, - hour=hour, - day_of_week=day_of_week, - day_of_month=day_of_month, - month_of_year=month_of_year, - timezone=CoreSettings.get_system_time_zone(), - ) - except Exception as e: - logger.error(f"Invalid cron expression '{settings['cron_expression']}': {e}") - raise ValueError(f"Invalid cron expression: {e}") + cron_expr = settings["cron_expression"] else: - # Use simple frequency-based scheduling - # Parse time + # Build a cron expression from simple frequency settings hour, minute = settings["time"].split(":") - - # Build crontab based on frequency - system_tz = CoreSettings.get_system_time_zone() if settings["frequency"] == "daily": - crontab, _ = CrontabSchedule.objects.get_or_create( - minute=minute, - hour=hour, - day_of_week="*", - day_of_month="*", - month_of_year="*", - timezone=system_tz, - ) + cron_expr = f"{minute} {hour} * * *" else: # weekly - crontab, _ = CrontabSchedule.objects.get_or_create( - minute=minute, - hour=hour, - day_of_week=str(settings["day_of_week"]), - day_of_month="*", - month_of_year="*", - timezone=system_tz, - ) + cron_expr = f"{minute} {hour} * * {settings['day_of_week']}" - # Create or update the periodic task - task, created = PeriodicTask.objects.update_or_create( - name=BACKUP_SCHEDULE_TASK_NAME, - defaults={ - "task": "apps.backups.tasks.scheduled_backup_task", - "crontab": crontab, - "enabled": True, - "kwargs": json.dumps({"retention_count": settings["retention_count"]}), - }, + create_or_update_periodic_task( + task_name=BACKUP_SCHEDULE_TASK_NAME, + celery_task_path="apps.backups.tasks.scheduled_backup_task", + kwargs={"retention_count": settings["retention_count"]}, + cron_expression=cron_expr, + enabled=True, ) - - # Clean up old crontab if it changed and is orphaned - if old_crontab and old_crontab.id != crontab.id: - _cleanup_orphaned_crontab(old_crontab) - - action = "Created" if created else "Updated" - logger.info(f"{action} backup schedule: {settings['frequency']} at {settings['time']}") - - -def _cleanup_orphaned_crontab(crontab_schedule): - """Delete old CrontabSchedule if no other tasks are using it.""" - if crontab_schedule is None: - return - - # Check if any other tasks are using this crontab - if PeriodicTask.objects.filter(crontab=crontab_schedule).exists(): - logger.debug(f"CrontabSchedule {crontab_schedule.id} still in use, not deleting") - return - - logger.debug(f"Cleaning up orphaned CrontabSchedule: {crontab_schedule.id}") - crontab_schedule.delete() 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..de3903c0 --- /dev/null +++ b/apps/connect/api_views.py @@ -0,0 +1,196 @@ +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 django.utils import timezone +from .models import Integration, EventSubscription, DeliveryLog +from .serializers import ( + IntegrationSerializer, + EventSubscriptionSerializer, + DeliveryLogSerializer, +) +from apps.accounts.permissions import ( + Authenticated, + permission_classes_by_action, + IsAdmin, +) +from .handlers.webhook import WebhookHandler +from .handlers.script import ScriptHandler + + +class IntegrationViewSet(viewsets.ModelViewSet): + queryset = Integration.objects.all() + serializer_class = IntegrationSerializer + + def get_permissions(self): + try: + perms = permission_classes_by_action[self.action] + except KeyError: + # Respect view/action-specific permission_classes if provided; fallback to Authenticated + perms = getattr(self, "permission_classes", [Authenticated]) + return [perm() for perm in perms] + + @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) + + @action(detail=True, methods=["post"], url_path="test", permission_classes=[IsAdmin]) + def test(self, request, pk=None): + """ + Execute a saved integration (connect) with a dummy payload to verify configuration. + """ + try: + integration = Integration.objects.get(pk=pk) + except Integration.DoesNotExist: + return Response({"detail": "Integration not found"}, status=status.HTTP_404_NOT_FOUND) + + # Build a dummy payload similar to system events + now = timezone.now().isoformat() + dummy_payload = { + "event": "test", + "timestamp": now, + "channel_name": "Test Channel", + "stream_name": "Test Stream", + "stream_url": "http://example.com/stream.m3u8", + "channel_url": "http://example.com/stream.m3u8", + "provider_name": "Test Provider", + "profile_used": "Default", + "test": True, + } + + # Choose handler based on saved type + if integration.type == "webhook": + handler = WebhookHandler(integration, None, dummy_payload) + elif integration.type == "script": + handler = ScriptHandler(integration, None, dummy_payload) + else: + return Response( + {"success": False, "error": f"Unsupported integration type: {integration.type}"}, + status=status.HTTP_400_BAD_REQUEST, + ) + + try: + result = handler.execute() + return Response( + { + "success": bool(result.get("success")), + "type": integration.type, + "request_payload": dummy_payload, + "result": result, + }, + status=status.HTTP_200_OK, + ) + except Exception as e: + return Response( + { + "success": False, + "type": integration.type, + "request_payload": dummy_payload, + "error": str(e), + }, + status=status.HTTP_502_BAD_GATEWAY, + ) + + +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..b6aef79c --- /dev/null +++ b/apps/connect/handlers/script.py @@ -0,0 +1,81 @@ +# connect/handlers/script.py +import os +import stat +import subprocess +from django.conf import settings +from .base import IntegrationHandler + + +def _is_path_allowed(real_path: str) -> bool: + # Ensure path is within one of the allowed directories + for base in getattr(settings, "CONNECT_ALLOWED_SCRIPT_DIRS", []): + base_abs = os.path.abspath(base) + os.sep + if real_path.startswith(base_abs): + return True + return False + + +class ScriptHandler(IntegrationHandler): + def execute(self): + raw_path = self.integration.config.get("path") + if not raw_path: + raise ValueError("Missing 'path' in integration config") + + # Resolve and validate path + real_path = os.path.abspath(os.path.realpath(raw_path)) + + if not os.path.exists(real_path): + raise FileNotFoundError(f"Script not found: {real_path}") + + if not _is_path_allowed(real_path): + raise PermissionError( + f"Script path '{real_path}' not within allowed directories: " + f"{getattr(settings, 'CONNECT_ALLOWED_SCRIPT_DIRS', [])}" + ) + + if getattr(settings, "CONNECT_SCRIPT_REQUIRE_EXECUTABLE", True): + if not os.access(real_path, os.X_OK): + raise PermissionError(f"Script is not executable: {real_path}") + + if getattr(settings, "CONNECT_SCRIPT_DISALLOW_WORLD_WRITABLE", True): + st = os.stat(real_path) + if st.st_mode & stat.S_IWOTH: + raise PermissionError( + f"Refusing to execute world-writable script: {real_path}" + ) + + # Build a sanitized minimal environment; avoid inheriting secrets + env = { + "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", + } + for key, value in (self.payload or {}).items(): + env_key = f"DISPATCHARR_{str(key).upper()}" + env[env_key] = "" if value is None else str(value) + + # Run with a timeout to prevent hanging scripts + timeout = getattr(settings, "CONNECT_SCRIPT_TIMEOUT", 10) + max_out = getattr(settings, "CONNECT_SCRIPT_MAX_OUTPUT", 65536) + + result = subprocess.run( + [real_path], + capture_output=True, + text=True, + env=env, + timeout=timeout, + cwd=os.path.dirname(real_path) or None, + ) + + # Truncate outputs to avoid excessive memory/logging + stdout = result.stdout or "" + stderr = result.stderr or "" + if len(stdout) > max_out: + stdout = stdout[:max_out] + "... [truncated]" + if len(stderr) > max_out: + stderr = stderr[:max_out] + "... [truncated]" + + return { + "exit_code": result.returncode, + "stdout": stdout, + "stderr": 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..0b33a85e --- /dev/null +++ b/apps/connect/models.py @@ -0,0 +1,47 @@ +from django.db import models + +SUPPORTED_EVENTS = { + "channel_start": "Channel Started", + "channel_stop": "Channel Stopped", + "channel_reconnect": "Channel Reconnected", + "channel_error": "Channel Error", + "channel_failover": "Channel Failover", + "stream_switch": "Stream Switch", + "recording_start": "Recording Started", + "recording_end": "Recording Ended", + "epg_refresh": "EPG Refreshed", + "m3u_refresh": "M3U Refreshed", + "client_connect": "Client Connected", + "client_disconnect": "Client Disconnected", + "login_failed": "Login Failed", + "epg_blocked": "EPG Blocked", + "m3u_blocked": "M3U Blocked", +} + +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 = list(SUPPORTED_EVENTS.items()) + 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..832fc14b --- /dev/null +++ b/apps/connect/serializers.py @@ -0,0 +1,68 @@ +from rest_framework import serializers +from .models import Integration, EventSubscription, DeliveryLog +import os + + +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", + ] + + def validate(self, attrs): + type = attrs.get("type") if "type" in attrs else getattr(self.instance, "type", None) + config = attrs.get("config") if "config" in attrs else getattr(self.instance, "config", {}) + + if type == "script": + path = (config or {}).get("path") + if not path or not isinstance(path, str): + raise serializers.ValidationError({"config": "Script config must include a 'path' string"}) + + real_path = os.path.abspath(os.path.realpath(path)) + if not os.path.exists(real_path): + raise serializers.ValidationError({"config": f"Script path does not exist: {path}"}) + elif type == "webhook": + url = (config or {}).get("url") + if not url or not isinstance(url, str): + raise serializers.ValidationError({"config": "Webhook config must include a 'url' string"}) + else: + raise serializers.ValidationError({"type": "Unsupported integration type"}) + + return attrs + + +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..6144a6b3 --- /dev/null +++ b/apps/connect/utils.py @@ -0,0 +1,115 @@ +# connect/utils.py +import logging, json +from django.template import Template, Context +from .models import EventSubscription, DeliveryLog, SUPPORTED_EVENTS +from .handlers.webhook import WebhookHandler +from .handlers.script import ScriptHandler +from apps.plugins.loader import PluginManager + +logger = logging.getLogger(__name__) + +HANDLERS = { + "webhook": WebhookHandler, + "script": ScriptHandler, +} + + +def trigger_event(event_name, payload): + if event_name not in SUPPORTED_EVENTS: + logger.debug(f"Unsupported event '{event_name}' - skipping") + return + + 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}'") + + # First, fetch all subscriptions and trigger + 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}" + ) + + pm = PluginManager.get() + pm.discover_plugins(sync_db=False, use_cache=True) + plugins = pm.list_plugins() + + logger.debug(f"Checking {len(plugins)} plugins for event '{event_name}'") + for plugin in plugins: + if not plugin["enabled"]: + logger.debug(f"Skipping disabled plugin id={plugin.key} name={plugin.name}") + continue + + logger.debug(json.dumps(plugin)) + for action in plugin["actions"]: + if "events" in action and event_name in action["events"]: + key = plugin["key"] + params = {"event": event_name, "payload": payload} + action_name = action.get("label") or action.get("id") + action_id = action.get("id") + logger.debug( + f"Triggering plugin action for event '{event_name}' on plugin id={key} action={action_name}" + ) + if action_id: + pm.run_action(key, action_id, params) diff --git a/apps/epg/api_views.py b/apps/epg/api_views.py index 00f7403f..15613d4d 100644 --- a/apps/epg/api_views.py +++ b/apps/epg/api_views.py @@ -31,7 +31,9 @@ class EPGSourceViewSet(viewsets.ModelViewSet): API endpoint that allows EPG sources to be viewed or edited. """ - queryset = EPGSource.objects.all() + queryset = EPGSource.objects.select_related( + "refresh_task__crontab", "refresh_task__interval" + ).all() serializer_class = EPGSourceSerializer def get_permissions(self): diff --git a/apps/epg/serializers.py b/apps/epg/serializers.py index e4d5f466..60ce097b 100644 --- a/apps/epg/serializers.py +++ b/apps/epg/serializers.py @@ -12,6 +12,7 @@ class EPGSourceSerializer(serializers.ModelSerializer): allow_null=True, validators=[validate_flexible_url] ) + cron_expression = serializers.CharField(required=False, allow_blank=True, default='') class Meta: model = EPGSource @@ -24,6 +25,7 @@ class EPGSourceSerializer(serializers.ModelSerializer): 'is_active', 'file_path', 'refresh_interval', + 'cron_expression', 'priority', 'status', 'last_message', @@ -37,6 +39,34 @@ class EPGSourceSerializer(serializers.ModelSerializer): """Return the count of EPG data entries instead of all IDs to prevent large payloads""" return obj.epgs.count() + def to_representation(self, instance): + data = super().to_representation(instance) + # Derive cron_expression from the linked PeriodicTask's crontab (single source of truth) + # But first check if we have a transient _cron_expression (from create/update before signal runs) + cron_expr = '' + if hasattr(instance, '_cron_expression'): + cron_expr = instance._cron_expression + elif instance.refresh_task_id and instance.refresh_task and instance.refresh_task.crontab: + ct = instance.refresh_task.crontab + cron_expr = f'{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}' + data['cron_expression'] = cron_expr + return data + + def update(self, instance, validated_data): + cron_expr = validated_data.pop('cron_expression', '') + instance._cron_expression = cron_expr + for attr, value in validated_data.items(): + setattr(instance, attr, value) + instance.save() + return instance + + def create(self, validated_data): + cron_expr = validated_data.pop('cron_expression', '') + instance = EPGSource(**validated_data) + instance._cron_expression = cron_expr + instance.save() + return instance + class ProgramDataSerializer(serializers.ModelSerializer): class Meta: model = ProgramData diff --git a/apps/epg/signals.py b/apps/epg/signals.py index e41d3aaf..89deaa66 100644 --- a/apps/epg/signals.py +++ b/apps/epg/signals.py @@ -2,7 +2,7 @@ from django.db.models.signals import post_save, post_delete, pre_save from django.dispatch import receiver from .models import EPGSource, EPGData from .tasks import refresh_epg_data, delete_epg_refresh_task_by_id -from django_celery_beat.models import PeriodicTask, IntervalSchedule +from core.scheduling import create_or_update_periodic_task, delete_periodic_task from core.utils import is_protected_path, send_websocket_update import json import logging @@ -74,6 +74,7 @@ def create_or_update_refresh_task(sender, instance, **kwargs): """ Create or update a Celery Beat periodic task when an EPGSource is created/updated. Skip creating tasks for dummy EPG sources as they don't need refreshing. + Supports both interval-based and cron-based scheduling via the shared utility. """ # Skip task creation for dummy EPGs if instance.source_type == 'dummy': @@ -84,38 +85,23 @@ def create_or_update_refresh_task(sender, instance, **kwargs): return task_name = f"epg_source-refresh-{instance.id}" - interval, _ = IntervalSchedule.objects.get_or_create( - every=int(instance.refresh_interval), - period=IntervalSchedule.HOURS + should_be_enabled = instance.is_active + + # Read cron_expression from transient attribute set by the serializer + cron_expr = getattr(instance, "_cron_expression", "") + + task = create_or_update_periodic_task( + task_name=task_name, + celery_task_path="apps.epg.tasks.refresh_epg_data", + kwargs={"source_id": instance.id}, + interval_hours=int(instance.refresh_interval), + cron_expression=cron_expr, + enabled=should_be_enabled, ) - task, created = PeriodicTask.objects.get_or_create(name=task_name, defaults={ - "interval": interval, - "task": "apps.epg.tasks.refresh_epg_data", - "kwargs": json.dumps({"source_id": instance.id}), - "enabled": instance.refresh_interval != 0 and instance.is_active, - }) - - update_fields = [] - if created: - task.interval = interval - - if task.interval != interval: - task.interval = interval - update_fields.append("interval") - - # Check both refresh_interval and is_active to determine if task should be enabled - should_be_enabled = instance.refresh_interval != 0 and instance.is_active - if task.enabled != should_be_enabled: - task.enabled = should_be_enabled - update_fields.append("enabled") - - if update_fields: - task.save(update_fields=update_fields) - if instance.refresh_task != task: instance.refresh_task = task - instance.save(update_fields=["refresh_task"]) # Fixed field name + instance.save(update_fields=["refresh_task"]) @receiver(post_delete, sender=EPGSource) def delete_refresh_task(sender, instance, **kwargs): diff --git a/apps/m3u/api_views.py b/apps/m3u/api_views.py index 73331f7a..26e182d9 100644 --- a/apps/m3u/api_views.py +++ b/apps/m3u/api_views.py @@ -37,7 +37,9 @@ import json class M3UAccountViewSet(viewsets.ModelViewSet): """Handles CRUD operations for M3U accounts""" - queryset = M3UAccount.objects.prefetch_related("channel_group") + queryset = M3UAccount.objects.select_related( + "refresh_task__crontab", "refresh_task__interval" + ).prefetch_related("channel_group") serializer_class = M3UAccountSerializer def get_permissions(self): diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index a607dc07..8bfa7635 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -139,6 +139,7 @@ class M3UAccountSerializer(serializers.ModelSerializer): auto_enable_new_groups_live = serializers.BooleanField(required=False, write_only=True) auto_enable_new_groups_vod = serializers.BooleanField(required=False, write_only=True) auto_enable_new_groups_series = serializers.BooleanField(required=False, write_only=True) + cron_expression = serializers.CharField(required=False, allow_blank=True, default="") class Meta: model = M3UAccount @@ -158,6 +159,7 @@ class M3UAccountSerializer(serializers.ModelSerializer): "locked", "channel_groups", "refresh_interval", + "cron_expression", "custom_properties", "account_type", "username", @@ -188,9 +190,23 @@ class M3UAccountSerializer(serializers.ModelSerializer): data["auto_enable_new_groups_live"] = custom_props.get("auto_enable_new_groups_live", True) data["auto_enable_new_groups_vod"] = custom_props.get("auto_enable_new_groups_vod", True) data["auto_enable_new_groups_series"] = custom_props.get("auto_enable_new_groups_series", True) + + # Derive cron_expression from the linked PeriodicTask's crontab (single source of truth) + # But first check if we have a transient _cron_expression (from create/update before signal runs) + cron_expr = "" + if hasattr(instance, '_cron_expression'): + cron_expr = instance._cron_expression + elif instance.refresh_task_id and instance.refresh_task and instance.refresh_task.crontab: + ct = instance.refresh_task.crontab + cron_expr = f"{ct.minute} {ct.hour} {ct.day_of_month} {ct.month_of_year} {ct.day_of_week}" + data["cron_expression"] = cron_expr return data def update(self, instance, validated_data): + # Pop cron_expression before it reaches model fields + cron_expr = validated_data.pop("cron_expression", "") + instance._cron_expression = cron_expr + # Handle enable_vod preference and auto_enable_new_groups settings enable_vod = validated_data.pop("enable_vod", None) auto_enable_new_groups_live = validated_data.pop("auto_enable_new_groups_live", None) @@ -244,6 +260,9 @@ class M3UAccountSerializer(serializers.ModelSerializer): return instance def create(self, validated_data): + # Pop cron_expression — it's not a model field + cron_expr = validated_data.pop("cron_expression", "") + # Handle enable_vod preference and auto_enable_new_groups settings during creation enable_vod = validated_data.pop("enable_vod", False) auto_enable_new_groups_live = validated_data.pop("auto_enable_new_groups_live", True) @@ -260,7 +279,11 @@ class M3UAccountSerializer(serializers.ModelSerializer): custom_props["auto_enable_new_groups_series"] = auto_enable_new_groups_series validated_data["custom_properties"] = custom_props - return super().create(validated_data) + # Build instance manually so we can attach transient attr before save triggers signal + instance = M3UAccount(**validated_data) + instance._cron_expression = cron_expr + instance.save() + return instance def get_filters(self, obj): filters = obj.filters.order_by("order") diff --git a/apps/m3u/signals.py b/apps/m3u/signals.py index d014ac92..ced6f754 100644 --- a/apps/m3u/signals.py +++ b/apps/m3u/signals.py @@ -3,7 +3,7 @@ from django.db.models.signals import post_save, post_delete, pre_save from django.dispatch import receiver from .models import M3UAccount from .tasks import refresh_single_m3u_account, refresh_m3u_groups, delete_m3u_refresh_task_by_id -from django_celery_beat.models import PeriodicTask, IntervalSchedule +from core.scheduling import create_or_update_periodic_task, delete_periodic_task import json import logging @@ -23,48 +23,26 @@ def refresh_account_on_save(sender, instance, created, **kwargs): def create_or_update_refresh_task(sender, instance, **kwargs): """ Create or update a Celery Beat periodic task when an M3UAccount is created/updated. + Supports both interval-based and cron-based scheduling via the shared utility. """ task_name = f"m3u_account-refresh-{instance.id}" + should_be_enabled = instance.is_active - interval, _ = IntervalSchedule.objects.get_or_create( - every=int(instance.refresh_interval), - period=IntervalSchedule.HOURS + # Read cron_expression from transient attribute set by the serializer + cron_expr = getattr(instance, "_cron_expression", "") + + task = create_or_update_periodic_task( + task_name=task_name, + celery_task_path="apps.m3u.tasks.refresh_single_m3u_account", + kwargs={"account_id": instance.id}, + interval_hours=int(instance.refresh_interval), + cron_expression=cron_expr, + enabled=should_be_enabled, ) - # Task should be enabled only if refresh_interval != 0 AND account is active - should_be_enabled = (instance.refresh_interval != 0) and instance.is_active - - # First check if the task already exists to avoid validation errors - try: - task = PeriodicTask.objects.get(name=task_name) - # Task exists, just update it - updated_fields = [] - - if task.enabled != should_be_enabled: - task.enabled = should_be_enabled - updated_fields.append("enabled") - - if task.interval != interval: - task.interval = interval - updated_fields.append("interval") - - if updated_fields: - task.save(update_fields=updated_fields) - - # Ensure instance has the task - if instance.refresh_task_id != task.id: - M3UAccount.objects.filter(id=instance.id).update(refresh_task=task) - - except PeriodicTask.DoesNotExist: - # Create new task if it doesn't exist - refresh_task = PeriodicTask.objects.create( - name=task_name, - interval=interval, - task="apps.m3u.tasks.refresh_single_m3u_account", - kwargs=json.dumps({"account_id": instance.id}), - enabled=should_be_enabled, - ) - M3UAccount.objects.filter(id=instance.id).update(refresh_task=refresh_task) + # Ensure instance has the task linked + if instance.refresh_task_id != task.id: + M3UAccount.objects.filter(id=instance.id).update(refresh_task=task) @receiver(post_delete, sender=M3UAccount) def delete_refresh_task(sender, instance, **kwargs): diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index f929a208..0a1d4fc4 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -1648,6 +1648,13 @@ def sync_auto_channels(account_id, scan_start_time=None): channels_updated = 0 channels_deleted = 0 + # Get all channel numbers that are already in use by other channels (not auto-created by this account) + used_numbers = set( + Channel.objects.exclude( + auto_created=True, auto_created_by=account + ).values_list("channel_number", flat=True) + ) + for group_relation in auto_sync_groups: channel_group = group_relation.channel_group start_number = group_relation.auto_sync_channel_start or 1.0 @@ -1665,6 +1672,8 @@ def sync_auto_channels(account_id, scan_start_time=None): stream_profile_id = None custom_logo_id = None custom_epg_id = None # New option: select specific EPG source (takes priority over force_dummy_epg) + channel_numbering_mode = "fixed" # Default mode + channel_numbering_fallback = 1 # Default fallback for provider mode if group_relation.custom_properties: group_custom_props = group_relation.custom_properties force_dummy_epg = group_custom_props.get("force_dummy_epg", False) @@ -1682,6 +1691,8 @@ def sync_auto_channels(account_id, scan_start_time=None): ) stream_profile_id = group_custom_props.get("stream_profile_id") custom_logo_id = group_custom_props.get("custom_logo_id") + channel_numbering_mode = group_custom_props.get("channel_numbering_mode", "fixed") + channel_numbering_fallback = group_custom_props.get("channel_numbering_fallback", 1) # Determine which group to use for created channels target_group = channel_group @@ -1697,7 +1708,7 @@ def sync_auto_channels(account_id, scan_start_time=None): ) logger.info( - f"Processing auto sync for group: {channel_group.name} (start: {start_number})" + f"Processing auto sync for group: {channel_group.name} (mode: {channel_numbering_mode}, start: {start_number})" ) # Get all current streams in this group for this M3U account, filter out stale streams @@ -1837,21 +1848,35 @@ def sync_auto_channels(account_id, scan_start_time=None): channels_to_renumber = [] temp_channel_number = start_number - # Get all channel numbers that are already in use by other channels (not auto-created by this account) - used_numbers = set( - Channel.objects.exclude( - auto_created=True, auto_created_by=account - ).values_list("channel_number", flat=True) - ) - for stream in current_streams: if stream.id in existing_channel_map: channel = existing_channel_map[stream.id] - # Find next available number starting from temp_channel_number - target_number = temp_channel_number - while target_number in used_numbers: - target_number += 1 + # Determine target number based on numbering mode + if channel_numbering_mode == "provider": + # Use provider number if available, otherwise use fallback with next available logic + if stream.stream_chno is not None: + target_number = stream.stream_chno + # If provider number is already used, find next available + if target_number in used_numbers: + target_number = channel_numbering_fallback + while target_number in used_numbers: + target_number += 1 + else: + # No provider number, use fallback and find next available + target_number = channel_numbering_fallback + while target_number in used_numbers: + target_number += 1 + elif channel_numbering_mode == "next_available": + # Find next available starting from 1 + target_number = 1 + while target_number in used_numbers: + target_number += 1 + else: # fixed mode (default) + # Find next available number starting from temp_channel_number + target_number = temp_channel_number + while target_number in used_numbers: + target_number += 1 # Add this number to used_numbers so we don't reuse it in this batch used_numbers.add(target_number) @@ -1863,9 +1888,11 @@ def sync_auto_channels(account_id, scan_start_time=None): f"Will renumber channel '{channel.name}' to {target_number}" ) - temp_channel_number += 1.0 - if temp_channel_number % 1 != 0: # Has decimal - temp_channel_number = int(temp_channel_number) + 1.0 + # Only increment temp_channel_number in fixed mode + if channel_numbering_mode == "fixed": + temp_channel_number += 1.0 + if temp_channel_number % 1 != 0: # Has decimal + temp_channel_number = int(temp_channel_number) + 1.0 # Bulk update channel numbers if any need renumbering if channels_to_renumber: @@ -2060,10 +2087,31 @@ def sync_auto_channels(account_id, scan_start_time=None): else: # Create new channel - # Find next available channel number - target_number = current_channel_number - while target_number in used_numbers: - target_number += 1 + # Determine channel number based on numbering mode + if channel_numbering_mode == "provider": + # Use provider number if available, otherwise use fallback with next available logic + if stream.stream_chno is not None: + target_number = stream.stream_chno + # If provider number is already used, find next available from fallback + if target_number in used_numbers: + target_number = channel_numbering_fallback + while target_number in used_numbers: + target_number += 1 + else: + # No provider number, use fallback and find next available + target_number = channel_numbering_fallback + while target_number in used_numbers: + target_number += 1 + elif channel_numbering_mode == "next_available": + # Find next available starting from 1 + target_number = 1 + while target_number in used_numbers: + target_number += 1 + else: # fixed mode (default) + # Find next available channel number starting from current_channel_number + target_number = current_channel_number + while target_number in used_numbers: + target_number += 1 # Add this number to used_numbers used_numbers.add(target_number) @@ -2190,10 +2238,11 @@ def sync_auto_channels(account_id, scan_start_time=None): f"Created auto channel: {channel.channel_number} - {channel.name}" ) - # Increment channel number for next iteration - current_channel_number += 1.0 - if current_channel_number % 1 != 0: # Has decimal - current_channel_number = int(current_channel_number) + 1.0 + # Increment channel number for next iteration (only in fixed mode) + if channel_numbering_mode == "fixed": + current_channel_number += 1.0 + if current_channel_number % 1 != 0: # Has decimal + current_channel_number = int(current_channel_number) + 1.0 except Exception as e: logger.error( diff --git a/apps/output/views.py b/apps/output/views.py index 1d53237a..21377489 100644 --- a/apps/output/views.py +++ b/apps/output/views.py @@ -183,8 +183,9 @@ def generate_m3u(request, profile_name=None, user=None): # Check if this is an XC API request (has username/password in GET params and user is authenticated) xc_username = request.GET.get('username') xc_password = request.GET.get('password') + is_xc_request = user is not None and xc_username and xc_password - if user is not None and xc_username and xc_password: + if is_xc_request: # This is an XC API request - use XC-style EPG URL base_url = build_absolute_uri_with_port(request, '') epg_url = f"{base_url}/xmltv.php?username={xc_username}&password={xc_password}" @@ -254,8 +255,12 @@ def generate_m3u(request, profile_name=None, user=None): f'tvg-chno="{formatted_channel_number}" {tvc_guide_stationid}group-title="{group_title}",{channel.name}\n' ) - # Determine the stream URL based on the direct parameter - if use_direct_urls: + # Determine the stream URL based on request type + if is_xc_request: + # XC API request - use XC-style stream URL format + base_url = build_absolute_uri_with_port(request, '') + stream_url = f"{base_url}/live/{xc_username}/{xc_password}/{channel.id}" + elif use_direct_urls: # Try to get the first stream's direct URL first_stream = channel.streams.order_by('channelstream__order').first() if first_stream and first_stream.url: diff --git a/apps/plugins/serializers.py b/apps/plugins/serializers.py index 172af265..9f568054 100644 --- a/apps/plugins/serializers.py +++ b/apps/plugins/serializers.py @@ -9,6 +9,9 @@ class PluginActionSerializer(serializers.Serializer): button_label = serializers.CharField(required=False, allow_blank=True) button_variant = serializers.CharField(required=False, allow_blank=True) button_color = serializers.CharField(required=False, allow_blank=True) + events = serializers.ListField( + child=serializers.CharField(), required=False, allow_empty=True + ) class PluginFieldOptionSerializer(serializers.Serializer): diff --git a/core/scheduling.py b/core/scheduling.py new file mode 100644 index 00000000..0b4b78c6 --- /dev/null +++ b/core/scheduling.py @@ -0,0 +1,203 @@ +""" +Reusable scheduling utilities for creating/updating/deleting +Celery Beat periodic tasks with interval or cron-based schedules. +""" + +import json +import logging + +from django_celery_beat.models import CrontabSchedule, IntervalSchedule, PeriodicTask + +from core.models import CoreSettings + +logger = logging.getLogger(__name__) + + +def parse_cron_expression(cron_expression): + """ + Parse a 5-part cron expression into its components. + + Args: + cron_expression: A string like "0 3 * * *" + + Returns: + dict with keys: minute, hour, day_of_month, month_of_year, day_of_week + + Raises: + ValueError: If the expression is not valid 5-part cron. + """ + parts = cron_expression.strip().split() + if len(parts) != 5: + raise ValueError( + "Cron expression must have 5 parts: minute hour day month weekday" + ) + return { + "minute": parts[0], + "hour": parts[1], + "day_of_month": parts[2], + "month_of_year": parts[3], + "day_of_week": parts[4], + } + + +def create_or_update_periodic_task( + task_name, + celery_task_path, + kwargs=None, + interval_hours=0, + cron_expression="", + enabled=True, +): + """ + Create or update a Celery Beat PeriodicTask. Supports both interval + (hours) and cron-based scheduling. + + When *cron_expression* is provided and non-empty it takes precedence + over *interval_hours*. An interval_hours of 0 (with no cron) means + the task is disabled. + + Args: + task_name: Unique PeriodicTask name. + celery_task_path: Dotted path to the Celery task function. + kwargs: dict of keyword arguments passed to the task. + interval_hours: Interval in hours (0 = disabled when no cron). + cron_expression: 5-part cron string (empty = use interval). + enabled: Whether the task should be enabled. + + Returns: + The PeriodicTask instance (created or updated). + """ + task_kwargs = json.dumps(kwargs or {}) + + # Determine effective enabled state + use_cron = bool(cron_expression and cron_expression.strip()) + should_be_enabled = enabled and (use_cron or interval_hours > 0) + + # Retrieve existing task (if any) to track old schedule objects + old_interval = None + old_crontab = None + try: + existing = PeriodicTask.objects.get(name=task_name) + old_interval = existing.interval + old_crontab = existing.crontab + except PeriodicTask.DoesNotExist: + existing = None + + if use_cron: + # ---- Cron-based schedule ---- + cron_parts = parse_cron_expression(cron_expression) + system_tz = CoreSettings.get_system_time_zone() + + crontab, _ = CrontabSchedule.objects.get_or_create( + minute=cron_parts["minute"], + hour=cron_parts["hour"], + day_of_week=cron_parts["day_of_week"], + day_of_month=cron_parts["day_of_month"], + month_of_year=cron_parts["month_of_year"], + timezone=system_tz, + ) + + defaults = { + "task": celery_task_path, + "crontab": crontab, + "interval": None, + "enabled": should_be_enabled, + "kwargs": task_kwargs, + } + + task, created = PeriodicTask.objects.update_or_create( + name=task_name, defaults=defaults + ) + + # Clean up old interval if we switched from interval → cron + if old_interval: + _cleanup_orphaned_interval(old_interval) + # Clean up old crontab if it changed + if old_crontab and old_crontab.id != crontab.id: + _cleanup_orphaned_crontab(old_crontab) + + else: + # ---- Interval-based schedule ---- + interval, _ = IntervalSchedule.objects.get_or_create( + every=max(int(interval_hours), 1) if interval_hours else 1, + period=IntervalSchedule.HOURS, + ) + + defaults = { + "task": celery_task_path, + "interval": interval, + "crontab": None, + "enabled": should_be_enabled, + "kwargs": task_kwargs, + } + + task, created = PeriodicTask.objects.update_or_create( + name=task_name, defaults=defaults + ) + + # Clean up old crontab if we switched from cron → interval + if old_crontab: + _cleanup_orphaned_crontab(old_crontab) + # Clean up old interval if it changed + if old_interval and old_interval.id != interval.id: + _cleanup_orphaned_interval(old_interval) + + action = "Created" if created else "Updated" + mode = "cron" if use_cron else "interval" + logger.info(f"{action} periodic task '{task_name}' ({mode}, enabled={should_be_enabled})") + return task + + +def delete_periodic_task(task_name): + """ + Delete a PeriodicTask by name and clean up orphaned schedules. + + Args: + task_name: The unique name of the PeriodicTask. + + Returns: + True if a task was found and deleted, False otherwise. + """ + try: + task = PeriodicTask.objects.get(name=task_name) + except PeriodicTask.DoesNotExist: + logger.warning(f"No PeriodicTask found with name '{task_name}'") + return False + + old_interval = task.interval + old_crontab = task.crontab + task_id = task.id + + task.delete() + logger.info(f"Deleted periodic task '{task_name}' (id={task_id})") + + if old_interval: + _cleanup_orphaned_interval(old_interval) + if old_crontab: + _cleanup_orphaned_crontab(old_crontab) + + return True + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + +def _cleanup_orphaned_interval(interval_schedule): + """Delete an IntervalSchedule if no PeriodicTasks reference it.""" + if interval_schedule is None: + return + if PeriodicTask.objects.filter(interval=interval_schedule).exists(): + return + logger.debug(f"Cleaning up orphaned IntervalSchedule {interval_schedule.id}") + interval_schedule.delete() + + +def _cleanup_orphaned_crontab(crontab_schedule): + """Delete a CrontabSchedule if no PeriodicTasks reference it.""" + if crontab_schedule is None: + return + if PeriodicTask.objects.filter(crontab=crontab_schedule).exists(): + return + logger.debug(f"Cleaning up orphaned CrontabSchedule {crontab_schedule.id}") + crontab_schedule.delete() diff --git a/core/utils.py b/core/utils.py index c0e87a42..bd80782e 100644 --- a/core/utils.py +++ b/core/utils.py @@ -397,6 +397,78 @@ def validate_flexible_url(value): # If it doesn't match our flexible patterns, raise the original error raise ValidationError("Enter a valid URL.") +def dispatch_event_system(event_type, channel_id=None, channel_name=None, **details): + 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 def log_system_event(event_type, channel_id=None, channel_name=None, **details): """ @@ -423,6 +495,9 @@ def log_system_event(event_type, channel_id=None, channel_name=None, **details): details=details ) + # Trigger connect integrations for specific events + dispatch_event_system(event_type, channel_id=channel_id, channel_name=channel_name, **details) + # Get max events from settings (default 100) try: from .models import CoreSettings @@ -517,4 +592,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 c9bdebfa..1fbf566b 100644 --- a/dispatcharr/settings.py +++ b/dispatcharr/settings.py @@ -34,6 +34,7 @@ INSTALLED_APPS = [ "apps.proxy.apps.ProxyConfig", "apps.proxy.ts_proxy", "apps.vod.apps.VODConfig", + "apps.connect.apps.ConnectConfig", "core", "daphne", "drf_spectacular", @@ -411,3 +412,18 @@ LOGGING = { "level": LOG_LEVEL, # Use user-configured level instead of hardcoded 'INFO' }, } + +# Connect script execution safety settings +# Allowed base directories for custom scripts; real paths must be inside +_allowed_dirs_env = os.environ.get("DISPATCHARR_ALLOWED_SCRIPT_DIRS", "/data/plugins") +CONNECT_ALLOWED_SCRIPT_DIRS = [p for p in _allowed_dirs_env.split(":") if p] + +# Max execution time (seconds) for scripts +CONNECT_SCRIPT_TIMEOUT = int(os.environ.get("DISPATCHARR_SCRIPT_TIMEOUT", "10")) + +# Truncate stdout/stderr to this many characters to avoid large outputs +CONNECT_SCRIPT_MAX_OUTPUT = int(os.environ.get("DISPATCHARR_SCRIPT_MAX_OUTPUT", "65536")) + +# Require executable bit and disallow world-writable files +CONNECT_SCRIPT_REQUIRE_EXECUTABLE = True +CONNECT_SCRIPT_DISALLOW_WORLD_WRITABLE = True 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 6010296f..5291bba2 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 @@ -178,9 +179,34 @@ export default class API { static async getChannels() { try { - const response = await request(`${host}/api/channels/channels/`); + // Paginate through channels to avoid heavy single response + const pageSize = 200; + let page = 1; + let allChannels = []; - return response; + while (true) { + const data = await request( + `${host}/api/channels/channels/?page=${page}&page_size=${pageSize}` + ); + + // Backward compatibility: if endpoint returns an array (legacy), just return it + if (Array.isArray(data)) { + allChannels = data; + break; + } + + const results = Array.isArray(data?.results) ? data.results : []; + allChannels = allChannels.concat(results); + + const hasMore = Boolean(data?.next); + if (!hasMore || results.length === 0) { + break; + } + + page += 1; + } + + return allChannels; } catch (e) { errorNotification('Failed to retrieve channels', e); } @@ -3133,4 +3159,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 39e1f634..611eecfb 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 */} { - // Skip if it's * or contains special characters - if ( - value === '*' || - value.includes('/') || - value.includes('-') || - value.includes(',') - ) { - return null; - } - const num = parseInt(value, 10); - if (isNaN(num) || num < min || num > max) { - return `${name} must be between ${min} and ${max}`; - } - return null; - }; - - const minuteError = validateRange(minute, 0, 59, 'Minute'); - if (minuteError) return { valid: false, error: minuteError }; - - const hourError = validateRange(hour, 0, 23, 'Hour'); - if (hourError) return { valid: false, error: hourError }; - - const dayError = validateRange(dayOfMonth, 1, 31, 'Day'); - if (dayError) return { valid: false, error: dayError }; - - const monthError = validateRange(month, 1, 12, 'Month'); - if (monthError) return { valid: false, error: monthError }; - - const weekdayError = validateRange(dayOfWeek, 0, 6, 'Weekday'); - if (weekdayError) return { valid: false, error: weekdayError }; - - return { valid: true, error: null }; -} - const DAYS_OF_WEEK = [ { value: '0', label: 'Sunday' }, { value: '1', label: 'Monday' }, @@ -262,8 +175,7 @@ export default function BackupManager() { const [scheduleLoading, setScheduleLoading] = useState(false); const [scheduleSaving, setScheduleSaving] = useState(false); const [scheduleChanged, setScheduleChanged] = useState(false); - const [advancedMode, setAdvancedMode] = useState(false); - const [cronError, setCronError] = useState(null); + const [scheduleType, setScheduleType] = useState('interval'); // For 12-hour display mode const [displayTime, setDisplayTime] = useState('3:00'); @@ -373,12 +285,8 @@ export default function BackupManager() { try { const settings = await API.getBackupSchedule(); - // Check if using cron expression (advanced mode) - if (settings.cron_expression) { - setAdvancedMode(true); - } - setSchedule(settings); + setScheduleType(settings.cron_expression ? 'cron' : 'interval'); // Initialize 12-hour display values const { time, period } = to12Hour(settings.time); @@ -398,25 +306,9 @@ export default function BackupManager() { loadSchedule(); }, []); - // Validate cron expression when switching to advanced mode - useEffect(() => { - if (advancedMode && schedule.cron_expression) { - const validation = validateCronExpression(schedule.cron_expression); - setCronError(validation.valid ? null : validation.error); - } else { - setCronError(null); - } - }, [advancedMode, schedule.cron_expression]); - const handleScheduleChange = (field, value) => { setSchedule((prev) => ({ ...prev, [field]: value })); setScheduleChanged(true); - - // Validate cron expression if in advanced mode - if (field === 'cron_expression' && advancedMode) { - const validation = validateCronExpression(value); - setCronError(validation.valid ? null : validation.error); - } }; // Handle time changes in 12-hour mode @@ -442,9 +334,11 @@ export default function BackupManager() { const handleSaveSchedule = async () => { setScheduleSaving(true); try { - const scheduleToSave = advancedMode - ? schedule - : { ...schedule, cron_expression: '' }; + // Clear cron_expression if not in cron mode + const scheduleToSave = + scheduleType === 'cron' + ? schedule + : { ...schedule, cron_expression: '' }; const updated = await API.updateBackupSchedule(scheduleToSave); setSchedule(updated); @@ -603,207 +497,161 @@ export default function BackupManager() { /> - - - Advanced (Cron Expression) - - setAdvancedMode(e.currentTarget.checked)} - label={advancedMode ? 'Enabled' : 'Disabled'} - disabled={!schedule.enabled} - size="sm" - /> - + { + setScheduleType(type); + if (type !== 'cron') { + handleScheduleChange('cron_expression', ''); + } + }} + cronValue={schedule.cron_expression} + onCronChange={(expr) => handleScheduleChange('cron_expression', expr)} + disabled={!schedule.enabled} + switchToCronLabel="Use custom cron schedule" + switchToIntervalLabel="Use simple schedule" + > + {/* Simple mode: frequency / time / day selectors */} + + + + handleScheduleChange('day_of_week', parseInt(value, 10)) + } + data={DAYS_OF_WEEK} + disabled={!schedule.enabled} + /> + )} + {is12Hour ? ( + <> + { + const hour = displayTime + ? displayTime.split(':')[0] + : '12'; + handleTimeChange12h(`${hour}:${value}`, null); + }} + data={Array.from({ length: 60 }, (_, i) => ({ + value: String(i).padStart(2, '0'), + label: String(i).padStart(2, '0'), + }))} + disabled={!schedule.enabled} + searchable + /> + { + const minute = schedule.time + ? schedule.time.split(':')[1] + : '00'; + handleTimeChange24h(`${value}:${minute}`); + }} + data={Array.from({ length: 24 }, (_, i) => ({ + value: String(i).padStart(2, '0'), + label: String(i).padStart(2, '0'), + }))} + disabled={!schedule.enabled} + searchable + /> + - handleScheduleChange('frequency', value) - } - data={[ - { value: 'daily', label: 'Daily' }, - { value: 'weekly', label: 'Weekly' }, - ]} - disabled={!schedule.enabled} - /> - {schedule.frequency === 'weekly' && ( - { - const minute = displayTime - ? displayTime.split(':')[1] - : '00'; - handleTimeChange12h(`${value}:${minute}`, null); - }} - data={Array.from({ length: 12 }, (_, i) => ({ - value: String(i + 1), - label: String(i + 1), - }))} - disabled={!schedule.enabled} - searchable - /> - handleTimeChange12h(null, value)} - data={[ - { value: 'AM', label: 'AM' }, - { value: 'PM', label: 'PM' }, - ]} - disabled={!schedule.enabled} - /> - - ) : ( - <> - { - const hour = schedule.time - ? schedule.time.split(':')[0] - : '00'; - handleTimeChange24h(`${hour}:${value}`); - }} - data={Array.from({ length: 60 }, (_, i) => ({ - value: String(i).padStart(2, '0'), - label: String(i).padStart(2, '0'), - }))} - disabled={!schedule.enabled} - searchable - /> - - )} - - - - handleScheduleChange('retention_count', value || 0) - } - min={0} - disabled={!schedule.enabled} - /> - - - - )} + + + handleScheduleChange('retention_count', value || 0) + } + min={0} + disabled={!schedule.enabled} + /> + + {/* Timezone info - only show in simple mode */} - {!advancedMode && schedule.enabled && schedule.time && ( + {scheduleType !== 'cron' && schedule.enabled && schedule.time && ( System Timezone: {userTimezone} • Backup will run at{' '} {schedule.time} {userTimezone} diff --git a/frontend/src/components/cards/PluginCard.jsx b/frontend/src/components/cards/PluginCard.jsx index e16cec34..b8e69af0 100644 --- a/frontend/src/components/cards/PluginCard.jsx +++ b/frontend/src/components/cards/PluginCard.jsx @@ -14,9 +14,11 @@ import { Switch, Text, UnstyledButton, + Badge, } from '@mantine/core'; import { ChevronDown, ChevronRight, Trash2 } from 'lucide-react'; import { getConfirmationDetails } from '../../utils/cards/PluginCardUtils.js'; +import { SUBSCRIPTION_EVENTS } from '../../constants.js'; const PluginFieldList = ({ plugin, settings, updateField }) => { return plugin.fields.map((f) => ( @@ -44,6 +46,14 @@ const PluginActionList = ({ {action.description} )} + + Event Triggers + + {action.events.map((event) => ( + + {SUBSCRIPTION_EVENTS[event] || event} + + ))} + {isLoading &&
Loading...
} + {!isLoading && ( + + {integrations.map((i) => ( + + ))} + + )} + + setIsConnectionModalOpen(false)} + /> + + ); +} + +function IntegrationRow({ integration, editConnection, deleteConnection }) { + const type = integration.type || 'webhook'; + const [enabled, setEnabled] = useState(!!integration.enabled); + const webhookUrl = integration?.config?.url || ''; + const scriptPath = integration?.config?.path || ''; + + const toggleIntegration = async () => { + try { + await API.updateConnectIntegration(integration.id, { + ...integration, + enabled: !enabled, + }); + setEnabled(!enabled); + } catch (error) { + console.error('Failed to update integration', error); + } finally { + } + }; + + return ( + + + + + {integration.type == 'webhook' ? : } + {integration.name} + + + + + {type === 'webhook' ? ( + + Target: + + + + {webhookUrl} + + + + + ) : ( + + Target: + + + + {scriptPath} + + + + + )} + + Triggers + + {integration.subscriptions.map( + (sub) => + sub.enabled && ( + + {SUBSCRIPTION_EVENTS[sub.event] || sub.event} + + ) + )} + + + + + + + + + ); +} diff --git a/frontend/src/pages/ConnectLogs.jsx b/frontend/src/pages/ConnectLogs.jsx new file mode 100644 index 00000000..08705a13 --- /dev/null +++ b/frontend/src/pages/ConnectLogs.jsx @@ -0,0 +1,299 @@ +import React, { useEffect, useMemo, useState, useCallback } from 'react'; +import { + Box, + Title, + Badge, + Group, + Text, + Paper, + NativeSelect, + Pagination, + Select, + LoadingOverlay, +} from '@mantine/core'; +import API from '../api'; +import useConnectStore from '../store/connect'; +import { FileCode, Webhook } from 'lucide-react'; +import { SUBSCRIPTION_EVENTS } from '../constants'; +import { CustomTable, useTable } from '../components/tables/CustomTable'; +import { copyToClipboard } from '../utils'; + +export default function ConnectLogsPage() { + const { integrations, fetchIntegrations } = useConnectStore(); + + const [logs, setLogs] = useState([]); + const [isLoading, setIsLoading] = useState(false); + const [count, setCount] = useState(0); + const [pagination, setPagination] = useState({ pageIndex: 0, pageSize: 50 }); + const [filters, setFilters] = useState({ type: '', integration: '' }); + + const pageCount = useMemo( + () => Math.max(1, Math.ceil(count / Math.max(1, pagination.pageSize))), + [count, pagination.pageSize] + ); + + const onPageSizeChange = useCallback((e) => { + const value = parseInt(e.target.value, 10); + setPagination((prev) => ({ ...prev, pageSize: value, pageIndex: 0 })); + }, []); + + const onPageIndexChange = useCallback((page) => { + setPagination((prev) => ({ ...prev, pageIndex: page - 1 })); + }, []); + + const fetchLogs = useCallback(async () => { + setIsLoading(true); + try { + const params = { + page: pagination.pageIndex + 1, + page_size: pagination.pageSize, + }; + if (filters.type) params.type = filters.type; + if (filters.integration) params.integration = filters.integration; + + const data = await API.getConnectLogs(params); + const results = Array.isArray(data) ? data : data?.results || []; + setLogs(results); + setCount(data?.count || results.length || 0); + } finally { + setIsLoading(false); + } + }, [pagination.pageIndex, pagination.pageSize, filters]); + + useEffect(() => { + // Load integrations for filter options if not already available + if (!integrations || integrations.length === 0) { + fetchIntegrations?.(); + } + }, []); + + useEffect(() => { + fetchLogs(); + }, [fetchLogs]); + + const columns = useMemo( + () => [ + { + header: 'Time', + accessorKey: 'created_at', + size: 180, + cell: ({ getValue }) => ( + {new Date(getValue()).toLocaleString()} + ), + }, + { + header: 'Integration', + accessorKey: 'subscription', + size: 200, + cell: ({ getValue }) => { + const subscription = getValue(); + const integration = integrations.find( + (i) => i.id === subscription?.integration + ); + const isWebhook = integration?.type === 'webhook'; + return ( + + {isWebhook ? : } + {integration?.name || '-'} + + ); + }, + }, + { + header: 'Event', + accessorKey: 'subscription', + size: 160, + cell: ({ getValue }) => ( + {SUBSCRIPTION_EVENTS[getValue()?.event] || '—'} + ), + }, + { + header: 'Response', + accessorKey: 'response_payload', + grow: true, + cell: ({ getValue }) => ( + + copyToClipboard(getValue() ? JSON.stringify(getValue()) : '') + } + > + {getValue() ? JSON.stringify(getValue()) : '—'} + + ), + }, + { + header: 'Error', + accessorKey: 'error_message', + size: 150, + cell: ({ getValue }) => ( + copyToClipboard(getValue() || '')} + style={{ cursor: 'pointer' }} + > + {getValue() || '—'} + + ), + }, + { + header: 'Status', + accessorKey: 'status', + size: 100, + cell: ({ getValue }) => ( + + {getValue()} + + ), + }, + ], + [integrations] + ); + + const data = useMemo(() => logs, [logs]); + const allRowIds = useMemo(() => logs.map((l) => l.id), [logs]); + + const renderHeaderCell = (header) => ( + + {header.column.columnDef.header} + + ); + + const table = useTable({ + columns, + data, + allRowIds, + enablePagination: false, + enableRowSelection: false, + enableRowVirtualization: false, + renderTopToolbar: false, + manualSorting: false, + manualFiltering: false, + manualPagination: true, + headerCellRenderFns: { + created_at: renderHeaderCell, + subscription: renderHeaderCell, + response_payload: renderHeaderCell, + error_message: renderHeaderCell, + status: renderHeaderCell, + }, + }); + + const startIdx = pagination.pageIndex * pagination.pageSize + 1; + const endIdx = Math.min( + (pagination.pageIndex + 1) * pagination.pageSize, + count + ); + const paginationString = `Showing ${startIdx}-${endIdx} of ${count}`; + + const integrationOptions = useMemo( + () => integrations.map((i) => ({ value: String(i.id), label: i.name })), + [integrations] + ); + + return ( + + + Connect Logs + + + + Type + + setFilters((prev) => ({ ...prev, integration: value })) + } + style={{ width: 250 }} + /> + + + +
+ + +
+ + + + Page Size + + + {paginationString} + + + + + + ); +} diff --git a/frontend/src/store/connect.jsx b/frontend/src/store/connect.jsx new file mode 100644 index 00000000..f09f67f5 --- /dev/null +++ b/frontend/src/store/connect.jsx @@ -0,0 +1,46 @@ +import { create } from 'zustand'; +import API from '../api'; + +const useConnectStore = create((set, get) => ({ + integrations: [], + isLoading: false, + error: null, + + fetchIntegrations: async () => { + set({ isLoading: true, error: null }); + try { + const list = await API.getConnectIntegrations(); + console.log(list); + set({ + integrations: Array.isArray(list) ? list : list?.results || [], + isLoading: false, + }); + } catch (error) { + set({ error, isLoading: false }); + } + }, + + addIntegration: (integration) => + set((state) => ({ integrations: [...state.integrations, integration] })), + + updateIntegration: (integration) => + set((state) => ({ + integrations: state.integrations.map((i) => + i.id === integration.id ? integration : i + ), + })), + + removeIntegration: (id) => + set((state) => ({ + integrations: state.integrations.filter((i) => i.id !== id), + })), + + updateIntegrationSubscriptions: (id, events) => + set((state) => ({ + integrations: state.integrations.map((i) => + i.id === id ? { ...i, subscriptions: events } : i + ), + })), +})); + +export default useConnectStore; diff --git a/frontend/src/utils/cronUtils.js b/frontend/src/utils/cronUtils.js new file mode 100644 index 00000000..b52b5013 --- /dev/null +++ b/frontend/src/utils/cronUtils.js @@ -0,0 +1,63 @@ +/** + * Cron expression validation utility. + * + * Shared across CronModal, BackupManager, and any other component + * that needs to validate 5-part cron expressions. + */ + +export function validateCronExpression(expression) { + if (!expression || expression.trim() === '') { + return { valid: false, error: 'Cron expression is required' }; + } + + const parts = expression.trim().split(/\s+/); + if (parts.length !== 5) { + return { + valid: false, + error: + 'Cron expression must have exactly 5 parts: minute hour day month weekday', + }; + } + + const [minute, hour, dayOfMonth, month, dayOfWeek] = parts; + + const cronPartRegex = + /^(\*\/\d+|\*|\d+(-\d+)?(\/\d+)?(,\d+(-\d+)?(\/\d+)?)*)$/; + + const fields = [ + { value: minute, label: 'minute', min: 0, max: 59 }, + { value: hour, label: 'hour', min: 0, max: 23 }, + { value: dayOfMonth, label: 'day', min: 1, max: 31 }, + { value: month, label: 'month', min: 1, max: 12 }, + { value: dayOfWeek, label: 'weekday', min: 0, max: 6 }, + ]; + + for (const { value, label, min, max } of fields) { + if (!cronPartRegex.test(value)) { + return { + valid: false, + error: `Invalid ${label} field (${min}-${max}, *, or cron syntax)`, + }; + } + + // Extra numeric-range check for plain numbers + if ( + !( + value === '*' || + value.includes('/') || + value.includes('-') || + value.includes(',') + ) + ) { + const num = parseInt(value, 10); + if (isNaN(num) || num < min || num > max) { + return { + valid: false, + error: `${label.charAt(0).toUpperCase() + label.slice(1)} must be between ${min} and ${max}`, + }; + } + } + } + + return { valid: true, error: null }; +}