From 50e9075bb50f99bff878266ebf8648da5bb72f51 Mon Sep 17 00:00:00 2001 From: dekzter Date: Sat, 25 Oct 2025 08:15:39 -0400 Subject: [PATCH 001/172] initial run of a binary and encoded redis client - no more encoding / decoding data into redis, huge PITA (still some outstanding spots I need to patch) --- apps/channels/api_views.py | 6 +- apps/channels/tasks.py | 4 +- apps/proxy/tasks.py | 2 +- apps/proxy/ts_proxy/channel_status.py | 208 ++++++------------ apps/proxy/ts_proxy/server.py | 75 +++---- .../ts_proxy/services/channel_service.py | 20 +- apps/proxy/ts_proxy/stream_manager.py | 14 +- apps/proxy/ts_proxy/url_utils.py | 8 +- .../multi_worker_connection_manager.py | 14 +- apps/proxy/vod_proxy/views.py | 16 +- core/redis_pubsub.py | 4 - core/tasks.py | 2 +- core/utils.py | 192 ++++++++-------- dispatcharr/persistent_lock.py | 2 +- 14 files changed, 253 insertions(+), 314 deletions(-) diff --git a/apps/channels/api_views.py b/apps/channels/api_views.py index 862de7f9..aa25ca85 100644 --- a/apps/channels/api_views.py +++ b/apps/channels/api_views.py @@ -1849,14 +1849,12 @@ class RecordingViewSet(viewsets.ModelViewSet): client_set_key = RedisKeys.clients(channel_uuid) client_ids = r.smembers(client_set_key) or [] stopped = 0 - for raw_id in client_ids: + for cid in client_ids: try: - cid = raw_id.decode("utf-8") if isinstance(raw_id, (bytes, bytearray)) else str(raw_id) meta_key = RedisKeys.client_metadata(channel_uuid, cid) ua = r.hget(meta_key, "user_agent") - ua_s = ua.decode("utf-8") if isinstance(ua, (bytes, bytearray)) else (ua or "") # Identify DVR recording client by its user agent - if ua_s and "Dispatcharr-DVR" in ua_s: + if ua and "Dispatcharr-DVR" in ua: try: ChannelService.stop_client(channel_uuid, cid) stopped += 1 diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index 3943cf16..28da31b3 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1875,14 +1875,14 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): md = r.hgetall(metadata_key) if md: def _gv(bkey): - return md.get(bkey.encode('utf-8')) + return md.get(bkey) def _d(bkey, cast=str): v = _gv(bkey) try: if v is None: return None - s = v.decode('utf-8') + s = v return cast(s) if cast is not str else s except Exception: return None diff --git a/apps/proxy/tasks.py b/apps/proxy/tasks.py index 68843712..d42e4d9a 100644 --- a/apps/proxy/tasks.py +++ b/apps/proxy/tasks.py @@ -31,7 +31,7 @@ def fetch_channel_stats(): while True: cursor, keys = redis_client.scan(cursor, match=channel_pattern) for key in keys: - channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key.decode('utf-8')) + channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key) if channel_id_match: ch_id = channel_id_match.group(1) channel_info = ChannelStatus.get_basic_channel_info(ch_id) diff --git a/apps/proxy/ts_proxy/channel_status.py b/apps/proxy/ts_proxy/channel_status.py index 8f1d0649..015d5ca2 100644 --- a/apps/proxy/ts_proxy/channel_status.py +++ b/apps/proxy/ts_proxy/channel_status.py @@ -38,19 +38,19 @@ class ChannelStatus: info = { 'channel_id': channel_id, - 'state': metadata.get(ChannelMetadataField.STATE.encode('utf-8'), b'unknown').decode('utf-8'), - 'url': metadata.get(ChannelMetadataField.URL.encode('utf-8'), b'').decode('utf-8'), - 'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE.encode('utf-8'), b'').decode('utf-8'), - 'started_at': metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0').decode('utf-8'), - 'owner': metadata.get(ChannelMetadataField.OWNER.encode('utf-8'), b'unknown').decode('utf-8'), - 'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0, + 'state': metadata.get(ChannelMetadataField.STATE, 'unknown'), + 'url': metadata.get(ChannelMetadataField.URL, ''), + 'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE, ''), + 'started_at': metadata.get(ChannelMetadataField.INIT_TIME, '0'), + 'owner': metadata.get(ChannelMetadataField.OWNER, 'unknown'), + 'buffer_index': int(buffer_index_value) if buffer_index_value else 0, } # Add stream ID and name information - stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID.encode('utf-8')) + stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID) if stream_id_bytes: try: - stream_id = int(stream_id_bytes.decode('utf-8')) + stream_id = int(stream_id_bytes) info['stream_id'] = stream_id # Look up stream name from database @@ -65,10 +65,10 @@ class ChannelStatus: logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}") # Add M3U profile information - m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE.encode('utf-8')) + m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE) if m3u_profile_id_bytes: try: - m3u_profile_id = int(m3u_profile_id_bytes.decode('utf-8')) + m3u_profile_id = int(m3u_profile_id_bytes) info['m3u_profile_id'] = m3u_profile_id # Look up M3U profile name from database @@ -83,22 +83,22 @@ class ChannelStatus: logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}") # Add timing information - state_changed_field = ChannelMetadataField.STATE_CHANGED_AT.encode('utf-8') + state_changed_field = ChannelMetadataField.STATE_CHANGED_AT if state_changed_field in metadata: - state_changed_at = float(metadata[state_changed_field].decode('utf-8')) + state_changed_at = float(metadata[state_changed_field]) info['state_changed_at'] = state_changed_at info['state_duration'] = time.time() - state_changed_at - init_time_field = ChannelMetadataField.INIT_TIME.encode('utf-8') + init_time_field = ChannelMetadataField.INIT_TIME if init_time_field in metadata: - created_at = float(metadata[init_time_field].decode('utf-8')) + created_at = float(metadata[init_time_field]) info['started_at'] = created_at info['uptime'] = time.time() - created_at # Add data throughput information - total_bytes_field = ChannelMetadataField.TOTAL_BYTES.encode('utf-8') + total_bytes_field = ChannelMetadataField.TOTAL_BYTES if total_bytes_field in metadata: - total_bytes = int(metadata[total_bytes_field].decode('utf-8')) + total_bytes = int(metadata[total_bytes_field]) info['total_bytes'] = total_bytes # Format total bytes in human-readable form @@ -128,40 +128,40 @@ class ChannelStatus: clients = [] for client_id in client_ids: - client_id_str = client_id.decode('utf-8') + client_id_str = client_id client_key = RedisKeys.client_metadata(channel_id, client_id_str) client_data = proxy_server.redis_client.hgetall(client_key) if client_data: client_info = { 'client_id': client_id_str, - 'user_agent': client_data.get(b'user_agent', b'unknown').decode('utf-8'), - 'worker_id': client_data.get(b'worker_id', b'unknown').decode('utf-8'), + 'user_agent': client_data.get('user_agent', 'unknown'), + 'worker_id': client_data.get('worker_id', 'unknown'), } - if b'connected_at' in client_data: - connected_at = float(client_data[b'connected_at'].decode('utf-8')) + if 'connected_at' in client_data: + connected_at = float(client_data['connected_at']) client_info['connected_at'] = connected_at client_info['connection_duration'] = time.time() - connected_at - if b'last_active' in client_data: - last_active = float(client_data[b'last_active'].decode('utf-8')) + if 'last_active' in client_data: + last_active = float(client_data['last_active']) client_info['last_active'] = last_active client_info['last_active_ago'] = time.time() - last_active # Add transfer rate statistics - if b'bytes_sent' in client_data: - client_info['bytes_sent'] = int(client_data[b'bytes_sent'].decode('utf-8')) + if 'bytes_sent' in client_data: + client_info['bytes_sent'] = int(client_data['bytes_sent']) # Add average transfer rate - if b'avg_rate_KBps' in client_data: - client_info['avg_rate_KBps'] = float(client_data[b'avg_rate_KBps'].decode('utf-8')) - elif b'transfer_rate_KBps' in client_data: # For backward compatibility - client_info['avg_rate_KBps'] = float(client_data[b'transfer_rate_KBps'].decode('utf-8')) + if 'avg_rate_KBps' in client_data: + client_info['avg_rate_KBps'] = float(client_data['avg_rate_KBps']) + elif 'transfer_rate_KBps' in client_data: # For backward compatibility + client_info['avg_rate_KBps'] = float(client_data['transfer_rate_KBps']) # Add current transfer rate - if b'current_rate_KBps' in client_data: - client_info['current_rate_KBps'] = float(client_data[b'current_rate_KBps'].decode('utf-8')) + if 'current_rate_KBps' in client_data: + client_info['current_rate_KBps'] = float(client_data['current_rate_KBps']) clients.append(client_info) @@ -235,7 +235,7 @@ class ChannelStatus: while True: cursor, keys = proxy_server.redis_client.scan(cursor, match=buffer_key_pattern, count=100) if keys: - all_buffer_keys.extend([k.decode('utf-8') for k in keys]) + all_buffer_keys.extend([k for k in keys]) if cursor == 0 or len(all_buffer_keys) >= 20: # Limit to 20 keys break @@ -265,61 +265,22 @@ class ChannelStatus: } # Add FFmpeg stream information - video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC.encode('utf-8')) - if video_codec: - info['video_codec'] = video_codec.decode('utf-8') - - resolution = metadata.get(ChannelMetadataField.RESOLUTION.encode('utf-8')) - if resolution: - info['resolution'] = resolution.decode('utf-8') - - source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS.encode('utf-8')) - if source_fps: - info['source_fps'] = float(source_fps.decode('utf-8')) - - pixel_format = metadata.get(ChannelMetadataField.PIXEL_FORMAT.encode('utf-8')) - if pixel_format: - info['pixel_format'] = pixel_format.decode('utf-8') - - source_bitrate = metadata.get(ChannelMetadataField.SOURCE_BITRATE.encode('utf-8')) - if source_bitrate: - info['source_bitrate'] = float(source_bitrate.decode('utf-8')) - - audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC.encode('utf-8')) - if audio_codec: - info['audio_codec'] = audio_codec.decode('utf-8') - - sample_rate = metadata.get(ChannelMetadataField.SAMPLE_RATE.encode('utf-8')) - if sample_rate: - info['sample_rate'] = int(sample_rate.decode('utf-8')) - - audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS.encode('utf-8')) - if audio_channels: - info['audio_channels'] = audio_channels.decode('utf-8') - - audio_bitrate = metadata.get(ChannelMetadataField.AUDIO_BITRATE.encode('utf-8')) - if audio_bitrate: - info['audio_bitrate'] = float(audio_bitrate.decode('utf-8')) + info['video_codec'] = metadata.get(ChannelMetadataField.VIDEO_CODEC) + info['resolution'] = metadata.get(ChannelMetadataField.RESOLUTION) + info['source_fps'] = metadata.get(ChannelMetadataField.SOURCE_FPS) + info['pixel_format'] = metadata.get(ChannelMetadataField.PIXEL_FORMAT) + info['source_bitrate'] = metadata.get(ChannelMetadataField.SOURCE_BITRATE) + info['audio_codec'] = metadata.get(ChannelMetadataField.AUDIO_CODEC) + info['sample_rate'] = metadata.get(ChannelMetadataField.SAMPLE_RATE) + info['audio_channels'] = metadata.get(ChannelMetadataField.AUDIO_CHANNELS) + info['audio_bitrate'] = metadata.get(ChannelMetadataField.AUDIO_BITRATE) # Add FFmpeg performance stats - ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED.encode('utf-8')) - if ffmpeg_speed: - info['ffmpeg_speed'] = float(ffmpeg_speed.decode('utf-8')) - - ffmpeg_fps = metadata.get(ChannelMetadataField.FFMPEG_FPS.encode('utf-8')) - if ffmpeg_fps: - info['ffmpeg_fps'] = float(ffmpeg_fps.decode('utf-8')) - - actual_fps = metadata.get(ChannelMetadataField.ACTUAL_FPS.encode('utf-8')) - if actual_fps: - info['actual_fps'] = float(actual_fps.decode('utf-8')) - - ffmpeg_bitrate = metadata.get(ChannelMetadataField.FFMPEG_BITRATE.encode('utf-8')) - if ffmpeg_bitrate: - info['ffmpeg_bitrate'] = float(ffmpeg_bitrate.decode('utf-8')) - stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE.encode('utf-8')) - if stream_type: - info['stream_type'] = stream_type.decode('utf-8') + info['ffmpeg_speed'] = metadata.get(ChannelMetadataField.FFMPEG_SPEED) + info['ffmpeg_fps'] = metadata.get(ChannelMetadataField.FFMPEG_FPS) + info['actual_fps'] = metadata.get(ChannelMetadataField.ACTUAL_FPS) + info['ffmpeg_bitrate'] = metadata.get(ChannelMetadataField.FFMPEG_BITRATE) + info['stream_type'] = metadata.get(ChannelMetadataField.STREAM_TYPE) return info @@ -364,33 +325,27 @@ class ChannelStatus: client_count = proxy_server.redis_client.scard(client_set_key) or 0 # Calculate uptime - init_time_bytes = metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0') - created_at = float(init_time_bytes.decode('utf-8')) + init_time_bytes = metadata.get(ChannelMetadataField.INIT_TIME, '0') + created_at = float(init_time_bytes) uptime = time.time() - created_at if created_at > 0 else 0 - # Safely decode bytes or use defaults - def safe_decode(bytes_value, default="unknown"): - if bytes_value is None: - return default - return bytes_value.decode('utf-8') - # Simplified info info = { 'channel_id': channel_id, - 'state': safe_decode(metadata.get(ChannelMetadataField.STATE.encode('utf-8'))), - 'url': safe_decode(metadata.get(ChannelMetadataField.URL.encode('utf-8')), ""), - 'stream_profile': safe_decode(metadata.get(ChannelMetadataField.STREAM_PROFILE.encode('utf-8')), ""), - 'owner': safe_decode(metadata.get(ChannelMetadataField.OWNER.encode('utf-8'))), - 'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0, + 'state': metadata.get(ChannelMetadataField.STATE), + 'url': metadata.get(ChannelMetadataField.URL, ""), + 'stream_profile': metadata.get(ChannelMetadataField.STREAM_PROFILE, ""), + 'owner': metadata.get(ChannelMetadataField.OWNER), + 'buffer_index': int(buffer_index_value) if buffer_index_value else 0, 'client_count': client_count, 'uptime': uptime } # Add stream ID and name information - stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID.encode('utf-8')) + stream_id_bytes = metadata.get(ChannelMetadataField.STREAM_ID) if stream_id_bytes: try: - stream_id = int(stream_id_bytes.decode('utf-8')) + stream_id = int(stream_id_bytes) info['stream_id'] = stream_id # Look up stream name from database @@ -405,9 +360,9 @@ class ChannelStatus: logger.warning(f"Invalid stream_id format in Redis: {stream_id_bytes}") # Add data throughput information to basic info - total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.TOTAL_BYTES.encode('utf-8')) + total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.TOTAL_BYTES) if total_bytes_bytes: - total_bytes = int(total_bytes_bytes.decode('utf-8')) + total_bytes = int(total_bytes_bytes) info['total_bytes'] = total_bytes # Calculate and add bitrate @@ -434,26 +389,25 @@ class ChannelStatus: if client_ids: # Get up to 10 clients for the basic view for client_id in list(client_ids)[:10]: - client_id_str = client_id.decode('utf-8') - client_key = RedisKeys.client_metadata(channel_id, client_id_str) + client_key = RedisKeys.client_metadata(channel_id, client_id) # Efficient way - just retrieve the essentials client_info = { - 'client_id': client_id_str, + 'client_id': client_id, } # Safely get user_agent and ip_address user_agent_bytes = proxy_server.redis_client.hget(client_key, 'user_agent') - client_info['user_agent'] = safe_decode(user_agent_bytes) + client_info['user_agent'] = user_agent_bytes ip_address_bytes = proxy_server.redis_client.hget(client_key, 'ip_address') if ip_address_bytes: - client_info['ip_address'] = safe_decode(ip_address_bytes) + client_info['ip_address'] = ip_address_bytes # Just get connected_at for client age connected_at_bytes = proxy_server.redis_client.hget(client_key, 'connected_at') if connected_at_bytes: - connected_at = float(connected_at_bytes.decode('utf-8')) + connected_at = float(connected_at_bytes) client_info['connected_since'] = time.time() - connected_at clients.append(client_info) @@ -462,10 +416,10 @@ class ChannelStatus: info['clients'] = clients # Add M3U profile information - m3u_profile_id_bytes = metadata.get(ChannelMetadataField.M3U_PROFILE.encode('utf-8')) - if m3u_profile_id_bytes: + m3u_profile_id = metadata.get(ChannelMetadataField.M3U_PROFILE) + if m3u_profile_id: try: - m3u_profile_id = int(m3u_profile_id_bytes.decode('utf-8')) + m3u_profile_id = int(m3u_profile_id) info['m3u_profile_id'] = m3u_profile_id # Look up M3U profile name from database @@ -477,32 +431,16 @@ class ChannelStatus: except (ImportError, DatabaseError) as e: logger.warning(f"Failed to get M3U profile name for ID {m3u_profile_id}: {e}") except ValueError: - logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id_bytes}") + logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id}") # Add stream info to basic info as well - video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC.encode('utf-8')) - if video_codec: - info['video_codec'] = video_codec.decode('utf-8') - - resolution = metadata.get(ChannelMetadataField.RESOLUTION.encode('utf-8')) - if resolution: - info['resolution'] = resolution.decode('utf-8') - - source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS.encode('utf-8')) - if source_fps: - info['source_fps'] = float(source_fps.decode('utf-8')) - ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED.encode('utf-8')) - if ffmpeg_speed: - info['ffmpeg_speed'] = float(ffmpeg_speed.decode('utf-8')) - audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC.encode('utf-8')) - if audio_codec: - info['audio_codec'] = audio_codec.decode('utf-8') - audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS.encode('utf-8')) - if audio_channels: - info['audio_channels'] = audio_channels.decode('utf-8') - stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE.encode('utf-8')) - if stream_type: - info['stream_type'] = stream_type.decode('utf-8') + info['video_codec'] = metadata.get(ChannelMetadataField.VIDEO_CODEC) + info['resolution'] = metadata.get(ChannelMetadataField.RESOLUTION) + info['source_fps'] = metadata.get(ChannelMetadataField.SOURCE_FPS) + info['ffmpeg_speed'] = metadata.get(ChannelMetadataField.FFMPEG_SPEED) + info['audio_codec'] = metadata.get(ChannelMetadataField.AUDIO_CODEC) + info['audio_channels'] = metadata.get(ChannelMetadataField.AUDIO_CHANNELS) + info['stream_type'] = metadata.get(ChannelMetadataField.STREAM_TYPE) return info except Exception as e: diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index cca827a9..7a59f530 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -157,7 +157,8 @@ class ProxyServer: socket_timeout=60, socket_connect_timeout=10, socket_keepalive=True, - health_check_interval=30 + health_check_interval=30, + decode_responses=True ) logger.info("Created fallback Redis PubSub client for event listener") @@ -178,8 +179,8 @@ class ProxyServer: continue try: - channel = message["channel"].decode("utf-8") - data = json.loads(message["data"].decode("utf-8")) + channel = message["channel"] + data = json.loads(message["data"]) event_type = data.get("event") channel_id = data.get("channel_id") @@ -374,7 +375,7 @@ class ProxyServer: try: lock_key = RedisKeys.channel_owner(channel_id) return self._execute_redis_command( - lambda: self.redis_client.get(lock_key).decode('utf-8') if self.redis_client.get(lock_key) else None + lambda: self.redis_client.get(lock_key) if self.redis_client.get(lock_key) else None ) except Exception as e: logger.error(f"Error getting channel owner: {e}") @@ -415,7 +416,7 @@ class ProxyServer: current_owner = self._execute_redis_command( lambda: self.redis_client.get(lock_key) ) - if current_owner and current_owner.decode('utf-8') == self.worker_id: + if current_owner and current_owner == self.worker_id: # Refresh TTL self._execute_redis_command( lambda: self.redis_client.expire(lock_key, ttl) @@ -440,7 +441,7 @@ class ProxyServer: # Only delete if we're the current owner to prevent race conditions current = self.redis_client.get(lock_key) - if current and current.decode('utf-8') == self.worker_id: + if current and current == self.worker_id: self.redis_client.delete(lock_key) logger.info(f"Released ownership of channel {channel_id}") @@ -462,7 +463,7 @@ class ProxyServer: current = self.redis_client.get(lock_key) # Only extend if we're still the owner - if current and current.decode('utf-8') == self.worker_id: + if current and current == self.worker_id: self.redis_client.expire(lock_key, ttl) return True return False @@ -478,15 +479,15 @@ class ProxyServer: metadata_key = RedisKeys.channel_metadata(channel_id) if self.redis_client.exists(metadata_key): metadata = self.redis_client.hgetall(metadata_key) - if b'state' in metadata: - state = metadata[b'state'].decode('utf-8') + if 'state' in metadata: + state = metadata['state'] active_states = [ChannelState.INITIALIZING, ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS, ChannelState.ACTIVE, ChannelState.BUFFERING] if state in active_states: logger.info(f"Channel {channel_id} already being initialized with state {state}") # Create buffer and client manager only if we don't have them if channel_id not in self.stream_buffers: - self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=self.redis_client) + self.stream_buffers[channel_id] = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer()) if channel_id not in self.client_managers: self.client_managers[channel_id] = ClientManager( channel_id, @@ -497,7 +498,7 @@ class ProxyServer: # Create buffer and client manager instances (or reuse if they exist) if channel_id not in self.stream_buffers: - buffer = StreamBuffer(channel_id, redis_client=self.redis_client) + buffer = StreamBuffer(channel_id, redis_client=RedisClient.get_buffer()) self.stream_buffers[channel_id] = buffer if channel_id not in self.client_managers: @@ -536,18 +537,18 @@ class ProxyServer: # If no url was passed, try to get from Redis if not url and existing_metadata: - url_bytes = existing_metadata.get(b'url') + url_bytes = existing_metadata.get('url') if url_bytes: - channel_url = url_bytes.decode('utf-8') + channel_url = url_bytes - ua_bytes = existing_metadata.get(b'user_agent') + ua_bytes = existing_metadata.get('user_agent') if ua_bytes: - channel_user_agent = ua_bytes.decode('utf-8') + channel_user_agent = ua_bytes # Get stream ID from metadata if not provided - if not channel_stream_id and b'stream_id' in existing_metadata: + if not channel_stream_id and 'stream_id' in existing_metadata: try: - channel_stream_id = int(existing_metadata[b'stream_id'].decode('utf-8')) + channel_stream_id = int(existing_metadata['stream_id']) logger.debug(f"Found stream_id {channel_stream_id} in metadata for channel {channel_id}") except (ValueError, TypeError) as e: logger.debug(f"Could not parse stream_id from metadata: {e}") @@ -562,7 +563,7 @@ class ProxyServer: # Create buffer but not stream manager (only if not already exists) if channel_id not in self.stream_buffers: - buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) + buffer = StreamBuffer(channel_id=channel_id, redis_client=RedisClient.get_buffer()) self.stream_buffers[channel_id] = buffer # Create client manager with channel_id and redis_client (only if not already exists) @@ -585,7 +586,7 @@ class ProxyServer: # Create buffer but not stream manager (only if not already exists) if channel_id not in self.stream_buffers: - buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) + buffer = StreamBuffer(channel_id=channel_id, redis_client=RedisClient.get_buffer()) self.stream_buffers[channel_id] = buffer # Create client manager with channel_id and redis_client (only if not already exists) @@ -624,12 +625,12 @@ class ProxyServer: # Verify the stream_id was set correctly in Redis stream_id_value = self.redis_client.hget(metadata_key, "stream_id") if stream_id_value: - logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is set in Redis for channel {channel_id}") + logger.info(f"Verified stream_id {stream_id_value} is set in Redis for channel {channel_id}") else: logger.warning(f"Failed to set stream_id in Redis for channel {channel_id}") # Create stream buffer - buffer = StreamBuffer(channel_id=channel_id, redis_client=self.redis_client) + buffer = StreamBuffer(channel_id=channel_id, redis_client=RedisClient.get_buffer()) logger.debug(f"Created StreamBuffer for channel {channel_id}") self.stream_buffers[channel_id] = buffer @@ -700,8 +701,8 @@ class ProxyServer: metadata = self.redis_client.hgetall(metadata_key) # Get channel state and owner - state = metadata.get(b'state', b'unknown').decode('utf-8') - owner = metadata.get(b'owner', b'').decode('utf-8') + state = metadata.get('state', 'unknown') + owner = metadata.get('owner', '') # States that indicate the channel is running properly valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, @@ -726,8 +727,8 @@ class ProxyServer: return True else: # Unknown or initializing state, check how long it's been in this state - if b'state_changed_at' in metadata: - state_changed_at = float(metadata[b'state_changed_at'].decode('utf-8')) + if 'state_changed_at' in metadata: + state_changed_at = float(metadata['state_changed_at']) state_age = time.time() - state_changed_at # If in initializing state for too long, consider it stale @@ -762,8 +763,8 @@ class ProxyServer: # If we have metadata, log details for debugging if metadata: - state = metadata.get(b'state', b'unknown').decode('utf-8') - owner = metadata.get(b'owner', b'unknown').decode('utf-8') + state = metadata.get('state', 'unknown') + owner = metadata.get('owner', 'unknown') logger.info(f"Zombie channel details - state: {state}, owner: {owner}") # Clean up Redis keys @@ -931,8 +932,8 @@ class ProxyServer: if self.redis_client: metadata_key = RedisKeys.channel_metadata(channel_id) metadata = self.redis_client.hgetall(metadata_key) - if metadata and b'state' in metadata: - channel_state = metadata[b'state'].decode('utf-8') + if metadata and 'state' in metadata: + channel_state = metadata['state'] # Check if channel has any clients left total_clients = 0 @@ -948,9 +949,9 @@ class ProxyServer: if channel_state in [ChannelState.CONNECTING, ChannelState.WAITING_FOR_CLIENTS]: # Get connection ready time from metadata connection_ready_time = None - if metadata and b'connection_ready_time' in metadata: + if metadata and 'connection_ready_time' in metadata: try: - connection_ready_time = float(metadata[b'connection_ready_time'].decode('utf-8')) + connection_ready_time = float(metadata['connection_ready_time']) except (ValueError, TypeError): pass @@ -981,8 +982,8 @@ class ProxyServer: # Grace period expired but we have clients - mark channel as active logger.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active") old_state = "unknown" - if metadata and b'state' in metadata: - old_state = metadata[b'state'].decode('utf-8') + if metadata and 'state' in metadata: + old_state = metadata['state'] if self.update_channel_state(channel_id, ChannelState.ACTIVE, { "grace_period_ended_at": str(time.time()), "clients_at_activation": str(total_clients) @@ -998,7 +999,7 @@ class ProxyServer: disconnect_value = self.redis_client.get(disconnect_key) if disconnect_value: try: - disconnect_time = float(disconnect_value.decode('utf-8')) + disconnect_time = float(disconnect_value) except (ValueError, TypeError) as e: logger.error(f"Invalid disconnect time for channel {channel_id}: {e}") @@ -1076,7 +1077,7 @@ class ProxyServer: for key in channel_keys: try: - channel_id = key.decode('utf-8').split(':')[2] + channel_id = key.split(':')[2] # Skip channels we already have locally if channel_id in self.stream_buffers: @@ -1170,8 +1171,8 @@ class ProxyServer: # Get current state for logging current_state = None metadata = self.redis_client.hgetall(metadata_key) - if metadata and b'state' in metadata: - current_state = metadata[b'state'].decode('utf-8') + if metadata and 'state' in metadata: + current_state = metadata['state'] # Only update if state is actually changing if current_state == new_state: diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 551e2d27..6e7d3878 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -59,7 +59,7 @@ class ChannelService: # Verify the stream_id was set stream_id_value = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID) if stream_id_value: - logger.debug(f"Verified stream_id {stream_id_value.decode('utf-8')} is now set in Redis") + logger.debug(f"Verified stream_id {stream_id_value} is now set in Redis") else: logger.error(f"Failed to set stream_id {stream_id} in Redis before initialization") @@ -129,7 +129,7 @@ class ChannelService: try: # This is inefficient but used for diagnostics - in production would use more targeted checks redis_keys = proxy_server.redis_client.keys(f"ts_proxy:*:{channel_id}*") - redis_keys = [k.decode('utf-8') for k in redis_keys] if redis_keys else [] + redis_keys = [k for k in redis_keys] if redis_keys else [] except Exception as e: logger.error(f"Error checking Redis keys: {e}") @@ -234,8 +234,8 @@ class ChannelService: metadata_key = RedisKeys.channel_metadata(channel_id) try: metadata = proxy_server.redis_client.hgetall(metadata_key) - if metadata and b'state' in metadata: - state = metadata[b'state'].decode('utf-8') + if metadata and 'state' in metadata: + state = metadata['state'] channel_info = {"state": state} # Immediately mark as stopping in metadata so clients detect it faster @@ -375,8 +375,8 @@ class ChannelService: metadata = proxy_server.redis_client.hgetall(metadata_key) # Extract state and owner - state = metadata.get(ChannelMetadataField.STATE.encode(), b'unknown').decode('utf-8') - owner = metadata.get(ChannelMetadataField.OWNER.encode(), b'unknown').decode('utf-8') + state = metadata.get(ChannelMetadataField.STATE.encode(), 'unknown') + owner = metadata.get(ChannelMetadataField.OWNER.encode(), 'unknown') # Valid states indicate channel is running properly valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING] @@ -402,7 +402,7 @@ class ChannelService: } if last_data: - last_data_time = float(last_data.decode('utf-8')) + last_data_time = float(last_data) data_age = time.time() - last_data_time details["last_data_age"] = data_age @@ -598,7 +598,7 @@ class ChannelService: def _update_stream_stats_in_db(stream_id, **stats): """Update stream stats in database""" from django.db import connection - + try: from apps.channels.models import Stream from django.utils import timezone @@ -624,7 +624,7 @@ class ChannelService: except Exception as e: logger.error(f"Error updating stream stats in database for stream {stream_id}: {e}") return False - + finally: # Always close database connection after update try: @@ -645,7 +645,7 @@ class ChannelService: metadata_key = RedisKeys.channel_metadata(channel_id) # First check if the key exists and what type it is - key_type = proxy_server.redis_client.type(metadata_key).decode('utf-8') + key_type = proxy_server.redis_client.type(metadata_key) logger.debug(f"Redis key {metadata_key} is of type: {key_type}") # Build metadata update dict diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index 99ae8027..73118768 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -89,7 +89,7 @@ class StreamManager: # Try to get stream_id specifically stream_id_bytes = buffer.redis_client.hget(metadata_key, "stream_id") if stream_id_bytes: - self.current_stream_id = int(stream_id_bytes.decode('utf-8')) + self.current_stream_id = int(stream_id_bytes) self.tried_stream_ids.add(self.current_stream_id) logger.info(f"Loaded stream ID {self.current_stream_id} from Redis for channel {buffer.channel_id}") else: @@ -362,7 +362,7 @@ class StreamManager: current_owner = self.buffer.redis_client.get(owner_key) # Use the worker_id that was passed in during initialization - if current_owner and self.worker_id and current_owner.decode('utf-8') == self.worker_id: + if current_owner and self.worker_id and current_owner == self.worker_id: # Determine the appropriate error message based on retry failures if self.tried_stream_ids and len(self.tried_stream_ids) > 0: error_message = f"All {len(self.tried_stream_ids)} stream options failed" @@ -948,10 +948,10 @@ class StreamManager: logger.debug(f"Updated m3u profile for channel {self.channel_id} to use profile from stream {stream_id}") else: logger.warning(f"Failed to update stream profile for channel {self.channel_id}") - + except Exception as e: logger.error(f"Error updating stream profile for channel {self.channel_id}: {e}") - + finally: # Always close database connection after profile update try: @@ -1348,9 +1348,9 @@ class StreamManager: current_state = None try: metadata = redis_client.hgetall(metadata_key) - state_field = ChannelMetadataField.STATE.encode('utf-8') + state_field = ChannelMetadataField.STATE if metadata and state_field in metadata: - current_state = metadata[state_field].decode('utf-8') + current_state = metadata[state_field] except Exception as e: logger.error(f"Error checking current state: {e}") @@ -1555,4 +1555,4 @@ class StreamManager: """Safely reset the URL switching state if it gets stuck""" self.url_switching = False self.url_switch_start_time = 0 - logger.info(f"Reset URL switching state for channel {self.channel_id}") \ No newline at end of file + logger.info(f"Reset URL switching state for channel {self.channel_id}") diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index db53cc74..0bfde312 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -215,9 +215,9 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] existing_stream_id = redis_client.get(f"channel_stream:{channel.id}") if existing_stream_id: # Decode bytes to string/int for proper Redis key lookup - existing_stream_id = existing_stream_id.decode('utf-8') + existing_stream_id = existing_stream_id existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}") - if existing_profile_id and int(existing_profile_id.decode('utf-8')) == profile.id: + if existing_profile_id and int(existing_profile_id) == profile.id: channel_using_profile = True logger.debug(f"Channel {channel.id} already using profile {profile.id}") @@ -353,9 +353,9 @@ def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = No existing_stream_id = redis_client.get(f"channel_stream:{channel.id}") if existing_stream_id: # Decode bytes to string/int for proper Redis key lookup - existing_stream_id = existing_stream_id.decode('utf-8') + existing_stream_id = existing_stream_id existing_profile_id = redis_client.get(f"stream_profile:{existing_stream_id}") - if existing_profile_id and int(existing_profile_id.decode('utf-8')) == profile.id: + if existing_profile_id and int(existing_profile_id) == profile.id: channel_using_profile = True logger.debug(f"Channel {channel.id} already using profile {profile.id}") diff --git a/apps/proxy/vod_proxy/multi_worker_connection_manager.py b/apps/proxy/vod_proxy/multi_worker_connection_manager.py index fefc8739..e1fe4c4b 100644 --- a/apps/proxy/vod_proxy/multi_worker_connection_manager.py +++ b/apps/proxy/vod_proxy/multi_worker_connection_manager.py @@ -219,7 +219,7 @@ class RedisBackedVODConnection: # Convert bytes keys/values to strings if needed if isinstance(list(data.keys())[0], bytes): - data = {k.decode('utf-8'): v.decode('utf-8') for k, v in data.items()} + data = {k: v for k, v in data.items()} return SerializableConnectionState.from_dict(data) except Exception as e: @@ -1115,14 +1115,14 @@ class MultiWorkerVODConnectionManager: # Convert bytes to strings if needed if isinstance(list(data.keys())[0], bytes): - data = {k.decode('utf-8'): v.decode('utf-8') for k, v in data.items()} + data = {k: v for k, v in data.items()} last_activity = float(data.get('last_activity', 0)) active_streams = int(data.get('active_streams', 0)) # Clean up if stale and no active streams if (current_time - last_activity > max_age_seconds) and active_streams == 0: - session_id = key.decode('utf-8').replace('vod_persistent_connection:', '') + session_id = key.replace('vod_persistent_connection:', '') logger.info(f"Cleaning up stale connection: {session_id}") # Clean up connection and related keys @@ -1219,7 +1219,7 @@ class MultiWorkerVODConnectionManager: if connection_data: # Convert bytes to strings if needed if isinstance(list(connection_data.keys())[0], bytes): - connection_data = {k.decode('utf-8'): v.decode('utf-8') for k, v in connection_data.items()} + connection_data = {k: v for k, v in connection_data.items()} profile_id = connection_data.get('m3u_profile_id') if profile_id: @@ -1279,7 +1279,7 @@ class MultiWorkerVODConnectionManager: # Convert bytes keys/values to strings if needed if isinstance(list(connection_data.keys())[0], bytes): - connection_data = {k.decode('utf-8'): v.decode('utf-8') for k, v in connection_data.items()} + connection_data = {k: v for k, v in connection_data.items()} # Check if content matches (using consolidated data) stored_content_type = connection_data.get('content_obj_type', '') @@ -1289,7 +1289,7 @@ class MultiWorkerVODConnectionManager: continue # Extract session ID - session_id = key.decode('utf-8').replace('vod_persistent_connection:', '') + session_id = key.replace('vod_persistent_connection:', '') # Check if Redis-backed connection exists and has no active streams redis_connection = RedisBackedVODConnection(session_id, self.redis_client) @@ -1367,4 +1367,4 @@ class MultiWorkerVODConnectionManager: return redis_connection.get_session_metadata() except Exception as e: logger.error(f"Error getting session info for {session_id}: {e}") - return None \ No newline at end of file + return None diff --git a/apps/proxy/vod_proxy/views.py b/apps/proxy/vod_proxy/views.py index 00ed8a10..12e1d071 100644 --- a/apps/proxy/vod_proxy/views.py +++ b/apps/proxy/vod_proxy/views.py @@ -550,14 +550,7 @@ class VODStreamView(View): connection_data = redis_client.hgetall(persistent_connection_key) if connection_data: - # Decode Redis hash data - decoded_data = {} - for k, v in connection_data.items(): - k_str = k.decode('utf-8') if isinstance(k, bytes) else k - v_str = v.decode('utf-8') if isinstance(v, bytes) else v - decoded_data[k_str] = v_str - - existing_profile_id = decoded_data.get('m3u_profile_id') + existing_profile_id = connection_data.get('m3u_profile_id') if existing_profile_id: try: existing_profile = M3UAccountProfile.objects.get( @@ -770,19 +763,16 @@ class VODStatsView(View): for key in keys: try: - key_str = key.decode('utf-8') if isinstance(key, bytes) else key connection_data = redis_client.hgetall(key) if connection_data: # Extract session ID from key - session_id = key_str.replace('vod_persistent_connection:', '') + session_id = key.replace('vod_persistent_connection:', '') # Decode Redis hash data combined_data = {} for k, v in connection_data.items(): - k_str = k.decode('utf-8') if isinstance(k, bytes) else k - v_str = v.decode('utf-8') if isinstance(v, bytes) else v - combined_data[k_str] = v_str + combined_data[k] = v # Get content info from the connection data (using correct field names) content_type = combined_data.get('content_obj_type', 'unknown') diff --git a/core/redis_pubsub.py b/core/redis_pubsub.py index 5d0032b0..b1f48b1f 100644 --- a/core/redis_pubsub.py +++ b/core/redis_pubsub.py @@ -201,10 +201,6 @@ class RedisPubSubManager: channel = message.get('channel') if channel: - # Decode binary channel name if needed - if isinstance(channel, bytes): - channel = channel.decode('utf-8') - # Find and call the appropriate handler handler = self.message_handlers.get(channel) if handler: diff --git a/core/tasks.py b/core/tasks.py index f757613b..257d8660 100644 --- a/core/tasks.py +++ b/core/tasks.py @@ -404,7 +404,7 @@ def fetch_channel_stats(): while True: cursor, keys = redis_client.scan(cursor, match=channel_pattern) for key in keys: - channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key.decode('utf-8')) + channel_id_match = re.search(r"ts_proxy:channel:(.*):metadata", key) if channel_id_match: ch_id = channel_id_match.group(1) channel_info = ChannelStatus.get_basic_channel_info(ch_id) diff --git a/core/utils.py b/core/utils.py index 36ac5fef..9cc4c26f 100644 --- a/core/utils.py +++ b/core/utils.py @@ -42,101 +42,116 @@ def natural_sort_key(text): return [convert(c) for c in re.split('([0-9]+)', text)] class RedisClient: + _initialized = False _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))) + + # 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) + + # Create Redis client with better defaults + client = redis.Redis( + host=redis_host, + port=redis_port, + db=redis_db, + 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 + ) + + # Validate connection with ping + client.ping() + if cls._initialized is False: + client.flushdb() + cls._initialized = True + + # 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 + + # Set optimal memory settings with environment variable support + # Get max memory from environment or use a larger default (512MB instead of 256MB) + #max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb') + #eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru') + + # Apply memory settings + #client.config_set('maxmemory-policy', eviction_policy) + #client.config_set('maxmemory', max_memory) + + #logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}") + + # 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}") + break + + except (ConnectionError, TimeoutError) as e: + retry_count += 1 + if retry_count >= max_retries: + logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}") + 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: + logger.error(f"Unexpected error connecting to Redis: {e}") + return None + + return client + @classmethod def get_client(cls, max_retries=5, retry_interval=1): if cls._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))) - - # 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) - - # Create Redis client with better defaults - client = redis.Redis( - host=redis_host, - port=redis_port, - db=redis_db, - 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 - ) - - # Validate connection with ping - client.ping() - client.flushdb() - - # 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 - - # Set optimal memory settings with environment variable support - # Get max memory from environment or use a larger default (512MB instead of 256MB) - #max_memory = os.environ.get('REDIS_MAX_MEMORY', '512mb') - #eviction_policy = os.environ.get('REDIS_EVICTION_POLICY', 'allkeys-lru') - - # Apply memory settings - #client.config_set('maxmemory-policy', eviction_policy) - #client.config_set('maxmemory', max_memory) - - #logger.info(f"Redis configured with maxmemory={max_memory}, policy={eviction_policy}") - - # 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}") - - cls._client = client - break - - except (ConnectionError, TimeoutError) as e: - retry_count += 1 - if retry_count >= max_retries: - logger.error(f"Failed to connect to Redis after {max_retries} attempts: {e}") - 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: - logger.error(f"Unexpected error connecting to Redis: {e}") - return 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""" @@ -165,7 +180,8 @@ class RedisClient: socket_connect_timeout=socket_connect_timeout, socket_keepalive=socket_keepalive, health_check_interval=health_check_interval, - retry_on_timeout=retry_on_timeout + retry_on_timeout=retry_on_timeout, + decode_responses=True ) # Validate connection with ping diff --git a/dispatcharr/persistent_lock.py b/dispatcharr/persistent_lock.py index 360c9b5d..fe62be43 100644 --- a/dispatcharr/persistent_lock.py +++ b/dispatcharr/persistent_lock.py @@ -48,7 +48,7 @@ class PersistentLock: Returns True if the expiration was successfully extended. """ current_value = self.redis_client.get(self.lock_key) - if current_value and current_value.decode("utf-8") == self.lock_token: + if current_value and current_value == self.lock_token: self.redis_client.expire(self.lock_key, self.lock_timeout) self.has_lock = False return True From 16183329623e163ec01145b721e2197571c12097 Mon Sep 17 00:00:00 2001 From: dekzter Date: Wed, 12 Nov 2025 17:52:11 -0500 Subject: [PATCH 002/172] byte to string fix --- apps/proxy/vod_proxy/connection_manager.py | 55 ++++++++++------------ 1 file changed, 26 insertions(+), 29 deletions(-) diff --git a/apps/proxy/vod_proxy/connection_manager.py b/apps/proxy/vod_proxy/connection_manager.py index dea5759b..00ca77aa 100644 --- a/apps/proxy/vod_proxy/connection_manager.py +++ b/apps/proxy/vod_proxy/connection_manager.py @@ -329,15 +329,15 @@ class VODConnectionManager: continue # Extract session info - stored_content_type = session_data.get(b'content_type', b'').decode('utf-8') - stored_content_uuid = session_data.get(b'content_uuid', b'').decode('utf-8') + stored_content_type = session_data.get('content_type', '') + stored_content_uuid = session_data.get('content_uuid', '') # Check if content matches if stored_content_type != content_type or stored_content_uuid != content_uuid: continue # Extract session ID from key - session_id = key.decode('utf-8').replace('vod_session:', '') + session_id = key.replace('vod_session:', '') # Check if session has an active persistent connection persistent_conn = self._persistent_connections.get(session_id) @@ -351,13 +351,13 @@ class VODConnectionManager: continue # Get stored client info for comparison - stored_client_ip = session_data.get(b'client_ip', b'').decode('utf-8') - stored_user_agent = session_data.get(b'user_agent', b'').decode('utf-8') + stored_client_ip = session_data.get('client_ip', '') + stored_user_agent = session_data.get('user_agent', '') # Check timeshift parameters match - stored_utc_start = session_data.get(b'utc_start', b'').decode('utf-8') - stored_utc_end = session_data.get(b'utc_end', b'').decode('utf-8') - stored_offset = session_data.get(b'offset', b'').decode('utf-8') + stored_utc_start = session_data.get('utc_start', '') + stored_utc_end = session_data.get('utc_end', '') + stored_offset = session_data.get('offset', '') current_utc_start = utc_start or "" current_utc_end = utc_end or "" @@ -394,7 +394,7 @@ class VODConnectionManager: 'session_id': session_id, 'score': score, 'reasons': match_reasons, - 'last_activity': float(session_data.get(b'last_activity', b'0').decode('utf-8')) + 'last_activity': float(session_data.get('last_activity', '0')) }) except Exception as e: @@ -536,7 +536,7 @@ class VODConnectionManager: # Get current bytes and add to it current_bytes = self.redis_client.hget(connection_key, "bytes_sent") if current_bytes: - total_bytes = int(current_bytes.decode('utf-8')) + bytes_sent + total_bytes = int(current_bytes) + bytes_sent else: total_bytes = bytes_sent update_data["bytes_sent"] = str(total_bytes) @@ -571,7 +571,7 @@ class VODConnectionManager: profile_id = None if b"m3u_profile_id" in connection_data: try: - profile_id = int(connection_data[b"m3u_profile_id"].decode('utf-8')) + profile_id = int(connection_data["m3u_profile_id"]) except ValueError: pass @@ -617,16 +617,14 @@ class VODConnectionManager: # Convert bytes to strings and parse numbers info = {} for key, value in connection_data.items(): - key_str = key.decode('utf-8') - value_str = value.decode('utf-8') # Parse numeric fields - if key_str in ['connected_at', 'last_activity']: - info[key_str] = float(value_str) - elif key_str in ['bytes_sent', 'position_seconds', 'm3u_profile_id']: - info[key_str] = int(value_str) + if key in ['connected_at', 'last_activity']: + info[key] = float(value) + elif key in ['bytes_sent', 'position_seconds', 'm3u_profile_id']: + info[key] = int(valuvaluee_str) else: - info[key_str] = value_str + info[key] = value return info @@ -676,14 +674,13 @@ class VODConnectionManager: for key in keys: try: - key_str = key.decode('utf-8') last_activity = self.redis_client.hget(key, "last_activity") if last_activity: - last_activity_time = float(last_activity.decode('utf-8')) + last_activity_time = float(last_activity) if current_time - last_activity_time > max_age_seconds: # Extract info for cleanup - parts = key_str.split(':') + parts = key.split(':') if len(parts) >= 5: content_type = parts[2] content_uuid = parts[3] @@ -1342,9 +1339,9 @@ class VODConnectionManager: session_data = self.redis_client.hgetall(session_key) if session_data: # Get session details for connection cleanup - content_type = session_data.get(b'content_type', b'').decode('utf-8') - content_uuid = session_data.get(b'content_uuid', b'').decode('utf-8') - profile_id = session_data.get(b'profile_id') + content_type = session_data.get('content_type', '') + content_uuid = session_data.get('content_uuid', '') + profile_id = session_data.get('profile_id') # Generate client_id from session_id (matches what's used during streaming) client_id = session_id @@ -1355,12 +1352,12 @@ class VODConnectionManager: self.remove_connection(content_type, content_uuid, client_id) # Remove from profile connections if counted (additional safety check) - if session_data.get(b'connection_counted') == b'True' and profile_id: - profile_key = self._get_profile_connections_key(int(profile_id.decode('utf-8'))) + if session_data.get('connection_counted') == 'True' and profile_id: + profile_key = self._get_profile_connections_key(int(profile_id)) current_count = int(self.redis_client.get(profile_key) or 0) if current_count > 0: self.redis_client.decr(profile_key) - logger.info(f"[{session_id}] Decremented profile {profile_id.decode('utf-8')} connections") + logger.info(f"[{session_id}] Decremented profile {profile_id} connections") # Remove session tracking key self.redis_client.delete(session_key) @@ -1380,7 +1377,7 @@ class VODConnectionManager: if session_related_keys: # Filter out keys we already deleted - remaining_keys = [k for k in session_related_keys if k.decode('utf-8') != session_key] + remaining_keys = [k for k in session_related_keys if k != session_key] if remaining_keys: self.redis_client.delete(*remaining_keys) logger.info(f"[{session_id}] Cleaned up {len(remaining_keys)} additional session-related keys") @@ -1410,7 +1407,7 @@ class VODConnectionManager: if self.redis_client: session_data = self.redis_client.hgetall(session_key) if session_data: - created_at = float(session_data.get(b'created_at', b'0').decode('utf-8')) + created_at = float(session_data.get('created_at', '0')) if current_time - created_at > max_age_seconds: logger.info(f"[{session_id}] Session older than {max_age_seconds}s") stale_sessions.append(session_id) From cd35e671e56a29e57e256351010147a4c8145c88 Mon Sep 17 00:00:00 2001 From: dekzter Date: Wed, 12 Nov 2025 17:52:48 -0500 Subject: [PATCH 003/172] byte to string fix --- apps/proxy/ts_proxy/stream_generator.py | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index 368691b8..25122d74 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -108,13 +108,13 @@ class StreamGenerator: metadata_key = RedisKeys.channel_metadata(self.channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) - if metadata and b'state' in metadata: - state = metadata[b'state'].decode('utf-8') + if metadata and 'state' in metadata: + state = metadata['state'] if state in ['waiting_for_clients', 'active']: logger.info(f"[{self.client_id}] Channel {self.channel_id} now ready (state={state})") return True elif state in ['error', 'stopped', 'stopping']: # Added 'stopping' to error states - error_message = metadata.get(b'error_message', b'Unknown error').decode('utf-8') + error_message = metadata.get('error_message', 'Unknown error') logger.error(f"[{self.client_id}] Channel {self.channel_id} in error state: {state}, message: {error_message}") # Send error packet before giving up yield create_ts_packet('error', f"Error: {error_message}") @@ -122,9 +122,9 @@ class StreamGenerator: else: # Improved logging to track initialization progress init_time = "unknown" - if b'init_time' in metadata: + if 'init_time' in metadata: try: - init_time_float = float(metadata[b'init_time'].decode('utf-8')) + init_time_float = float(metadata['init_time']) init_duration = time.time() - init_time_float init_time = f"{init_duration:.1f}s ago" except: @@ -267,8 +267,8 @@ class StreamGenerator: # Also check channel state in metadata metadata_key = RedisKeys.channel_metadata(self.channel_id) metadata = proxy_server.redis_client.hgetall(metadata_key) - if metadata and b'state' in metadata: - state = metadata[b'state'].decode('utf-8') + if metadata and 'state' in metadata: + state = metadata['state'] if state in ['error', 'stopped', 'stopping']: logger.info(f"[{self.client_id}] Channel in {state} state, terminating stream") return False @@ -397,8 +397,6 @@ class StreamGenerator: if metadata: stream_id_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID) if stream_id_bytes: - stream_id = int(stream_id_bytes.decode('utf-8')) - # Check if we're the last client if self.channel_id in proxy_server.client_managers: client_count = proxy_server.client_managers[self.channel_id].get_total_client_count() @@ -459,4 +457,4 @@ def create_stream_generator(channel_id, client_id, client_ip, client_user_agent, Returns a function that can be passed to StreamingHttpResponse. """ generator = StreamGenerator(channel_id, client_id, client_ip, client_user_agent, channel_initializing) - return generator.generate \ No newline at end of file + return generator.generate From 2ca2a51e2e63dcdc8c8ab7942472caaf2b9f4a0d Mon Sep 17 00:00:00 2001 From: dekzter Date: Sat, 6 Dec 2025 09:36:50 -0500 Subject: [PATCH 004/172] reverted fix so only updated values are modified --- apps/proxy/ts_proxy/channel_status.py | 34 +++++++++++++++++++++------ 1 file changed, 27 insertions(+), 7 deletions(-) diff --git a/apps/proxy/ts_proxy/channel_status.py b/apps/proxy/ts_proxy/channel_status.py index 015d5ca2..4ac8f63f 100644 --- a/apps/proxy/ts_proxy/channel_status.py +++ b/apps/proxy/ts_proxy/channel_status.py @@ -434,13 +434,33 @@ class ChannelStatus: logger.warning(f"Invalid m3u_profile_id format in Redis: {m3u_profile_id}") # Add stream info to basic info as well - info['video_codec'] = metadata.get(ChannelMetadataField.VIDEO_CODEC) - info['resolution'] = metadata.get(ChannelMetadataField.RESOLUTION) - info['source_fps'] = metadata.get(ChannelMetadataField.SOURCE_FPS) - info['ffmpeg_speed'] = metadata.get(ChannelMetadataField.FFMPEG_SPEED) - info['audio_codec'] = metadata.get(ChannelMetadataField.AUDIO_CODEC) - info['audio_channels'] = metadata.get(ChannelMetadataField.AUDIO_CHANNELS) - info['stream_type'] = metadata.get(ChannelMetadataField.STREAM_TYPE) + video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC) + if video_codec: + info['video_codec'] = video_codec + + resolution = metadata.get(ChannelMetadataField.RESOLUTION) + if resolution: + info['resolution'] = resolution + + source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS) + if source_fps: + info['source_fps'] = float(source_fps) + + ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED) + if ffmpeg_speed: + info['ffmpeg_speed'] = float(ffmpeg_speed) + + audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC) + if audio_codec: + info['audio_codec'] = audio_codec + + audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS) + if audio_channels: + info['audio_channels'] = audio_channels + + stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE) + if stream_type: + info['stream_type'] = stream_type return info except Exception as e: From dba70b3c585e0f2dce6cdcf157045992e617b236 Mon Sep 17 00:00:00 2001 From: dekzter Date: Tue, 27 Jan 2026 15:23:07 -0500 Subject: [PATCH 005/172] outstanding commits, may revisit later --- .../migrations/0004_user_stream_priority.py | 18 +++ apps/channels/models.py | 153 +++++++++++++++++- apps/channels/tasks.py | 5 +- apps/proxy/ts_proxy/client_manager.py | 16 +- apps/proxy/ts_proxy/server.py | 6 - apps/proxy/ts_proxy/utils.py | 4 +- apps/proxy/ts_proxy/views.py | 26 +-- docker/init/02-postgres.sh | 2 - docker/init/03-init-dispatcharr.sh | 2 +- docker/init/99-init-dev.sh | 50 +++--- docker/uwsgi.dev.ini | 4 +- frontend/prettier.config.js | 4 +- frontend/src/components/forms/User.jsx | 40 +++-- frontend/src/pages/Users.jsx | 35 ---- 14 files changed, 242 insertions(+), 123 deletions(-) create mode 100644 apps/accounts/migrations/0004_user_stream_priority.py diff --git a/apps/accounts/migrations/0004_user_stream_priority.py b/apps/accounts/migrations/0004_user_stream_priority.py new file mode 100644 index 00000000..35c8bbae --- /dev/null +++ b/apps/accounts/migrations/0004_user_stream_priority.py @@ -0,0 +1,18 @@ +# Generated by Django 5.2.4 on 2025-10-21 19:32 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('accounts', '0003_alter_user_custom_properties'), + ] + + operations = [ + migrations.AddField( + model_name='user', + name='stream_priority', + field=models.IntegerField(default=0, help_text='Priority level for streaming tasks. Lower values indicate higher priority.'), + ), + ] diff --git a/apps/channels/models.py b/apps/channels/models.py index 3dfb392b..7282d025 100644 --- a/apps/channels/models.py +++ b/apps/channels/models.py @@ -166,7 +166,7 @@ class Stream(models.Model): return stream_profile - def get_stream(self): + def get_stream(self, requester=None): """ Finds an available stream for the requested channel and returns the selected stream and profile. """ @@ -349,7 +349,145 @@ class Channel(models.Model): return stream_profile - def get_stream(self): + def _pick_channel_to_preempt( + self, + profile_id, + requester_level, + redis_client, + exclude_channel_ids=None, + cooldown_seconds=30, + ): + """ + Pick the lowest-impact channel to terminate on the given profile. + Returns: Optional[int] channel_id to preempt + """ + exclude_channel_ids = set(exclude_channel_ids or []) + candidates = [] + + # 1) Try to get active channel IDs for this profile from an index set if available + ch_set_key = f"ts_proxy:profile:{profile_id}:channels" + try: + ch_ids = { (int(x) if not isinstance(x, int) else x) for x in (redis_client.smembers(ch_set_key) or set()) } + except Exception: + ch_ids = set() + + logger.debug("Candidate channels for preemption:") + logger.debug(ch_ids) + + # 2) Fallback: scan metadata keys and filter by m3u_profile == profile_id + if not ch_ids: + cursor = 0 + pattern = "ts_proxy:channel:*:metadata" + while True: + cursor, keys = redis_client.scan(cursor=cursor, match=pattern, count=500) + if keys: + # Prefer HGET m3u_profile if metadata is a hash + pipe = redis_client.pipeline() + for k in keys: + pipe.hget(k, "m3u_profile") + prof_vals = pipe.execute() + for k, prof_val in zip(keys, prof_vals): + try: + pid = int(prof_val) if prof_val is not None else None + except Exception: + pid = None + + if pid == profile_id: + parts = k.split(":") # ts_proxy:channel:{id}:metadata + if len(parts) >= 4: + try: + ch_ids.add(int(parts[2])) + except Exception: + pass + if cursor == 0: + break + + logger.debug("Candidate channels for preemption:") + logger.debug(ch_ids) + + if not ch_ids: + return None + + # 3) Score candidates + for ch_id in ch_ids: + if ch_id in exclude_channel_ids: + continue + + # Skip if recently preempted + last_preempt_key = f"ts_proxy:channel:{ch_id}:last_preempt" + try: + last_preempt = float(redis_client.get(last_preempt_key) or 0.0) + except Exception: + last_preempt = 0.0 + if last_preempt and (time.time() - last_preempt) < cooldown_seconds: + continue + + # Clients and their levels + clients_key = f"ts_proxy:channel:{ch_id}:clients" + member_ids = list(redis_client.smembers(clients_key) or []) + viewer_count = len(member_ids) + max_viewer_level = 0 + if viewer_count: + pipe = redis_client.pipeline() + for cid in member_ids: + pipe.hget(f"ts_proxy:channel:{ch_id}:clients:{cid}", "user_level") + levels_raw = pipe.execute() + levels = [] + for lv in levels_raw: + try: + levels.append(int(lv or 0)) + except Exception: + levels.append(0) + max_viewer_level = max(levels or [0]) + + # Only preempt if requester strictly outranks this channel's viewers + if requester_level <= max_viewer_level: + continue + + # Metadata (protected/recording/started_at_ts) + meta_key = f"ts_proxy:channel:{ch_id}:metadata" + try: + protected, recording, started_at_ts = redis_client.hmget( + meta_key, "protected", "recording", "started_at_ts" + ) + except Exception: + protected = recording = started_at_ts = None + + protected = str(protected or "0") in ("1", "true", "True") + recording = str(recording or "0") in ("1", "true", "True") + if protected or recording: + continue + + try: + started_at_ts = float(started_at_ts) if started_at_ts is not None else None + except Exception: + started_at_ts = None + if started_at_ts is None: + started_at_ts = time.time() # treat unknown as newest + + # Score: lower is safer to terminate + has_viewers = 1 if viewer_count > 0 else 0 + score = (has_viewers, max_viewer_level, viewer_count, started_at_ts) + candidates.append((score, ch_id)) + + logger.debug("Candidate channels after scoring:") + logger.debug(candidates) + + if not candidates: + return None + + candidates.sort(key=lambda x: x[0]) + victim_id = candidates[0][1] + + # Mark preempt timestamp to avoid thrashing + try: + redis_client.set(f"ts_proxy:channel:{victim_id}:last_preempt", str(time.time()), ex=3600) + except Exception: + pass + + return victim_id + + def get_stream(self, requester=None): """ Finds an available stream for the requested channel and returns the selected stream and profile. @@ -438,6 +576,17 @@ class Channel(models.Model): None, ) # Return newly assigned stream and matched profile else: + # At capacity: try to preempt a lower-impact channel on this profile + victim_channel_id = self._pick_channel_to_preempt( + profile_id=profile.id, + requester_level=requester.user_level if requester else 100, + redis_client=redis_client, + exclude_channel_ids=None, + ) + if victim_channel_id: + logger.info(f"Preempting channel {victim_channel_id} for new stream on profile {profile.id}") + # return self.id, profile.id, victim_channel_id + # This profile is at max connections has_streams_but_maxed_out = True logger.debug( diff --git a/apps/channels/tasks.py b/apps/channels/tasks.py index ffb2b97a..b63577d2 100755 --- a/apps/channels/tasks.py +++ b/apps/channels/tasks.py @@ -1900,11 +1900,8 @@ def run_recording(recording_id, channel_id, start_time_str, end_time_str): metadata_key = RedisKeys.channel_metadata(str(channel.uuid)) md = r.hgetall(metadata_key) if md: - def _gv(bkey): - return md.get(bkey) - def _d(bkey, cast=str): - v = _gv(bkey) + v = md.get(key) try: if v is None: return None diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index bffecdde..80be26f1 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/apps/proxy/ts_proxy/client_manager.py @@ -1,10 +1,8 @@ """Client connection management for TS streams""" import threading -import logging import time import json -import gevent from typing import Set, Optional from apps.proxy.config import TSConfig as Config from redis.exceptions import ConnectionError, TimeoutError @@ -123,7 +121,7 @@ class ClientManager: # Check for stale activity using last_active field last_active = self.redis_client.hget(client_key, "last_active") if last_active: - last_active_time = float(last_active.decode('utf-8')) + last_active_time = float(last_active) ghost_timeout = self.heartbeat_interval * getattr(Config, 'GHOST_CLIENT_MULTIPLIER', 5.0) if current_time - last_active_time > ghost_timeout: @@ -224,7 +222,7 @@ class ClientManager: except Exception as e: logger.error(f"Error notifying owner of client activity: {e}") - def add_client(self, client_id, client_ip, user_agent=None): + def add_client(self, client_id, client_ip, user_agent=None, user=None): """Add a client with duplicate prevention""" if client_id in self._registered_clients: logger.debug(f"Client {client_id} already registered, skipping") @@ -242,7 +240,8 @@ class ClientManager: "ip_address": client_ip, "connected_at": current_time, "last_active": current_time, - "worker_id": self.worker_id or "unknown" + "worker_id": self.worker_id or "unknown", + "user_level": user.user_level if user is not None else 100, # default to a high value since no user means the non-user specific M3U/HDHR } try: @@ -303,8 +302,6 @@ class ClientManager: def remove_client(self, client_id): """Remove a client from this channel and Redis""" - client_ip = None - with self.lock: if client_id in self.clients: self.clients.remove(client_id) @@ -317,11 +314,6 @@ class ClientManager: if self.redis_client: # Get client IP before removing the data client_key = f"ts_proxy:channel:{self.channel_id}:clients:{client_id}" - client_data = self.redis_client.hgetall(client_key) - if client_data and b'ip_address' in client_data: - client_ip = client_data[b'ip_address'].decode('utf-8') - elif client_data and 'ip_address' in client_data: - client_ip = client_data['ip_address'] # Remove from channel's client set self.redis_client.srem(self.client_set_key, client_id) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 9773e487..7d4a9f89 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -1121,12 +1121,6 @@ class ProxyServer: else: # Grace period expired with clients - mark channel as active logger.info(f"Grace period expired with {total_clients} clients - marking channel {channel_id} as active") -<<<<<<< HEAD - old_state = "unknown" - if metadata and 'state' in metadata: - old_state = metadata['state'] -======= ->>>>>>> origin/dev if self.update_channel_state(channel_id, ChannelState.ACTIVE, { "grace_period_ended_at": str(time.time()), "clients_at_activation": str(total_clients) diff --git a/apps/proxy/ts_proxy/utils.py b/apps/proxy/ts_proxy/utils.py index 20a6e140..ff9e1780 100644 --- a/apps/proxy/ts_proxy/utils.py +++ b/apps/proxy/ts_proxy/utils.py @@ -83,7 +83,7 @@ def create_ts_packet(packet_type='null', message=None): # Add message to payload if provided if message: - msg_bytes = message.encode('utf-8') + msg_bytes = message packet[4:4+min(len(msg_bytes), 180)] = msg_bytes[:180] return bytes(packet) @@ -113,4 +113,4 @@ def get_logger(component_name=None): # Default if detection fails logger_name = "ts_proxy" - return logging.getLogger(logger_name) \ No newline at end of file + return logging.getLogger(logger_name) diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index 91f254a7..0f9c1228 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -45,7 +45,7 @@ logger = get_logger() @api_view(["GET"]) -def stream_ts(request, channel_id): +def stream_ts(request, channel_id, user=None): if not network_access_allowed(request, "STREAMS"): return JsonResponse({"error": "Forbidden"}, status=403) @@ -80,9 +80,9 @@ def stream_ts(request, channel_id): metadata_key = RedisKeys.channel_metadata(channel_id) if proxy_server.redis_client.exists(metadata_key): metadata = proxy_server.redis_client.hgetall(metadata_key) - state_field = ChannelMetadataField.STATE.encode("utf-8") + state_field = ChannelMetadataField.STATE if state_field in metadata: - channel_state = metadata[state_field].decode("utf-8") + channel_state = metadata[state_field] # Active/running states - channel is operational, don't reinitialize if channel_state in [ @@ -120,7 +120,7 @@ def stream_ts(request, channel_id): else: owner_field = ChannelMetadataField.OWNER.encode("utf-8") if owner_field in metadata: - owner = metadata[owner_field].decode("utf-8") + owner = metadata[owner_field] owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat" if proxy_server.redis_client.exists(owner_heartbeat_key): # Owner is still active with unknown state - don't reinitialize @@ -373,7 +373,7 @@ def stream_ts(request, channel_id): metadata_key, ChannelMetadataField.STATE ) if state_bytes: - current_state = state_bytes.decode("utf-8") + current_state = state_bytes logger.debug( f"[{client_id}] Current state of channel {channel_id}: {current_state}" ) @@ -449,12 +449,12 @@ def stream_ts(request, channel_id): ) if url_bytes: - url = url_bytes.decode("utf-8") + url = url_bytes if ua_bytes: - stream_user_agent = ua_bytes.decode("utf-8") + stream_user_agent = ua_bytes # Extract transcode setting from Redis if profile_bytes: - profile_str = profile_bytes.decode("utf-8") + profile_str = profile_bytes use_transcode = ( profile_str == PROXY_PROFILE_NAME or profile_str == "None" ) @@ -490,7 +490,7 @@ def stream_ts(request, channel_id): # Register client buffer = proxy_server.stream_buffers[channel_id] client_manager = proxy_server.client_managers[channel_id] - client_manager.add_client(client_id, client_ip, client_user_agent) + client_manager.add_client(client_id, client_ip, client_user_agent, user) logger.info(f"[{client_id}] Client registered with channel {channel_id}") # Create a stream generator for this client @@ -553,7 +553,7 @@ def stream_xc(request, username, password, channel_id): channel = get_object_or_404(Channel, id=channel_id) # @TODO: we've got the file 'type' via extension, support this when we support multiple outputs - return stream_ts(request._request, str(channel.uuid)) + return stream_ts(request._request, str(channel.uuid), user) @csrf_exempt @@ -681,7 +681,7 @@ def channel_status(request, channel_id=None): ) for key in keys: channel_id_match = re.search( - r"ts_proxy:channel:(.*):metadata", key.decode("utf-8") + r"ts_proxy:channel:(.*):metadata", key ) if channel_id_match: ch_id = channel_id_match.group(1) @@ -802,7 +802,7 @@ def next_stream(request, channel_id): metadata_key, ChannelMetadataField.STREAM_ID ) if stream_id_bytes: - current_stream_id = int(stream_id_bytes.decode("utf-8")) + current_stream_id = int(stream_id_bytes) logger.info( f"Found current stream ID {current_stream_id} in Redis for channel {channel_id}" ) @@ -812,7 +812,7 @@ def next_stream(request, channel_id): metadata_key, ChannelMetadataField.M3U_PROFILE ) if profile_id_bytes: - profile_id = int(profile_id_bytes.decode("utf-8")) + profile_id = int(profile_id_bytes) logger.info( f"Found M3U profile ID {profile_id} in Redis for channel {channel_id}" ) diff --git a/docker/init/02-postgres.sh b/docker/init/02-postgres.sh index e36dd744..233a5fb8 100644 --- a/docker/init/02-postgres.sh +++ b/docker/init/02-postgres.sh @@ -162,5 +162,3 @@ ensure_utf8_encoding() { echo "Database $POSTGRES_DB converted to UTF8 and permissions set." fi } - - diff --git a/docker/init/03-init-dispatcharr.sh b/docker/init/03-init-dispatcharr.sh index 5fbef23d..bfe081e2 100644 --- a/docker/init/03-init-dispatcharr.sh +++ b/docker/init/03-init-dispatcharr.sh @@ -64,4 +64,4 @@ if [ "$(id -u)" = "0" ]; then fi chmod +x /data -fi \ No newline at end of file +fi diff --git a/docker/init/99-init-dev.sh b/docker/init/99-init-dev.sh index 861c8307..fc8ef18a 100644 --- a/docker/init/99-init-dev.sh +++ b/docker/init/99-init-dev.sh @@ -1,25 +1,33 @@ #!/bin/bash -echo "🚀 Development Mode - Setting up Frontend..." +if [ ! -e "/tmp/init" ]; then + echo "🚀 Development Mode - Setting up Frontend..." -# Install Node.js -if ! command -v node 2>&1 >/dev/null -then - echo "=== setting up nodejs ===" - curl -sL https://deb.nodesource.com/setup_23.x -o /tmp/nodesource_setup.sh - bash /tmp/nodesource_setup.sh - apt-get update - apt-get install -y --no-install-recommends \ - nodejs -fi - -# Install frontend dependencies -cd /app/frontend && npm install -# Install pip dependencies -cd /app && pip install -r requirements.txt - -# Install debugpy for remote debugging -if [ "$DISPATCHARR_DEBUG" = "true" ]; then - echo "=== setting up debugpy ===" - pip install debugpy + # Install Node.js + if ! command -v node 2>&1 >/dev/null + then + echo "=== setting up nodejs ===" + curl -sL https://deb.nodesource.com/setup_23.x -o /tmp/nodesource_setup.sh + bash /tmp/nodesource_setup.sh + apt-get update + apt-get install -y --no-install-recommends \ + nodejs + fi + + # Install frontend dependencies + cd /app/frontend && npm install + # Install pip dependencies + cd /app && pip install -r requirements.txt + + # Install debugpy for remote debugging + if [ "$DISPATCHARR_DEBUG" = "true" ]; then + echo "=== setting up debugpy ===" + pip install debugpy + fi + + if [[ "$DISPATCHARR_ENV" = "dev" ]]; then + touch /tmp/init + fi +else + echo "Development mode initialization already done. Skipping dev setup." fi diff --git a/docker/uwsgi.dev.ini b/docker/uwsgi.dev.ini index e476e216..e6d0806d 100644 --- a/docker/uwsgi.dev.ini +++ b/docker/uwsgi.dev.ini @@ -8,7 +8,7 @@ exec-pre = python /app/scripts/wait_for_redis.py ; Start Redis first -attach-daemon = redis-server +attach-daemon = redis-server --protected-mode no ; Then start other services with configurable nice level (default: 5 for low priority) ; Users can override via CELERY_NICE_LEVEL environment variable in docker-compose attach-daemon = nice -n $(CELERY_NICE_LEVEL) celery -A dispatcharr worker --autoscale=6,1 @@ -59,4 +59,4 @@ logformat-strftime = true log-date = %%Y-%%m-%%d %%H:%%M:%%S,000 # Use formatted time with environment variable for log level log-format = %(ftime) $(DISPATCHARR_LOG_LEVEL) uwsgi.requests Worker ID: %(wid) %(method) %(status) %(uri) %(msecs)ms -log-buffering = 1024 # Add buffer size limit for logging \ No newline at end of file +log-buffering = 1024 # Add buffer size limit for logging diff --git a/frontend/prettier.config.js b/frontend/prettier.config.js index d2b86878..e143dfef 100644 --- a/frontend/prettier.config.js +++ b/frontend/prettier.config.js @@ -3,8 +3,8 @@ export default { semi: true, // Add semicolons at the end of statements singleQuote: true, // Use single quotes instead of double tabWidth: 2, // Set the indentation width - trailingComma: "es5", // Add trailing commas where valid in ES5 + trailingComma: 'es5', // Add trailing commas where valid in ES5 printWidth: 80, // Wrap lines at 80 characters bracketSpacing: true, // Add spaces inside object braces - arrowParens: "always", // Always include parentheses around arrow function parameters + arrowParens: 'always', // Always include parentheses around arrow function parameters }; diff --git a/frontend/src/components/forms/User.jsx b/frontend/src/components/forms/User.jsx index 619b156f..6ec8705e 100644 --- a/frontend/src/components/forms/User.jsx +++ b/frontend/src/components/forms/User.jsx @@ -12,8 +12,9 @@ import { Stack, MultiSelect, ActionIcon, + NumberInput, } from '@mantine/core'; -import { RotateCcwKey, X } from 'lucide-react'; +import { RotateCcwKey } from 'lucide-react'; import { useForm } from '@mantine/form'; import useChannelsStore from '../../store/channels'; import { USER_LEVELS, USER_LEVEL_LABELS } from '../../constants'; @@ -209,26 +210,23 @@ const User = ({ user = null, isOpen, onClose }) => { key={form.key('last_name')} /> - - - - - } - /> - + + + + } + /> {showPermissions && ( { const authUser = useAuthStore((s) => s.user); - const [selectedUser, setSelectedUser] = useState(null); - const [userModalOpen, setUserModalOpen] = useState(false); - const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false); - const [deleteTarget, setDeleteTarget] = useState(null); - const [userToDelete, setUserToDelete] = useState(null); - if (!authUser.id) { return <>; } - const closeUserModal = () => { - setSelectedUser(null); - setUserModalOpen(false); - }; - const editUser = (user) => { - setSelectedUser(user); - setUserModalOpen(true); - }; - - const deleteUser = (id) => { - // Get user details for the confirmation dialog - const user = users.find((u) => u.id === id); - setUserToDelete(user); - setDeleteTarget(id); - - // Skip warning if it's been suppressed - if (isWarningSuppressed('delete-user')) { - return executeDeleteUser(id); - } - - setConfirmDeleteOpen(true); - }; - - const executeDeleteUser = async (id) => { - await API.deleteUser(id); - setConfirmDeleteOpen(false); - }; - return ( From 4796184a67e0eb32e5448c221346956c8b5ba884 Mon Sep 17 00:00:00 2001 From: Jeff Casimir Date: Sun, 1 Feb 2026 13:42:48 -0700 Subject: [PATCH 006/172] Fix uWSGI segfaults by disabling threads when using gevent Mixing threading and gevent concurrency models causes uWSGI workers to crash with segmentation faults, particularly on ARM64/Python 3.13. The production uwsgi.ini was already correct (gevent without threads). This fixes the dev and debug configs to match. References: - https://github.com/gevent/gevent/issues/1784 - https://github.com/unbit/uwsgi/issues/2457 Co-Authored-By: Claude Opus 4.5 --- docker/uwsgi.debug.ini | 2 -- docker/uwsgi.dev.ini | 4 +--- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/docker/uwsgi.debug.ini b/docker/uwsgi.debug.ini index 69c040f2..d2847335 100644 --- a/docker/uwsgi.debug.ini +++ b/docker/uwsgi.debug.ini @@ -28,8 +28,6 @@ static-map = /static=/app/static # Worker configuration workers = 1 -threads = 8 -enable-threads = true lazy-apps = true # HTTP server diff --git a/docker/uwsgi.dev.ini b/docker/uwsgi.dev.ini index e476e216..c4c5d0aa 100644 --- a/docker/uwsgi.dev.ini +++ b/docker/uwsgi.dev.ini @@ -28,10 +28,8 @@ vacuum = true die-on-term = true static-map = /static=/app/static -# Worker management (Optimize for I/O bound tasks) +# Worker management workers = 4 -threads = 2 -enable-threads = true # Optimize for streaming http = 0.0.0.0:5656 From 1dd368278ca733b9976b4a01a02df438c6d10eeb Mon Sep 17 00:00:00 2001 From: dekzter Date: Fri, 13 Mar 2026 09:57:22 -0400 Subject: [PATCH 007/172] reverted value checking for consistency, added in user id to client info --- .../migrations/0004_user_stream_priority.py | 18 ----- apps/proxy/ts_proxy/channel_status.py | 70 +++++++++++++++---- apps/proxy/ts_proxy/client_manager.py | 3 +- apps/proxy/ts_proxy/server.py | 7 +- 4 files changed, 59 insertions(+), 39 deletions(-) delete mode 100644 apps/accounts/migrations/0004_user_stream_priority.py diff --git a/apps/accounts/migrations/0004_user_stream_priority.py b/apps/accounts/migrations/0004_user_stream_priority.py deleted file mode 100644 index 35c8bbae..00000000 --- a/apps/accounts/migrations/0004_user_stream_priority.py +++ /dev/null @@ -1,18 +0,0 @@ -# Generated by Django 5.2.4 on 2025-10-21 19:32 - -from django.db import migrations, models - - -class Migration(migrations.Migration): - - dependencies = [ - ('accounts', '0003_alter_user_custom_properties'), - ] - - operations = [ - migrations.AddField( - model_name='user', - name='stream_priority', - field=models.IntegerField(default=0, help_text='Priority level for streaming tasks. Lower values indicate higher priority.'), - ), - ] diff --git a/apps/proxy/ts_proxy/channel_status.py b/apps/proxy/ts_proxy/channel_status.py index 4ac8f63f..ffb6be2c 100644 --- a/apps/proxy/ts_proxy/channel_status.py +++ b/apps/proxy/ts_proxy/channel_status.py @@ -265,22 +265,64 @@ class ChannelStatus: } # Add FFmpeg stream information - info['video_codec'] = metadata.get(ChannelMetadataField.VIDEO_CODEC) - info['resolution'] = metadata.get(ChannelMetadataField.RESOLUTION) - info['source_fps'] = metadata.get(ChannelMetadataField.SOURCE_FPS) - info['pixel_format'] = metadata.get(ChannelMetadataField.PIXEL_FORMAT) - info['source_bitrate'] = metadata.get(ChannelMetadataField.SOURCE_BITRATE) - info['audio_codec'] = metadata.get(ChannelMetadataField.AUDIO_CODEC) - info['sample_rate'] = metadata.get(ChannelMetadataField.SAMPLE_RATE) - info['audio_channels'] = metadata.get(ChannelMetadataField.AUDIO_CHANNELS) - info['audio_bitrate'] = metadata.get(ChannelMetadataField.AUDIO_BITRATE) + video_codec = metadata.get(ChannelMetadataField.VIDEO_CODEC) + if video_codec: + info['video_codec'] = video_codec + + resolution = metadata.get(ChannelMetadataField.RESOLUTION) + if resolution: + info['resolution'] = resolution + + source_fps = metadata.get(ChannelMetadataField.SOURCE_FPS) + if source_fps: + info['source_fps'] = source_fps + + pixel_format = metadata.get(ChannelMetadataField.PIXEL_FORMAT) + if pixel_format: + info['pixel_format'] = pixel_format + + source_bitrate = metadata.get(ChannelMetadataField.SOURCE_BITRATE) + if source_bitrate: + info['source_bitrate'] = source_bitrate + + audio_codec = metadata.get(ChannelMetadataField.AUDIO_CODEC) + if audio_codec: + info['audio_codec'] = audio_codec + + sample_rate = metadata.get(ChannelMetadataField.SAMPLE_RATE) + if sample_rate: + info['sample_rate'] = sample_rate + + audio_channels = metadata.get(ChannelMetadataField.AUDIO_CHANNELS) + if audio_channels: + info['audio_channels'] = audio_channels + + audio_bitrate = metadata.get(ChannelMetadataField.AUDIO_BITRATE) + if audio_bitrate: + info['audio_bitrate'] = audio_bitrate + # Add FFmpeg performance stats - info['ffmpeg_speed'] = metadata.get(ChannelMetadataField.FFMPEG_SPEED) - info['ffmpeg_fps'] = metadata.get(ChannelMetadataField.FFMPEG_FPS) - info['actual_fps'] = metadata.get(ChannelMetadataField.ACTUAL_FPS) - info['ffmpeg_bitrate'] = metadata.get(ChannelMetadataField.FFMPEG_BITRATE) - info['stream_type'] = metadata.get(ChannelMetadataField.STREAM_TYPE) + ffmpeg_speed = metadata.get(ChannelMetadataField.FFMPEG_SPEED) + if ffmpeg_speed: + info['ffmpeg_speed'] = ffmpeg_speed + + ffmpeg_fps = metadata.get(ChannelMetadataField.FFMPEG_FPS) + if ffmpeg_fps: + info['ffmpeg_fps'] = ffmpeg_fps + + actual_fps = metadata.get(ChannelMetadataField.ACTUAL_FPS) + if actual_fps: + info['actual_fps'] = actual_fps + + ffmpeg_bitrate = metadata.get(ChannelMetadataField.FFMPEG_BITRATE) + if ffmpeg_bitrate: + info['ffmpeg_bitrate'] = ffmpeg_bitrate + + stream_type = metadata.get(ChannelMetadataField.STREAM_TYPE) + if stream_type: + info['stream_type'] = stream_type + return info diff --git a/apps/proxy/ts_proxy/client_manager.py b/apps/proxy/ts_proxy/client_manager.py index 394ad5ea..c0e5ab0b 100644 --- a/apps/proxy/ts_proxy/client_manager.py +++ b/apps/proxy/ts_proxy/client_manager.py @@ -247,7 +247,8 @@ class ClientManager: "connected_at": current_time, "last_active": current_time, "worker_id": self.worker_id or "unknown", - "user_level": user.user_level if user is not None else 100, # default to a high value since no user means the non-user specific M3U/HDHR + "user_id": str(user.id) if user is not None else "unknown", + # "user_level": user.user_level if user is not None else 100, # default to a high value since no user means the non-user specific M3U/HDHR } try: diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 1999b491..4c51768a 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -459,10 +459,6 @@ class ProxyServer: lock_key = RedisKeys.channel_owner(channel_id) current = self.redis_client.get(lock_key) -<<<<<<< HEAD - # Only extend if we're still the owner - if current and current == self.worker_id: -======= if current is None: # Key expired — re-acquire if we have the stream_manager if channel_id in self.stream_managers: @@ -476,8 +472,7 @@ class ProxyServer: return False return False - if current.decode('utf-8') == self.worker_id: ->>>>>>> origin/dev + if current == self.worker_id: self.redis_client.expire(lock_key, ttl) return True From b22f9e8e07a68047821214fd2b666770e29dab65 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Sat, 14 Mar 2026 18:38:59 -0500 Subject: [PATCH 008/172] Enhancement: Account expiration tracking and notifications for M3U profiles --- CHANGELOG.md | 6 + .../0019_m3uaccountprofile_exp_date.py | 57 +++++++ apps/m3u/models.py | 79 +++++---- apps/m3u/serializers.py | 66 +++++++- apps/m3u/signals.py | 86 +++++++++- apps/m3u/tasks.py | 159 +++++++++++++++++- dispatcharr/settings.py | 5 + frontend/src/api.js | 1 + frontend/src/components/forms/M3U.jsx | 39 ++++- frontend/src/components/forms/M3UProfile.jsx | 32 ++++ frontend/src/components/tables/M3UsTable.jsx | 88 +++++++++- frontend/src/store/notifications.jsx | 13 +- 12 files changed, 578 insertions(+), 53 deletions(-) create mode 100644 apps/m3u/migrations/0019_m3uaccountprofile_exp_date.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6571bfcf..232521b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -41,6 +41,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Search and filter controls**: Added search and filter controls to the recordings list. - **Stream generator throttling**: Cached the `ProxyServer` singleton reference per client and throttled Redis resource checks (1 s) and non-owner health checks (2 s), eliminating 3+ Redis round-trips per stream loop iteration. - **Automatic crash recovery on worker restart**: A `worker_ready` Celery signal now fires `recover_recordings_on_startup` automatically when the worker starts, so recordings stuck in "recording" status are recovered without manual intervention. +- Account expiration tracking and notifications for M3U profiles + - A new `exp_date` field on `M3UAccountProfile` stores the account expiration date as a proper `DateTimeField`. For Xtream Codes accounts the field is auto-synced from `custom_properties.user_info.exp_date` on every save (supports both Unix timestamps and ISO date strings). For non-XC M3U accounts the date can be entered manually via the account or profile form. + - The M3U accounts table now shows an **Expiration** column displaying the earliest expiration date across all profiles for that account (color-coded: red = expired, orange = expiring soon, green = OK). Hovering the cell shows a tooltip with per-profile expiration details including inactive-profile labels. + - A daily Celery Beat task (`check_xc_account_expirations`) checks all active profiles with an expiration date and manages system notifications: a normal-priority warning is raised for profiles expiring within 7 days; a high-priority alert is raised once the profile has already expired. Warning and expired notifications use separate keys so dismissing the 7-day warning does not suppress the expiration alert. + - Notifications are also updated immediately when a profile is saved: if the expiration date is cleared or pushed beyond the 7-day window, any existing warning/expired notifications are deleted; if the date falls within the window or is already past, the matching notification is updated in place. + - Non-XC accounts expose a `DateTimePicker` on both the M3U account form and the profile form. ### Changed diff --git a/apps/m3u/migrations/0019_m3uaccountprofile_exp_date.py b/apps/m3u/migrations/0019_m3uaccountprofile_exp_date.py new file mode 100644 index 00000000..bae4d9dd --- /dev/null +++ b/apps/m3u/migrations/0019_m3uaccountprofile_exp_date.py @@ -0,0 +1,57 @@ +# Generated by Django 6.0.3 on 2026-03-14 19:41 + +from datetime import datetime, timezone + +from django.db import migrations, models + + +def populate_exp_date_from_custom_properties(apps, schema_editor): + """Backfill exp_date from custom_properties['user_info']['exp_date'].""" + M3UAccountProfile = apps.get_model('m3u', 'M3UAccountProfile') + profiles_to_update = [] + + for profile in M3UAccountProfile.objects.filter( + custom_properties__isnull=False, + ).exclude(custom_properties={}): + user_info = profile.custom_properties.get('user_info', {}) + raw_exp = user_info.get('exp_date') + if raw_exp is None: + continue + + parsed = None + try: + if isinstance(raw_exp, (int, float)): + parsed = datetime.fromtimestamp(float(raw_exp), tz=timezone.utc) + elif isinstance(raw_exp, str): + try: + parsed = datetime.fromtimestamp(float(raw_exp), tz=timezone.utc) + except ValueError: + parsed = datetime.fromisoformat(raw_exp) + except (ValueError, TypeError, OSError): + pass + + if parsed is not None: + profile.exp_date = parsed + profiles_to_update.append(profile) + + if profiles_to_update: + M3UAccountProfile.objects.bulk_update(profiles_to_update, ['exp_date'], batch_size=500) + + +class Migration(migrations.Migration): + + dependencies = [ + ('m3u', '0018_add_profile_custom_properties'), + ] + + operations = [ + migrations.AddField( + model_name='m3uaccountprofile', + name='exp_date', + field=models.DateTimeField(blank=True, help_text='Account expiration date, auto-synced from custom_properties on save', null=True), + ), + migrations.RunPython( + populate_exp_date_from_custom_properties, + reverse_code=migrations.RunPython.noop, + ), + ] diff --git a/apps/m3u/models.py b/apps/m3u/models.py index b812ad6c..fbe22d8c 100644 --- a/apps/m3u/models.py +++ b/apps/m3u/models.py @@ -1,3 +1,4 @@ +from datetime import datetime, timezone from django.db import models from django.core.exceptions import ValidationError from core.models import UserAgent @@ -264,11 +265,16 @@ class M3UAccountProfile(models.Model): ) current_viewers = models.PositiveIntegerField(default=0) custom_properties = models.JSONField( - default=dict, - blank=True, - null=True, + default=dict, + blank=True, + null=True, help_text="Custom properties for storing account information from provider (e.g., XC account details, expiration dates)" ) + exp_date = models.DateTimeField( + null=True, + blank=True, + help_text="Account expiration date, auto-synced from custom_properties on save", + ) class Meta: constraints = [ @@ -280,36 +286,51 @@ class M3UAccountProfile(models.Model): def __str__(self): return f"{self.name} ({self.m3u_account.name})" - def get_account_expiration(self): - """Get account expiration date from custom properties if available""" + def save(self, *args, **kwargs): + """Auto-sync exp_date from custom_properties for XC accounts on every save. + For non-XC accounts, exp_date is set directly and left untouched here.""" + parsed = self._parse_exp_date_from_custom_properties() + if parsed is not None: + # XC account with exp_date in custom_properties — always sync + self.exp_date = parsed + # else: keep whatever exp_date is already set (manual entry for non-XC) + super().save(*args, **kwargs) + + @staticmethod + def _parse_exp_date(raw_value): + """Parse a raw exp_date value (unix timestamp or ISO string) into a datetime.""" + if raw_value is None: + return None + try: + if isinstance(raw_value, (int, float)): + return datetime.fromtimestamp(float(raw_value), tz=timezone.utc) + elif isinstance(raw_value, str): + try: + return datetime.fromtimestamp(float(raw_value), tz=timezone.utc) + except ValueError: + return datetime.fromisoformat(raw_value) + except (ValueError, TypeError, OSError): + pass + return None + + def _parse_exp_date_from_custom_properties(self): + """Extract exp_date from custom_properties JSON.""" if not self.custom_properties: return None - user_info = self.custom_properties.get('user_info', {}) - exp_date = user_info.get('exp_date') - - if exp_date: - try: - from datetime import datetime - # XC exp_date is typically a Unix timestamp - if isinstance(exp_date, (int, float)): - return datetime.fromtimestamp(exp_date) - elif isinstance(exp_date, str): - # Try to parse as timestamp first, then as ISO date - try: - return datetime.fromtimestamp(float(exp_date)) - except ValueError: - return datetime.fromisoformat(exp_date) - except (ValueError, TypeError): - pass - - return None + return self._parse_exp_date(user_info.get('exp_date')) + + def get_account_expiration(self): + """Get account expiration date — uses the dedicated field if set, otherwise parses JSON.""" + if self.exp_date: + return self.exp_date + return self._parse_exp_date_from_custom_properties() def get_account_status(self): """Get account status from custom properties if available""" if not self.custom_properties: return None - + user_info = self.custom_properties.get('user_info', {}) return user_info.get('status') @@ -317,7 +338,7 @@ class M3UAccountProfile(models.Model): """Get maximum connections from custom properties if available""" if not self.custom_properties: return None - + user_info = self.custom_properties.get('user_info', {}) return user_info.get('max_connections') @@ -325,7 +346,7 @@ class M3UAccountProfile(models.Model): """Get active connections from custom properties if available""" if not self.custom_properties: return None - + user_info = self.custom_properties.get('user_info', {}) return user_info.get('active_cons') @@ -333,7 +354,7 @@ class M3UAccountProfile(models.Model): """Get last refresh timestamp from custom properties if available""" if not self.custom_properties: return None - + last_refresh = self.custom_properties.get('last_refresh') if last_refresh: try: @@ -341,7 +362,7 @@ class M3UAccountProfile(models.Model): return datetime.fromisoformat(last_refresh) except (ValueError, TypeError): pass - + return None diff --git a/apps/m3u/serializers.py b/apps/m3u/serializers.py index 22c4057a..c629cac9 100644 --- a/apps/m3u/serializers.py +++ b/apps/m3u/serializers.py @@ -52,12 +52,14 @@ class M3UAccountProfileSerializer(serializers.ModelSerializer): "search_pattern", "replace_pattern", "custom_properties", + "exp_date", "account", ] read_only_fields = ["id", "account"] extra_kwargs = { 'search_pattern': {'required': False, 'allow_blank': True}, 'replace_pattern': {'required': False, 'allow_blank': True}, + 'exp_date': {'required': False, 'allow_null': True}, } def create(self, validated_data): @@ -90,14 +92,14 @@ class M3UAccountProfileSerializer(serializers.ModelSerializer): def update(self, instance, validated_data): if instance.is_default: - # For default profiles, only allow updating name and custom_properties (for notes) - allowed_fields = {'name', 'custom_properties'} + # For default profiles, only allow updating name, custom_properties, and exp_date + allowed_fields = {'name', 'custom_properties', 'exp_date'} # Remove any fields that aren't allowed for default profiles disallowed_fields = set(validated_data.keys()) - allowed_fields if disallowed_fields: raise serializers.ValidationError( - f"Default profiles can only modify name and notes. " + f"Default profiles can only modify name, notes, and expiration. " f"Cannot modify: {', '.join(disallowed_fields)}" ) @@ -117,6 +119,12 @@ class M3UAccountSerializer(serializers.ModelSerializer): """Serializer for M3U Account""" filters = serializers.SerializerMethodField() + earliest_expiration = serializers.SerializerMethodField() + all_expirations = serializers.SerializerMethodField() + exp_date = serializers.DateTimeField( + required=False, allow_null=True, write_only=True, + help_text="Expiration date for the default profile (write-through)", + ) # Include user_agent as a mandatory field using its primary key. user_agent = serializers.PrimaryKeyRelatedField( queryset=UserAgent.objects.all(), @@ -172,6 +180,9 @@ class M3UAccountSerializer(serializers.ModelSerializer): "auto_enable_new_groups_live", "auto_enable_new_groups_vod", "auto_enable_new_groups_series", + "earliest_expiration", + "all_expirations", + "exp_date", ] extra_kwargs = { "password": { @@ -200,9 +211,19 @@ class M3UAccountSerializer(serializers.ModelSerializer): 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 + + # Surface default profile's exp_date for the form + default_profile = instance.profiles.filter(is_default=True).first() + data["exp_date"] = ( + default_profile.exp_date.isoformat() if default_profile and default_profile.exp_date else None + ) + return data def update(self, instance, validated_data): + # Pop exp_date — it's written to the default profile, not the account + exp_date = validated_data.pop("exp_date", "__NOT_SET__") + # Pop cron_expression before it reaches model fields # If not present (partial update), preserve the existing cron from the PeriodicTask if "cron_expression" in validated_data: @@ -264,9 +285,19 @@ class M3UAccountSerializer(serializers.ModelSerializer): memberships_to_update, ["enabled"] ) + # Write exp_date through to the default profile + if exp_date != "__NOT_SET__": + default_profile = instance.profiles.filter(is_default=True).first() + if default_profile: + default_profile.exp_date = exp_date + default_profile.save() + return instance def create(self, validated_data): + # Pop exp_date — it's written to the default profile after creation + exp_date = validated_data.pop("exp_date", None) + # Pop cron_expression — it's not a model field cron_expr = validated_data.pop("cron_expression", "") @@ -290,12 +321,41 @@ class M3UAccountSerializer(serializers.ModelSerializer): instance = M3UAccount(**validated_data) instance._cron_expression = cron_expr instance.save() + + # Write exp_date through to the default profile created by post_save signal + if exp_date is not None: + default_profile = instance.profiles.filter(is_default=True).first() + if default_profile: + default_profile.exp_date = exp_date + default_profile.save() + return instance def get_filters(self, obj): filters = obj.filters.order_by("order") return M3UFilterSerializer(filters, many=True).data + def get_earliest_expiration(self, obj): + """Return the soonest exp_date across all active profiles for this account.""" + profiles_with_exp = obj.profiles.filter( + is_active=True, exp_date__isnull=False + ).order_by("exp_date") + first = profiles_with_exp.first() + return first.exp_date.isoformat() if first else None + + def get_all_expirations(self, obj): + """Return exp_date info for every profile that has one (for tooltip).""" + profiles = obj.profiles.filter(exp_date__isnull=False).order_by("exp_date") + return [ + { + "profile_id": p.id, + "profile_name": p.name, + "exp_date": p.exp_date.isoformat(), + "is_active": p.is_active, + } + for p in profiles + ] + class ServerGroupSerializer(serializers.ModelSerializer): """Serializer for Server Group""" diff --git a/apps/m3u/signals.py b/apps/m3u/signals.py index 3de67c90..ce24d015 100644 --- a/apps/m3u/signals.py +++ b/apps/m3u/signals.py @@ -1,7 +1,7 @@ # apps/m3u/signals.py from django.db.models.signals import post_save, post_delete, pre_save from django.dispatch import receiver -from .models import M3UAccount +from .models import M3UAccount, M3UAccountProfile from .tasks import refresh_single_m3u_account, refresh_m3u_groups, delete_m3u_refresh_task_by_id from core.scheduling import create_or_update_periodic_task, delete_periodic_task import json @@ -68,6 +68,62 @@ def create_or_update_refresh_task(sender, instance, created, update_fields=None, if instance.refresh_task_id != task.id: M3UAccount.objects.filter(id=instance.id).update(refresh_task=task) +@receiver(post_save, sender=M3UAccountProfile) +def update_profile_expiration_notification(sender, instance, created, update_fields=None, **kwargs): + """ + When a profile's exp_date is set or changed, immediately update its expiration notification + so the frontend reflects the new state without waiting for the daily celery task. + """ + # Only act when exp_date was involved in the save + if not created and update_fields is not None and "exp_date" not in update_fields: + return + + try: + if not instance.exp_date: + # exp_date was cleared — remove any existing notifications immediately + from core.models import SystemNotification + from core.utils import send_notification_dismissed + + keys = [f"xc-exp-warning-{instance.id}", f"xc-exp-expired-{instance.id}"] + deleted_keys = list( + SystemNotification.objects.filter(notification_key__in=keys) + .values_list("notification_key", flat=True) + ) + SystemNotification.objects.filter(notification_key__in=deleted_keys).delete() + for key in deleted_keys: + send_notification_dismissed(key) + return + + from apps.m3u.tasks import evaluate_profile_expiration_notification + evaluate_profile_expiration_notification(instance) + except Exception as e: + logger.error(f"Error updating expiration notification for profile {instance.id}: {str(e)}") + + +@receiver(post_delete, sender=M3UAccountProfile) +def cleanup_profile_notifications(sender, instance, **kwargs): + """ + Delete expiration notifications for a profile when it is deleted. + Handles both direct deletion and cascade deletion from M3UAccount. + """ + try: + from core.models import SystemNotification + from core.utils import send_notification_dismissed + + keys = [f"xc-exp-warning-{instance.id}", f"xc-exp-expired-{instance.id}"] + deleted_keys = list( + SystemNotification.objects.filter(notification_key__in=keys) + .values_list("notification_key", flat=True) + ) + if deleted_keys: + SystemNotification.objects.filter(notification_key__in=deleted_keys).delete() + for key in deleted_keys: + send_notification_dismissed(key) + logger.debug(f"Cleaned up {len(deleted_keys)} notifications for deleted profile {instance.id}") + except Exception as e: + logger.error(f"Error cleaning up notifications for profile {instance.id}: {str(e)}") + + @receiver(post_delete, sender=M3UAccount) def delete_refresh_task(sender, instance, **kwargs): """ @@ -107,6 +163,34 @@ def update_status_on_active_change(sender, instance, **kwargs): else: # When deactivating, set status to disabled instance.status = M3UAccount.Status.DISABLED + # Clean up any expiration notifications for all profiles of this account + try: + from core.models import SystemNotification + from core.utils import send_notification_dismissed + + profile_ids = list( + M3UAccountProfile.objects.filter(m3u_account=instance) + .values_list("id", flat=True) + ) + keys = [ + key + for pid in profile_ids + for key in [f"xc-exp-warning-{pid}", f"xc-exp-expired-{pid}"] + ] + if keys: + deleted_keys = list( + SystemNotification.objects.filter(notification_key__in=keys) + .values_list("notification_key", flat=True) + ) + if deleted_keys: + SystemNotification.objects.filter(notification_key__in=deleted_keys).delete() + for key in deleted_keys: + send_notification_dismissed(key) + logger.debug( + f"Cleaned up {len(deleted_keys)} notifications for deactivated M3U account {instance.id}" + ) + except Exception as notify_err: + logger.error(f"Error cleaning up notifications on account deactivation: {notify_err}") except M3UAccount.DoesNotExist: # New record, will use default status pass diff --git a/apps/m3u/tasks.py b/apps/m3u/tasks.py index 5259cefc..0af5d382 100644 --- a/apps/m3u/tasks.py +++ b/apps/m3u/tasks.py @@ -11,7 +11,7 @@ from celery.result import AsyncResult from celery import shared_task, current_app, group from django.conf import settings from django.core.cache import cache -from django.db import transaction +from django.db import models, transaction from .models import M3UAccount from apps.channels.models import Stream, ChannelGroup, ChannelGroupM3UAccount from asgiref.sync import async_to_sync @@ -3185,3 +3185,160 @@ def send_m3u_update(account_id, action, progress, **kwargs): # Explicitly clear data reference to help garbage collection data = None + + +def evaluate_profile_expiration_notification(profile): + """ + Evaluate a single M3UAccountProfile's expiration date and create, update, + or delete the corresponding SystemNotification accordingly. + + Returns the notification key that should remain active (warning or expired), + or None if the profile is not expiring soon and any stale notifications were removed. + This return value is used by the bulk task to track active keys for stale cleanup. + """ + from core.models import SystemNotification + from core.utils import send_websocket_notification, send_notification_dismissed + + now = timezone.now() + warning_threshold = now + timezone.timedelta(days=7) + exp = profile.exp_date + warning_key = f"xc-exp-warning-{profile.id}" + expired_key = f"xc-exp-expired-{profile.id}" + + if exp <= now: + # Already expired — delete warning, create/update expired notification + deleted_warning = list( + SystemNotification.objects.filter(notification_key=warning_key) + .values_list("notification_key", flat=True) + ) + SystemNotification.objects.filter(notification_key=warning_key).delete() + for key in deleted_warning: + send_notification_dismissed(key) + + notification, created = SystemNotification.objects.update_or_create( + notification_key=expired_key, + defaults={ + "notification_type": SystemNotification.NotificationType.WARNING, + "priority": SystemNotification.Priority.HIGH, + "title": f"Account Expired: {profile.name}", + "message": ( + f'Profile "{profile.name}" on M3U account ' + f'"{profile.m3u_account.name}" has expired ' + f"(expired {exp.strftime('%Y-%m-%d %H:%M UTC')})." + ), + "action_data": { + "profile_id": profile.id, + "account_id": profile.m3u_account.id, + "account_name": profile.m3u_account.name, + "profile_name": profile.name, + "exp_date": exp.isoformat(), + }, + "is_active": True, + "admin_only": False, + }, + ) + send_websocket_notification(notification) + return expired_key + + elif exp <= warning_threshold: + # Expiring within 7 days — delete expired notification, create/update warning + deleted_expired = list( + SystemNotification.objects.filter(notification_key=expired_key) + .values_list("notification_key", flat=True) + ) + SystemNotification.objects.filter(notification_key=expired_key).delete() + for key in deleted_expired: + send_notification_dismissed(key) + + days_left = (exp - now).days + if days_left == 0: + expires_in_str = "today" + elif days_left == 1: + expires_in_str = "in 1 day" + else: + expires_in_str = f"in {days_left} days" + + notification, created = SystemNotification.objects.update_or_create( + notification_key=warning_key, + defaults={ + "notification_type": SystemNotification.NotificationType.WARNING, + "priority": SystemNotification.Priority.NORMAL, + "title": f"Account Expiring: {profile.name}", + "message": ( + f'Profile "{profile.name}" on M3U account ' + f'"{profile.m3u_account.name}" expires {expires_in_str} ' + f"(expires {exp.strftime('%Y-%m-%d %H:%M UTC')})." + ), + "action_data": { + "profile_id": profile.id, + "account_id": profile.m3u_account.id, + "account_name": profile.m3u_account.name, + "profile_name": profile.name, + "exp_date": exp.isoformat(), + }, + "is_active": True, + "admin_only": False, + }, + ) + send_websocket_notification(notification) + return warning_key + + else: + # Not expiring soon — delete any stale notifications + deleted_keys = list( + SystemNotification.objects.filter( + notification_key__in=[warning_key, expired_key] + ).values_list("notification_key", flat=True) + ) + SystemNotification.objects.filter( + notification_key__in=[warning_key, expired_key] + ).delete() + for key in deleted_keys: + send_notification_dismissed(key) + return None + + +@shared_task +def check_xc_account_expirations(): + """ + Daily task: check all account profiles for upcoming expirations. + Creates/updates SystemNotifications for profiles expiring within 7 days. + Uses separate notification keys for warning vs expired so users can + dismiss the 7-day warning and still receive the expired notification. + """ + from apps.m3u.models import M3UAccountProfile + from core.models import SystemNotification + from core.utils import send_notification_dismissed + + # Find all active profiles with an exp_date that is set + expiring_profiles = ( + M3UAccountProfile.objects.filter( + m3u_account__is_active=True, + is_active=True, + exp_date__isnull=False, + ) + .select_related("m3u_account") + ) + + active_notification_keys = set() + + for profile in expiring_profiles: + active_key = evaluate_profile_expiration_notification(profile) + if active_key: + active_notification_keys.add(active_key) + + # Delete stale notifications for profiles whose expiration was extended + stale = SystemNotification.objects.filter( + is_active=True, + ).filter( + models.Q(notification_key__startswith="xc-exp-warning-") | + models.Q(notification_key__startswith="xc-exp-expired-") + ).exclude(notification_key__in=active_notification_keys) + stale_keys = list(stale.values_list("notification_key", flat=True)) + stale.delete() + for key in stale_keys: + send_notification_dismissed(key) + + logger.info( + f"Account expiration check complete: {len(active_notification_keys)} active notifications" + ) diff --git a/dispatcharr/settings.py b/dispatcharr/settings.py index 50bac5d5..46e03d1f 100644 --- a/dispatcharr/settings.py +++ b/dispatcharr/settings.py @@ -267,6 +267,11 @@ CELERY_BEAT_SCHEDULE = { "task": "core.tasks.check_for_version_update", "schedule": 86400.0, # Once every 24 hours }, + # Check for XC account expirations daily + "check-xc-account-expirations": { + "task": "apps.m3u.tasks.check_xc_account_expirations", + "schedule": 86400.0, # Once every 24 hours + }, } MEDIA_ROOT = BASE_DIR / "media" diff --git a/frontend/src/api.js b/frontend/src/api.js index 4fa3d46c..b1c103d8 100644 --- a/frontend/src/api.js +++ b/frontend/src/api.js @@ -1286,6 +1286,7 @@ export default class API { body = new FormData(); for (const prop in values) { + if (values[prop] === null || values[prop] === undefined) continue; body.append(prop, values[prop]); } } else { diff --git a/frontend/src/components/forms/M3U.jsx b/frontend/src/components/forms/M3U.jsx index 4b3522af..423b2a05 100644 --- a/frontend/src/components/forms/M3U.jsx +++ b/frontend/src/components/forms/M3U.jsx @@ -29,6 +29,7 @@ import useEPGsStore from '../../store/epgs'; import useVODStore from '../../store/useVODStore'; import M3UFilters from './M3UFilters'; import ScheduleInput from './ScheduleInput'; +import { DateTimePicker } from '@mantine/dates'; const M3U = ({ m3uAccount = null, @@ -69,6 +70,7 @@ const M3U = ({ stale_stream_days: 7, priority: 0, enable_vod: false, + exp_date: null, }, validate: { @@ -101,6 +103,7 @@ const M3U = ({ ? m3uAccount.priority : 0, enable_vod: m3uAccount.enable_vod || false, + exp_date: m3uAccount.exp_date ? new Date(m3uAccount.exp_date) : null, }); // Determine schedule type from existing data @@ -131,6 +134,16 @@ const M3U = ({ const onSubmit = async () => { const { create_epg, ...values } = form.getValues(); + // Convert exp_date Date object to ISO string for the API + if (values.account_type === 'XC') { + // XC accounts have exp_date auto-managed server-side; don't send it + delete values.exp_date; + } else if (values.exp_date instanceof Date) { + values.exp_date = values.exp_date.toISOString(); + } else if (!values.exp_date) { + values.exp_date = null; + } + // Determine which schedule type is active based on field values const hasCronExpression = values.cron_expression && values.cron_expression.trim() !== ''; @@ -359,13 +372,25 @@ const M3U = ({ )} {form.getValues().account_type != 'XC' && ( - + <> + + + + )} diff --git a/frontend/src/components/forms/M3UProfile.jsx b/frontend/src/components/forms/M3UProfile.jsx index 025d3cae..7c24b212 100644 --- a/frontend/src/components/forms/M3UProfile.jsx +++ b/frontend/src/components/forms/M3UProfile.jsx @@ -16,6 +16,7 @@ import { Textarea, NumberInput, } from '@mantine/core'; +import { DateTimePicker } from '@mantine/dates'; import { useWebSocket } from '../../WebSocket'; import usePlaylistsStore from '../../store/playlists'; @@ -32,6 +33,8 @@ const RegexFormAndView = ({ profile = null, m3u, isOpen, onClose }) => { const [sampleInput, setSampleInput] = useState(''); const isDefaultProfile = profile?.is_default; + const isXC = m3u?.account_type === 'XC'; + const defaultValues = useMemo( () => ({ name: profile?.name || '', @@ -39,6 +42,7 @@ const RegexFormAndView = ({ profile = null, m3u, isOpen, onClose }) => { search_pattern: profile?.search_pattern || '', replace_pattern: profile?.replace_pattern || '', notes: profile?.custom_properties?.notes || '', + exp_date: profile?.exp_date ? new Date(profile.exp_date) : null, }), [profile] ); @@ -73,6 +77,17 @@ const RegexFormAndView = ({ profile = null, m3u, isOpen, onClose }) => { const onSubmit = async (values) => { console.log('submiting'); + // Convert exp_date for submission + let expDateValue = values.exp_date; + if (isXC) { + // XC accounts have exp_date auto-managed; don't send it + expDateValue = undefined; + } else if (expDateValue instanceof Date) { + expDateValue = expDateValue.toISOString(); + } else if (!expDateValue) { + expDateValue = null; + } + // For default profiles, only send name and custom_properties (notes) let submitValues; if (isDefaultProfile) { @@ -99,6 +114,11 @@ const RegexFormAndView = ({ profile = null, m3u, isOpen, onClose }) => { }; } + // Add exp_date for non-XC accounts + if (expDateValue !== undefined) { + submitValues.exp_date = expDateValue; + } + if (profile?.id) { await API.updateM3UProfile(m3u.id, { id: profile.id, @@ -257,6 +277,18 @@ const RegexFormAndView = ({ profile = null, m3u, isOpen, onClose }) => { )} + {!isXC && ( + setValue('exp_date', value)} + /> + )} +