import redis import logging import time import os import threading import re from django.conf import settings from redis.exceptions import ConnectionError, TimeoutError from django.core.cache import cache from asgiref.sync import async_to_sync from channels.layers import get_channel_layer from django.core.validators import URLValidator from django.core.exceptions import ValidationError import gc _REDIS_TLS_HINT = " (TLS is enabled — verify certificate paths and that Redis is configured for TLS)" logger = logging.getLogger(__name__) # Import the command detector from .command_utils import is_management_command def natural_sort_key(text): """ Convert a string into a list of string and number chunks for natural sorting. "PPV 10" becomes ['PPV ', 10] so it sorts correctly with "PPV 2". This function enables natural/alphanumeric sorting where numbers within strings are treated as actual numbers rather than strings. Args: text (str): The text to convert for sorting Returns: list: A list of strings and integers for proper sorting Example: >>> sorted(['PPV 1', 'PPV 10', 'PPV 2'], key=natural_sort_key) ['PPV 1', 'PPV 2', 'PPV 10'] """ def convert(chunk): return int(chunk) if chunk.isdigit() else chunk.lower() return [convert(c) for c in re.split('([0-9]+)', text)] class RedisClient: _client = None _buffer = None _pubsub_client = None @classmethod def _init_client(cls, decode_responses=True, max_retries=5, retry_interval=1): retry_count = 0 while retry_count < max_retries: try: # Get connection parameters from settings or environment redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost')) redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379))) redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0))) redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', '')) redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', '')) # Use standardized settings socket_timeout = getattr(settings, 'REDIS_SOCKET_TIMEOUT', 5) socket_connect_timeout = getattr(settings, 'REDIS_SOCKET_CONNECT_TIMEOUT', 5) health_check_interval = getattr(settings, 'REDIS_HEALTH_CHECK_INTERVAL', 30) socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True) retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True) # TLS params from settings (empty dict when TLS is disabled) ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {}) # Create Redis client with better defaults client = redis.Redis( host=redis_host, port=redis_port, db=redis_db, password=redis_password if redis_password else None, username=redis_user if redis_user else None, socket_timeout=socket_timeout, socket_connect_timeout=socket_connect_timeout, socket_keepalive=socket_keepalive, health_check_interval=health_check_interval, retry_on_timeout=retry_on_timeout, decode_responses=decode_responses, **ssl_params ) # Validate connection with ping client.ping() # Disable persistence on first connection - improves performance # Only try to disable if not in a read-only environment try: client.config_set('save', '') # Disable RDB snapshots client.config_set('appendonly', 'no') # Disable AOF logging # Disable protected mode when in debug mode if os.environ.get('DISPATCHARR_DEBUG', '').lower() == 'true': client.config_set('protected-mode', 'no') # Disable protected mode in debug logger.warning("Redis protected mode disabled for debug environment") logger.trace("Redis persistence disabled for better performance") except redis.exceptions.ResponseError as e: # Improve error handling for Redis configuration errors if "OOM" in str(e): logger.error(f"Redis OOM during configuration: {e}") # Try to increase maxmemory as an emergency measure try: client.config_set('maxmemory', '768mb') logger.warning("Applied emergency Redis memory increase to 768MB") except: pass else: logger.error(f"Redis configuration error: {e}") logger.info(f"Connected to Redis at {redis_host}:{redis_port}/{redis_db}") return client except (ConnectionError, TimeoutError) as e: retry_count += 1 _tls_hint = _REDIS_TLS_HINT if ssl_params else "" if retry_count >= max_retries: logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}{_tls_hint}") return None else: # Use exponential backoff for retries wait_time = retry_interval * (2 ** (retry_count - 1)) logger.warning(f"Redis connection failed. Retrying in {wait_time}s... ({retry_count}/{max_retries})") time.sleep(wait_time) except Exception as e: _tls_hint = "" try: _tls_hint = _REDIS_TLS_HINT if ssl_params else "" except NameError: pass logger.error(f"Unexpected error connecting to Redis: {e}{_tls_hint}") return None return None @classmethod def get_client(cls, max_retries=5, retry_interval=1): """Get Redis client optimized for non-binary data (decoded responses)""" if cls._client is None: cls._client = cls._init_client(decode_responses=True, max_retries=max_retries, retry_interval=retry_interval) return cls._client @classmethod def get_buffer(cls, max_retries=5, retry_interval=1): """Get Redis client optimized for binary data (no decoding)""" if cls._buffer is None: cls._buffer = cls._init_client(decode_responses=False, max_retries=max_retries, retry_interval=retry_interval) return cls._buffer @classmethod def get_pubsub_client(cls, max_retries=5, retry_interval=1): """Get Redis client optimized for PubSub operations""" if cls._pubsub_client is None: retry_count = 0 while retry_count < max_retries: try: # Get connection parameters from settings or environment redis_host = os.environ.get("REDIS_HOST", getattr(settings, 'REDIS_HOST', 'localhost')) redis_port = int(os.environ.get("REDIS_PORT", getattr(settings, 'REDIS_PORT', 6379))) redis_db = int(os.environ.get("REDIS_DB", getattr(settings, 'REDIS_DB', 0))) redis_password = os.environ.get("REDIS_PASSWORD", getattr(settings, 'REDIS_PASSWORD', '')) redis_user = os.environ.get("REDIS_USER", getattr(settings, 'REDIS_USER', '')) # Use standardized settings but without socket timeouts for PubSub # Important: socket_timeout is None for PubSub operations socket_connect_timeout = getattr(settings, 'REDIS_SOCKET_CONNECT_TIMEOUT', 5) socket_keepalive = getattr(settings, 'REDIS_SOCKET_KEEPALIVE', True) health_check_interval = getattr(settings, 'REDIS_HEALTH_CHECK_INTERVAL', 30) retry_on_timeout = getattr(settings, 'REDIS_RETRY_ON_TIMEOUT', True) ssl_params = getattr(settings, 'REDIS_SSL_PARAMS', {}) # Create Redis client with PubSub-optimized settings - no timeout client = redis.Redis( host=redis_host, port=redis_port, db=redis_db, password=redis_password if redis_password else None, username=redis_user if redis_user else None, socket_timeout=None, # Critical: No timeout for PubSub operations socket_connect_timeout=socket_connect_timeout, socket_keepalive=socket_keepalive, health_check_interval=health_check_interval, retry_on_timeout=retry_on_timeout, decode_responses=True, **ssl_params ) # Validate connection with ping client.ping() logger.info(f"Connected to Redis for PubSub at {redis_host}:{redis_port}/{redis_db}") # We don't need the keepalive thread anymore since we're using proper PubSub handling cls._pubsub_client = client break except (ConnectionError, TimeoutError) as e: retry_count += 1 _tls_hint = _REDIS_TLS_HINT if ssl_params else "" if retry_count >= max_retries: logger.error(f"Failed to connect to Redis for PubSub after {max_retries} attempts: {e}{_tls_hint}") return None else: # Use exponential backoff for retries wait_time = retry_interval * (2 ** (retry_count - 1)) logger.warning(f"Redis PubSub connection failed. Retrying in {wait_time}s... ({retry_count}/{max_retries})") time.sleep(wait_time) except Exception as e: _tls_hint = _REDIS_TLS_HINT if ssl_params else "" logger.error(f"Unexpected error connecting to Redis for PubSub: {e}{_tls_hint}") return None return cls._pubsub_client def acquire_task_lock(task_name, id): """Acquire a lock to prevent concurrent task execution.""" redis_client = RedisClient.get_client() lock_id = f"task_lock_{task_name}_{id}" # Use the Redis SET command with NX (only set if not exists) and EX (set expiration) lock_acquired = redis_client.set(lock_id, "locked", ex=300, nx=True) if not lock_acquired: logger.warning(f"Lock for {task_name} and id={id} already acquired. Task will not proceed.") return lock_acquired def release_task_lock(task_name, id): """Release the lock after task execution.""" redis_client = RedisClient.get_client() lock_id = f"task_lock_{task_name}_{id}" # Remove the lock redis_client.delete(lock_id) class TaskLockRenewer: """Periodically renews a Redis task lock to prevent expiry during long-running tasks. Use as a context manager after acquiring a lock: if acquire_task_lock("my_task", task_id): with TaskLockRenewer("my_task", task_id): # ... long-running work ... release_task_lock("my_task", task_id) A daemon thread extends the lock TTL at regular intervals so that slow downloads or large parsing jobs don't lose their lock mid-operation. """ def __init__(self, task_name, id, ttl=300, renewal_interval=120): self.task_name = task_name self.id = id self.ttl = ttl self.renewal_interval = renewal_interval self.lock_id = f"task_lock_{task_name}_{id}" self._stop_event = threading.Event() self._thread = None def _renew_loop(self): """Background loop that extends the lock TTL until stopped.""" while not self._stop_event.wait(self.renewal_interval): try: redis_client = RedisClient.get_client() if redis_client.exists(self.lock_id): redis_client.expire(self.lock_id, self.ttl) logger.debug( f"Renewed lock {self.lock_id} TTL to {self.ttl}s" ) else: # Lock was deleted externally (e.g. manual release) — stop renewing logger.warning( f"Lock {self.lock_id} no longer exists, stopping renewal" ) break except Exception as e: logger.error(f"Error renewing lock {self.lock_id}: {e}") def start(self): """Start the background renewal thread.""" self._stop_event.clear() self._thread = threading.Thread( target=self._renew_loop, daemon=True, name=f"lock-renew-{self.task_name}-{self.id}" ) self._thread.start() return self def stop(self): """Stop the renewal thread.""" self._stop_event.set() if self._thread and self._thread.is_alive(): self._thread.join(timeout=5) self._thread = None def __enter__(self): self.start() return self def __exit__(self, exc_type, exc_val, exc_tb): self.stop() return False def send_websocket_update(group_name, event_type, data, collect_garbage=False): """ Standardized function to send WebSocket updates with proper memory management. In uWSGI + gevent deployments, async_to_sync creates an asyncio event loop whose native EpollSelector blocks the entire OS thread, freezing all gevent greenlets in the worker. To avoid this, the actual send is offloaded to a real OS thread from gevent's native threadpool when monkey-patching is active. """ channel_layer = get_channel_layer() message = {'type': event_type, 'data': data} def _do_send(): try: async_to_sync(channel_layer.group_send)(group_name, message) except Exception as e: logger.warning(f"Failed to send WebSocket update: {e}") try: import gevent.monkey if gevent.monkey.is_module_patched('threading'): from gevent import get_hub get_hub().threadpool.spawn(_do_send) return except Exception: pass # Not in a gevent-patched environment — call directly _do_send() if collect_garbage: gc.collect() def send_websocket_event(event, success, data): """Acquire a lock to prevent concurrent task execution.""" data_payload = {"success": success, "type": event} if data: # Make a copy to avoid modifying the original data_payload.update(data) # Use the standardized function send_websocket_update('updates', 'update', data_payload) # Help garbage collection by clearing references data_payload = None # Add memory monitoring utilities def get_memory_usage(): """Returns current memory usage in MB""" import psutil process = psutil.Process(os.getpid()) return process.memory_info().rss / (1024 * 1024) def monitor_memory_usage(func): """Decorator to monitor memory usage before and after function execution""" def wrapper(*args, **kwargs): import gc # Force garbage collection before measuring gc.collect() # Get initial memory usage start_mem = get_memory_usage() logger.debug(f"Memory usage before {func.__name__}: {start_mem:.2f} MB") # Call the original function result = func(*args, **kwargs) # Force garbage collection before measuring again gc.collect() # Get final memory usage end_mem = get_memory_usage() logger.debug(f"Memory usage after {func.__name__}: {end_mem:.2f} MB (Change: {end_mem - start_mem:.2f} MB)") return result return wrapper def cleanup_memory(log_usage=False, force_collection=True): """ Comprehensive memory cleanup function to reduce memory footprint Args: log_usage: Whether to log memory usage before and after cleanup force_collection: Whether to force garbage collection """ logger.trace("Starting memory cleanup django memory cleanup") # Skip logging if log level is not set to debug or more verbose (like trace) current_log_level = logger.getEffectiveLevel() if not current_log_level <= logging.DEBUG: log_usage = False if log_usage: try: import psutil process = psutil.Process() before_mem = process.memory_info().rss / (1024 * 1024) logger.debug(f"Memory before cleanup: {before_mem:.2f} MB") except (ImportError, Exception) as e: logger.debug(f"Error getting memory usage: {e}") # Clear any object caches from Django ORM from django.db import connection, reset_queries reset_queries() # Force garbage collection if force_collection: # Run full collection gc.collect(generation=2) # Clear cyclic references gc.collect(generation=0) if log_usage: try: import psutil process = psutil.Process() after_mem = process.memory_info().rss / (1024 * 1024) logger.debug(f"Memory after cleanup: {after_mem:.2f} MB (change: {after_mem-before_mem:.2f} MB)") except (ImportError, Exception): pass logger.trace("Memory cleanup complete for django") def is_protected_path(file_path): """ Determine if a file path is in a protected directory that shouldn't be deleted. Args: file_path (str): The file path to check Returns: bool: True if the path is protected, False otherwise """ if not file_path: return False # List of protected directory prefixes protected_dirs = [ '/data/epgs', # EPG files mapped from host '/data/uploads', # User uploaded files '/data/m3us' # M3U files mapped from host ] # Check if the path starts with any protected directory for protected_dir in protected_dirs: if file_path.startswith(protected_dir): return True return False def validate_flexible_url(value): """ Custom URL validator that accepts URLs with hostnames that aren't FQDNs. This allows URLs like "http://hostname/" which Django's standard URLValidator rejects. """ if not value: return # Allow empty values since the field is nullable # Create a standard Django URL validator url_validator = URLValidator() try: # First try the standard validation url_validator(value) except ValidationError as e: # If standard validation fails, check if it's a non-FQDN hostname import re # More flexible pattern for non-FQDN hostnames with paths # Matches: http://hostname, https://hostname/, http://hostname:port/path/to/file.xml, rtp://192.168.2.1, rtsp://192.168.178.1, udp://239.0.0.1:1234 # Also matches FQDNs for rtsp/rtp/udp protocols: rtsp://FQDN/path?query=value # Also supports authentication: rtsp://user:pass@hostname/path non_fqdn_pattern = r'^(rts?p|https?|udp)://([a-zA-Z0-9_\-\.]+:[^\s@]+@)?([a-zA-Z0-9]([a-zA-Z0-9\-\.]{0,61}[a-zA-Z0-9])?|[0-9.]+)?(\:[0-9]+)?(/[^\s]*)?$' non_fqdn_match = re.match(non_fqdn_pattern, value) if non_fqdn_match: return # Accept non-FQDN hostnames and rtsp/rtp/udp URLs with optional authentication # 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 = dict(details) 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 # remove empty keys for k in list(payload.keys()): if not payload[k]: del payload[k] trigger_event(event_type, payload) except Exception as e: # Don't fail main path if connect dispatch fails pass def log_system_event(event_type, channel_id=None, channel_name=None, **details): """ Log a system event and maintain the configured max history. Args: event_type: Type of event (e.g., 'channel_start', 'client_connect') channel_id: Optional UUID of the channel channel_name: Optional name of the channel **details: Additional details to store in the event (stored as JSON) Example: log_system_event('channel_start', channel_id=uuid, channel_name='CNN', stream_url='http://...', user='admin') """ from core.models import SystemEvent, CoreSettings try: # Create the event SystemEvent.objects.create( event_type=event_type, channel_id=channel_id, channel_name=channel_name, 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 system_settings = CoreSettings.objects.filter(key='system_settings').first() if system_settings and isinstance(system_settings.value, dict): max_events = int(system_settings.value.get('max_system_events', 100)) else: max_events = 100 except Exception: max_events = 100 # Delete old events beyond the limit (keep it efficient with a single query) total_count = SystemEvent.objects.count() if total_count > max_events: # Get the ID of the event at the cutoff point cutoff_event = SystemEvent.objects.values_list('id', flat=True)[max_events] # Delete all events with ID less than cutoff (older events) SystemEvent.objects.filter(id__lt=cutoff_event).delete() except Exception as e: # Don't let event logging break the main application logger.error(f"Failed to log system event {event_type}: {e}") def send_websocket_notification(notification): """ Send a system notification to all connected WebSocket clients. Args: notification: A SystemNotification model instance or dict with notification data Example: from core.models import SystemNotification notification = SystemNotification.create_version_notification('0.19.0', 'https://...') send_websocket_notification(notification) """ try: channel_layer = get_channel_layer() # Convert model instance to dict if needed if hasattr(notification, 'id'): notification_data = { 'id': notification.id, 'notification_key': notification.notification_key, 'notification_type': notification.notification_type, 'priority': notification.priority, 'title': notification.title, 'message': notification.message, 'action_data': notification.action_data, 'is_active': notification.is_active, 'admin_only': notification.admin_only, 'created_at': notification.created_at.isoformat() if notification.created_at else None, } else: notification_data = notification async_to_sync(channel_layer.group_send)( 'updates', { 'type': 'update', 'data': { 'type': 'system_notification', 'notification': notification_data, } } ) logger.debug(f"Sent WebSocket notification: {notification_data.get('title', 'Unknown')}") except Exception as e: logger.error(f"Failed to send WebSocket notification: {e}") def send_notification_dismissed(notification_key): """ Notify all connected clients that a notification was dismissed. Useful for syncing dismissal state across multiple browser tabs/sessions. Args: notification_key: The unique key of the dismissed notification """ try: channel_layer = get_channel_layer() async_to_sync(channel_layer.group_send)( 'updates', { 'type': 'update', 'data': { 'type': 'notification_dismissed', 'notification_key': notification_key, } } ) except Exception as e: logger.error(f"Failed to send notification dismissed event: {e}")