From d6a452548c69cf3e92952df7ead4ff1218c817d0 Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Thu, 20 Mar 2025 15:28:12 -0500 Subject: [PATCH] Automatic stream switching is working. --- apps/proxy/config.py | 15 +- apps/proxy/ts_proxy/config_helper.py | 5 + apps/proxy/ts_proxy/server.py | 40 +- .../ts_proxy/services/channel_service.py | 42 +- apps/proxy/ts_proxy/stream_manager.py | 536 ++++++++++++------ apps/proxy/ts_proxy/url_utils.py | 167 +++++- apps/proxy/ts_proxy/views.py | 6 +- 7 files changed, 593 insertions(+), 218 deletions(-) diff --git a/apps/proxy/config.py b/apps/proxy/config.py index ffa0b559..28d3b872 100644 --- a/apps/proxy/config.py +++ b/apps/proxy/config.py @@ -23,34 +23,37 @@ class HLSConfig(BaseConfig): class TSConfig(BaseConfig): """Configuration settings for TS proxy""" - + # Connection settings CONNECTION_TIMEOUT = 10 # seconds to wait for initial connection MAX_RETRIES = 3 # maximum connection retry attempts - + # Buffer settings INITIAL_BEHIND_CHUNKS = 4 # How many chunks behind to start a client (4 chunks = ~1MB) CHUNK_BATCH_SIZE = 5 # How many chunks to fetch in one batch KEEPALIVE_INTERVAL = 0.5 # Seconds between keepalive packets when at buffer head - + # Streaming settings TARGET_BITRATE = 8000000 # Target bitrate (8 Mbps) STREAM_TIMEOUT = 10 # Disconnect after this many seconds of no data HEALTH_CHECK_INTERVAL = 5 # Check stream health every N seconds - + # Resource management CLEANUP_INTERVAL = 60 # Check for inactive channels every 60 seconds CHANNEL_SHUTDOWN_DELAY = 0 # How long to wait after last client before shutdown (seconds) - + # Client tracking settings CLIENT_RECORD_TTL = 5 # How long client records persist in Redis (seconds). Client will be considered MIA after this time. CLEANUP_CHECK_INTERVAL = 1 # How often to check for disconnected clients (seconds) CHANNEL_INIT_GRACE_PERIOD = 5 # How long to wait for first client after initialization (seconds) CLIENT_HEARTBEAT_INTERVAL = 1 # How often to send client heartbeats (seconds) GHOST_CLIENT_MULTIPLIER = 5.0 # How many heartbeat intervals before client considered ghost (5 would mean 5 secondsif heartbeat interval is 1) - + # TS packets are 188 bytes # Make chunk size a multiple of TS packet size for perfect alignment # ~1MB is ideal for streaming (matches typical media buffer sizes) BUFFER_CHUNK_SIZE = 188 * 1361 # ~256KB + # Maximum number of stream switch attempts before giving up + MAX_STREAM_SWITCHES = 10 + diff --git a/apps/proxy/ts_proxy/config_helper.py b/apps/proxy/ts_proxy/config_helper.py index fbdac746..5a16581a 100644 --- a/apps/proxy/ts_proxy/config_helper.py +++ b/apps/proxy/ts_proxy/config_helper.py @@ -68,3 +68,8 @@ class ConfigHelper: def max_retries(): """Get maximum retry attempts""" return ConfigHelper.get('MAX_RETRIES', 3) + + @staticmethod + def max_stream_switches(): + """Get maximum number of stream switch attempts""" + return ConfigHelper.get('MAX_STREAM_SWITCHES', 10) diff --git a/apps/proxy/ts_proxy/server.py b/apps/proxy/ts_proxy/server.py index 427b2089..4c7897c1 100644 --- a/apps/proxy/ts_proxy/server.py +++ b/apps/proxy/ts_proxy/server.py @@ -329,7 +329,7 @@ class ProxyServer: logger.error(f"Error extending ownership: {e}") return False - def initialize_channel(self, url, channel_id, user_agent=None, transcode=False): + def initialize_channel(self, url, channel_id, user_agent=None, transcode=False, stream_id=None): """Initialize a channel without redundant active key""" try: # Create buffer and client manager instances @@ -347,6 +347,7 @@ class ProxyServer: # Get channel URL from Redis if available channel_url = url channel_user_agent = user_agent + channel_stream_id = stream_id # Store the stream ID # First check if channel metadata already exists existing_metadata = None @@ -365,6 +366,14 @@ class ProxyServer: if ua_bytes: channel_user_agent = ua_bytes.decode('utf-8') + # Get stream ID from metadata if not provided + if not channel_stream_id and b'stream_id' in existing_metadata: + try: + channel_stream_id = int(existing_metadata[b'stream_id'].decode('utf-8')) + 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}") + # Check if channel is already owned current_owner = self.get_channel_owner(channel_id) @@ -419,9 +428,23 @@ class ProxyServer: if channel_user_agent: metadata["user_agent"] = channel_user_agent - # Set channel metadata + # CRITICAL FIX: Make sure stream_id is always set in metadata and properly logged + if channel_stream_id: + metadata["stream_id"] = str(channel_stream_id) + logger.info(f"Storing stream_id {channel_stream_id} in metadata for channel {channel_id}") + else: + logger.warning(f"No stream_id provided for channel {channel_id} during initialization") + + # Set channel metadata BEFORE creating the StreamManager self.redis_client.hset(metadata_key, mapping=metadata) - self.redis_client.expire(metadata_key, 30) # 1 hour TTL + self.redis_client.expire(metadata_key, 3600) # Increased TTL from 30 seconds to 1 hour + + # 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}") + 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) @@ -429,8 +452,15 @@ class ProxyServer: self.stream_buffers[channel_id] = buffer # Only the owner worker creates the actual stream manager - stream_manager = StreamManager(channel_id, channel_url, buffer, user_agent=channel_user_agent, transcode=transcode) - logger.debug(f"Created StreamManager for channel {channel_id}") + stream_manager = StreamManager( + channel_id, + channel_url, + buffer, + user_agent=channel_user_agent, + transcode=transcode, + stream_id=channel_stream_id # Pass stream ID to the manager + ) + logger.info(f"Created StreamManager for channel {channel_id} with stream ID {channel_stream_id}") self.stream_managers[channel_id] = stream_manager # Create client manager with channel_id, redis_client AND worker_id diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index b1f3bbb3..3fe6aedf 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -20,7 +20,7 @@ class ChannelService: """Service class for channel operations""" @staticmethod - def initialize_channel(channel_id, stream_url, user_agent, transcode=False, profile_value=None): + def initialize_channel(channel_id, stream_url, user_agent, transcode=False, profile_value=None, stream_id=None): """ Initialize a channel with the given parameters. @@ -30,16 +30,50 @@ class ChannelService: user_agent: User agent for the stream connection transcode: Whether to transcode the stream profile_value: Stream profile value to store in metadata + stream_id: ID of the stream being used Returns: bool: Success status """ - success = proxy_server.initialize_channel(stream_url, channel_id, user_agent, transcode) + # FIXED: First, ensure that Redis metadata including stream_id is set BEFORE channel initialization + # This ensures the stream ID is available when the StreamManager looks it up + if stream_id and proxy_server.redis_client: + metadata_key = RedisKeys.channel_metadata(channel_id) + # Check if metadata already exists + if proxy_server.redis_client.exists(metadata_key): + # Just update the existing metadata with stream_id + proxy_server.redis_client.hset(metadata_key, "stream_id", str(stream_id)) + logger.info(f"Pre-set stream ID {stream_id} in Redis for channel {channel_id}") + else: + # Create initial metadata with essential values + initial_metadata = { + "stream_id": str(stream_id), + "temp_init": str(time.time()) + } + proxy_server.redis_client.hset(metadata_key, mapping=initial_metadata) + logger.info(f"Created initial metadata with stream_id {stream_id} for channel {channel_id}") + + # Verify the stream_id was set + stream_id_value = proxy_server.redis_client.hget(metadata_key, "stream_id") + if stream_id_value: + logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is now set in Redis") + else: + logger.error(f"Failed to set stream_id {stream_id} in Redis before initialization") + + # Now proceed with channel initialization + success = proxy_server.initialize_channel(stream_url, channel_id, user_agent, transcode, stream_id) # Store additional metadata if initialization was successful - if success and proxy_server.redis_client and profile_value: + if success and proxy_server.redis_client: metadata_key = RedisKeys.channel_metadata(channel_id) - proxy_server.redis_client.hset(metadata_key, "profile", profile_value) + update_data = {} + if profile_value: + update_data["profile"] = profile_value + if stream_id: + update_data["stream_id"] = str(stream_id) + + if update_data: + proxy_server.redis_client.hset(metadata_key, mapping=update_data) return success diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index d16173e5..281ba842 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -17,13 +17,14 @@ from .utils import detect_stream_type from .redis_keys import RedisKeys from .constants import ChannelState, EventType, StreamType, TS_PACKET_SIZE from .config_helper import ConfigHelper +from .url_utils import get_alternate_streams, get_stream_info_for_switch logger = logging.getLogger("ts_proxy") class StreamManager: """Manages a connection to a TS stream without using raw sockets""" - def __init__(self, channel_id, url, buffer, user_agent=None, transcode=False): + def __init__(self, channel_id, url, buffer, user_agent=None, transcode=False, stream_id=None): # Basic properties self.channel_id = channel_id self.url = url @@ -54,6 +55,38 @@ class StreamManager: self._buffer_check_timers = [] self.stopping = False + # Add tracking for tried streams and current stream + self.current_stream_id = stream_id + self.tried_stream_ids = set() + + # IMPROVED LOGGING: Better handle and track stream ID + if stream_id: + self.tried_stream_ids.add(stream_id) + logger.info(f"Initialized stream manager for channel {buffer.channel_id} with stream ID {stream_id}") + else: + # Try to get stream ID from Redis metadata if available + if hasattr(buffer, 'redis_client') and buffer.redis_client: + try: + metadata_key = RedisKeys.channel_metadata(channel_id) + + # Log all metadata for debugging purposes + metadata = buffer.redis_client.hgetall(metadata_key) + if metadata: + logger.debug(f"Redis metadata for channel {channel_id}: {metadata}") + + # 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.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: + logger.warning(f"No stream_id found in Redis for channel {channel_id}") + except Exception as e: + logger.warning(f"Error loading stream ID from Redis: {e}") + else: + logger.warning(f"Unable to get stream ID for channel {channel_id} - stream switching may not work correctly") + logger.info(f"Initialized stream manager for channel {buffer.channel_id}") def _create_session(self): @@ -81,9 +114,13 @@ class StreamManager: return session def run(self): - """Main execution loop using HTTP streaming with improved connection handling""" + """Main execution loop using HTTP streaming with improved connection handling and stream switching""" # Add a stop flag to the class properties self.stop_requested = False + # Add tracking for stream switching attempts + stream_switch_attempts = 0 + # Get max stream switches from config using the helper method + max_stream_switches = ConfigHelper.max_stream_switches() # Prevent infinite switching loops try: # Check stream type before connecting @@ -102,200 +139,256 @@ class StreamManager: logger.info(f"Starting stream for URL: {self.url}") - while self.running: - if self.transcode: - if self.url_switching: - logger.debug("Skipping connection attempt during URL switch") - time.sleep(.1) - continue - # Generate transcode command - logger.debug(f"Building transcode command for channel {self.channel_id}") - channel = get_object_or_404(Channel, uuid=self.channel_id) + # Main stream switching loop - we'll try different streams if needed + while self.running and stream_switch_attempts <= max_stream_switches: + # Reset connection retry count for this specific URL + self.retry_count = 0 + url_failed = False + if self.url_switching: + logger.debug("Skipping connection attempt during URL switch") + time.sleep(0.1) + continue + # Connection retry loop for current URL + while self.running and self.retry_count < self.max_retries and not url_failed: - # Use FFmpeg specifically for HLS streams - if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: - from core.models import StreamProfile - try: - stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) - logger.info("Using FFmpeg stream profile for HLS content") - except StreamProfile.DoesNotExist: - # Fall back to channel's profile if FFmpeg not found - stream_profile = channel.get_stream_profile() - logger.warning("FFmpeg profile not found, using channel default profile") - else: - stream_profile = channel.get_stream_profile() + logger.info(f"Connection attempt {self.retry_count + 1}/{self.max_retries} for URL: {self.url}") - self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) - # Start command process for transcoding - logger.debug(f"Starting transcode process: {self.transcode_cmd}") - self.transcode_process = subprocess.Popen( - self.transcode_cmd, - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, # Suppress FFmpeg logs - bufsize=188 * 64 # Buffer optimized for TS packets - ) - self.socket = self.transcode_process.stdout # Read from FFmpeg output - self.connected = True - - if self.socket is not None: - # Set channel state to waiting for clients - self._set_waiting_for_clients() - - # Main fetch loop - while self.running and self.connected: - if self.fetch_chunk(): - self.last_data_time = time.time() - else: - if not self.running: - break - time.sleep(0.1) - else: + # Handle connection based on whether we transcode or not + connection_result = False try: - # Using direct HTTP streaming - if self.url_switching: - logger.debug("Skipping connection attempt during URL switch") - time.sleep(.1) - continue - logger.debug(f"Using TS Proxy to connect to stream: {self.url}") - # Create new session for each connection attempt - session = self._create_session() - self.current_session = session - - # Stream the URL with proper timeout handling - response = session.get( - self.url, - stream=True, - timeout=(10, 60) # 10s connect timeout, 60s read timeout - ) - self.current_response = response - - if response.status_code == 200: - self.connected = True - self.healthy = True - logger.info(f"Successfully connected to stream source") - - # Set channel state to waiting for clients - self._set_waiting_for_clients() - - # Process the stream in chunks with improved error handling - try: - chunk_count = 0 - for chunk in response.iter_content(chunk_size=self.chunk_size): - # Check if we've been asked to stop - if self.stop_requested: - logger.info(f"Stream loop for channel {self.channel_id} stopping due to request") - break - - if chunk: - # Add chunk to buffer with TS packet alignment - success = self.buffer.add_chunk(chunk) - - if success: - self.last_data_time = time.time() - chunk_count += 1 - - # Update last data timestamp in Redis - if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: - last_data_key = RedisKeys.last_data(self.buffer.channel_id) - self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) - except (AttributeError, ConnectionError) as e: - if self.stop_requested: - logger.debug(f"Expected connection error during shutdown: {e}") - elif hasattr(self, 'url_switching') and self.url_switching: - # This is expected during URL switching, just log at debug level - logger.debug(f"Expected connection error during URL switch: {e}") - else: - # Unexpected error during normal operation - logger.error(f"Unexpected stream error: {e}") - except Exception as e: - # Handle the specific 'NoneType' object has no attribute 'read' error - if "'NoneType' object has no attribute 'read'" in str(e): - logger.warning(f"Connection closed by server (read {chunk_count} chunks before disconnect)") - else: - # Re-raise unexpected AttributeErrors - logger.error(f"Unexpected AttributeError: {e}") - raise + if self.transcode: + connection_result = self._establish_transcode_connection() else: - logger.error(f"Failed to connect to stream: HTTP {response.status_code}") - time.sleep(2) + connection_result = self._establish_http_connection() - except requests.exceptions.ReadTimeout: - logger.warning("Read timeout - server stopped sending data") + if connection_result: + # Store connection start time to measure success duration + connection_start_time = time.time() + + # Successfully connected - read stream data until disconnect/error + self._process_stream_data() + # If we get here, the connection was closed/failed + + # Reset stream switch attempts if the connection lasted longer than threshold + # This indicates we had a stable connection for a while before failing + connection_duration = time.time() - connection_start_time + stable_connection_threshold = 30 # 30 seconds threshold + if connection_duration > stable_connection_threshold: + logger.info(f"Stream was stable for {connection_duration:.1f} seconds, resetting switch attempts counter") + stream_switch_attempts = 0 + + # Connection failed or ended - decide what to do next + if self.stop_requested or not self.running: + # Normal shutdown requested + return + + # Connection failed, increment retry count + self.retry_count += 1 self.connected = False - time.sleep(1) - except requests.RequestException as e: - logger.error(f"HTTP request error: {e}") + # If we've reached max retries, mark this URL as failed + if self.retry_count >= self.max_retries: + url_failed = True + logger.warning(f"Maximum retry attempts ({self.max_retries}) reached for URL: {self.url}") + else: + # Wait with exponential backoff before retrying + timeout = min(.25 ** self.retry_count, 3) # Cap at 3 seconds + logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count}/{self.max_retries})") + time.sleep(timeout) + + except Exception as e: + logger.error(f"Connection error: {e}", exc_info=True) + self.retry_count += 1 self.connected = False - time.sleep(5) - finally: - # Clean up response and session - if self.current_response: - try: - self.current_response.close() - except Exception as e: - logger.debug(f"Error closing response: {e}") - self.current_response = None + if self.retry_count >= self.max_retries: + url_failed = True + else: + # Wait with exponential backoff before retrying + timeout = min(2 ** self.retry_count, 10) + logger.info(f"Reconnecting in {timeout} seconds after error... (attempt {self.retry_count}/{self.max_retries})") + time.sleep(timeout) - if self.current_session: - try: - self.current_session.close() - except Exception as e: - logger.debug(f"Error closing session: {e}") - self.current_session = None + # If URL failed and we're still running, try switching to another stream + if url_failed and self.running: + logger.info(f"URL {self.url} failed after {self.retry_count} attempts, trying next stream") - # Connection retry logic - if self.running and not self.connected: - self.retry_count += 1 - if self.retry_count > self.max_retries: - logger.error(f"Maximum retry attempts ({self.max_retries}) exceeded") + # Try to switch to next stream + switch_result = self._try_next_stream() + if switch_result: + # Successfully switched to a new stream, continue with the new URL + stream_switch_attempts += 1 + logger.info(f"Successfully switched to new URL: {self.url} (switch attempt {stream_switch_attempts}/{max_stream_switches})") + # Reset retry count for the new stream - important for the loop to work correctly + self.retry_count = 0 + # Continue outer loop with new URL - DON'T add a break statement here + else: + # No more streams to try + logger.error(f"Failed to find alternative streams after {stream_switch_attempts} attempts") break - - timeout = min(2 ** self.retry_count, 30) - - # When a connection fails and reconnect is needed: - self.reconnecting = True - - # Cancel all existing buffer timers during reconnect - for timer in list(self._buffer_check_timers): - try: - if timer and timer.is_alive(): - timer.cancel() - except Exception as e: - logger.error(f"Error canceling buffer timer: {e}") - self._buffer_check_timers = [] - - logger.info(f"Reconnecting in {timeout} seconds... (attempt {self.retry_count})") - time.sleep(timeout) - - self.reconnecting = False # Reset flag after sleep + elif not self.running: + # Normal shutdown was requested + break except Exception as e: logger.error(f"Stream error: {e}", exc_info=True) finally: self.connected = False - - if self.socket: - try: - self._close_socket() - except: - pass - - if self.current_response: - try: - self.current_response.close() - except: - pass - - if self.current_session: - try: - self.current_session.close() - except: - pass - + self._close_all_connections() logger.info(f"Stream manager stopped") + def _establish_transcode_connection(self): + """Establish a connection using transcoding""" + try: + logger.debug(f"Building transcode command for channel {self.channel_id}") + channel = get_object_or_404(Channel, uuid=self.channel_id) + + # Use FFmpeg specifically for HLS streams + if hasattr(self, 'force_ffmpeg') and self.force_ffmpeg: + from core.models import StreamProfile + try: + stream_profile = StreamProfile.objects.get(name='ffmpeg', locked=True) + logger.info("Using FFmpeg stream profile for HLS content") + except StreamProfile.DoesNotExist: + # Fall back to channel's profile if FFmpeg not found + stream_profile = channel.get_stream_profile() + logger.warning("FFmpeg profile not found, using channel default profile") + else: + stream_profile = channel.get_stream_profile() + + # Build and start transcode command + self.transcode_cmd = stream_profile.build_command(self.url, self.user_agent) + logger.debug(f"Starting transcode process: {self.transcode_cmd}") + + self.transcode_process = subprocess.Popen( + self.transcode_cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, # Suppress FFmpeg logs + bufsize=188 * 64 # Buffer optimized for TS packets + ) + + self.socket = self.transcode_process.stdout # Read from FFmpeg output + self.connected = True + + # Set channel state to waiting for clients + self._set_waiting_for_clients() + + return True + except Exception as e: + logger.error(f"Error establishing transcode connection: {e}", exc_info=True) + self._close_socket() + return False + + def _establish_http_connection(self): + """Establish a direct HTTP connection to the stream""" + try: + logger.debug(f"Using TS Proxy to connect to stream: {self.url}") + + # Create new session for each connection attempt + session = self._create_session() + self.current_session = session + + # Stream the URL with proper timeout handling + response = session.get( + self.url, + stream=True, + timeout=(10, 60) # 10s connect timeout, 60s read timeout + ) + self.current_response = response + + if response.status_code == 200: + self.connected = True + self.healthy = True + logger.info(f"Successfully connected to stream source") + + # Set channel state to waiting for clients + self._set_waiting_for_clients() + + return True + else: + logger.error(f"Failed to connect to stream: HTTP {response.status_code}") + self._close_connection() + return False + except requests.exceptions.RequestException as e: + logger.error(f"HTTP request error: {e}") + self._close_connection() + return False + except Exception as e: + logger.error(f"Error establishing HTTP connection: {e}", exc_info=True) + self._close_connection() + return False + + def _process_stream_data(self): + """Process stream data until disconnect or error""" + try: + if self.transcode: + # Handle transcoded stream data + while self.running and self.connected: + if self.fetch_chunk(): + self.last_data_time = time.time() + else: + if not self.running: + break + time.sleep(0.1) + else: + # Handle direct HTTP connection + chunk_count = 0 + try: + for chunk in self.current_response.iter_content(chunk_size=self.chunk_size): + # Check if we've been asked to stop + if self.stop_requested or self.url_switching: + break + + if chunk: + # Add chunk to buffer with TS packet alignment + success = self.buffer.add_chunk(chunk) + + if success: + self.last_data_time = time.time() + chunk_count += 1 + + # Update last data timestamp in Redis + if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + last_data_key = RedisKeys.last_data(self.buffer.channel_id) + self.buffer.redis_client.set(last_data_key, str(time.time()), ex=60) + except (AttributeError, ConnectionError) as e: + if self.stop_requested or self.url_switching: + logger.debug(f"Expected connection error during shutdown/URL switch: {e}") + else: + logger.error(f"Unexpected stream error: {e}") + raise + except Exception as e: + logger.error(f"Error processing stream data: {e}", exc_info=True) + + # If we exit the loop, connection is closed or failed + self.connected = False + + def _close_all_connections(self): + """Close all connection resources""" + if self.socket: + try: + self._close_socket() + except Exception as e: + logger.debug(f"Error closing socket: {e}") + + if self.current_response: + try: + self.current_response.close() + except Exception as e: + logger.debug(f"Error closing response: {e}") + + if self.current_session: + try: + self.current_session.close() + except Exception as e: + logger.debug(f"Error closing session: {e}") + + # Clear references + self.socket = None + self.current_response = None + self.current_session = None + self.transcode_process = None + def stop(self): """Stop the stream manager and cancel all timers""" # Add at the beginning of your stop method @@ -359,6 +452,9 @@ class StreamManager: # Reset retry counter to allow immediate reconnect self.retry_count = 0 + # Reset tried streams when manually switching URL + self.tried_stream_ids = set() + # Also reset buffer position to prevent stale data after URL change if hasattr(self.buffer, 'reset_buffer_position'): try: @@ -582,3 +678,89 @@ class StreamManager: except Exception as e: logger.error(f"Error in buffer check: {e}") + def _try_next_stream(self): + """ + Try to switch to the next available stream for this channel. + + Returns: + bool: True if successfully switched to a new stream, False otherwise + """ + try: + logger.info(f"Trying to find alternative stream for channel {self.channel_id}, current stream ID: {self.current_stream_id}") + + # Get alternate streams excluding the current one + alternate_streams = get_alternate_streams(self.channel_id, self.current_stream_id) + logger.info(f"Found {len(alternate_streams)} potential alternate streams for channel {self.channel_id}") + + # Filter out streams we've already tried + untried_streams = [s for s in alternate_streams if s['stream_id'] not in self.tried_stream_ids] + if untried_streams: + ids_to_try = ', '.join([str(s['stream_id']) for s in untried_streams]) + logger.info(f"Found {len(untried_streams)} untried streams for channel {self.channel_id}: [{ids_to_try}]") + else: + logger.warning(f"No untried streams available for channel {self.channel_id}, tried: {self.tried_stream_ids}") + + if not untried_streams: + # Check if we have streams but they've all been tried + if alternate_streams and len(self.tried_stream_ids) > 0: + logger.warning(f"All {len(alternate_streams)} alternate streams have been tried for channel {self.channel_id}") + return False + + # Get the next stream to try + next_stream = untried_streams[0] + stream_id = next_stream['stream_id'] + + # Add to tried streams + self.tried_stream_ids.add(stream_id) + + # Get stream info including URL + logger.info(f"Trying next stream ID {stream_id} for channel {self.channel_id}") + stream_info = get_stream_info_for_switch(self.channel_id, stream_id) + + if 'error' in stream_info or not stream_info.get('url'): + logger.error(f"Error getting info for stream {stream_id}: {stream_info.get('error', 'No URL')}") + return False + + # Update URL and user agent + new_url = stream_info['url'] + new_user_agent = stream_info['user_agent'] + new_transcode = stream_info['transcode'] + + logger.info(f"Switching from URL {self.url} to {new_url} for channel {self.channel_id}") + + # Update stream ID tracking + self.current_stream_id = stream_id + + # Store the new user agent and transcode settings + self.user_agent = new_user_agent + self.transcode = new_transcode + + # Update stream metadata in Redis + if hasattr(self.buffer, 'redis_client') and self.buffer.redis_client: + metadata_key = RedisKeys.channel_metadata(self.channel_id) + self.buffer.redis_client.hset(metadata_key, mapping={ + "url": new_url, + "user_agent": new_user_agent, + "profile": stream_info['profile'], + "stream_id": str(stream_id), + "stream_switch_time": str(time.time()), + "stream_switch_reason": "max_retries_exceeded" + }) + + # Log the switch + logger.info(f"Stream metadata updated for channel {self.channel_id} to stream ID {stream_id}") + + # IMPORTANT: Just update the URL, don't stop the channel or release resources + switch_result = self.update_url(new_url) + if not switch_result: + logger.error(f"Failed to update URL for stream ID {stream_id}") + return False + + logger.info(f"Successfully switched to stream ID {stream_id} with URL {new_url}") + return True + + except Exception as e: + logger.error(f"Error trying next stream for channel {self.channel_id}: {e}", exc_info=True) + return False + + diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index cbce957c..e478c1c0 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -4,7 +4,7 @@ Utilities for handling stream URLs and transformations. import logging import re -from typing import Optional, Tuple +from typing import Optional, Tuple, List from django.shortcuts import get_object_or_404 from apps.channels.models import Channel, Stream from apps.m3u.models import M3UAccount, M3UAccountProfile @@ -100,30 +100,147 @@ def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] Returns: dict: Stream information including URL, user agent and transcode flag """ - channel = get_object_or_404(Channel, uuid=channel_id) + try: + channel = get_object_or_404(Channel, uuid=channel_id) - # Use the target stream if specified, otherwise use current stream - if target_stream_id: - stream_id = target_stream_id - # Find a compatible profile for this stream - profiles = M3UAccountProfile.objects.filter(stream=stream_id) - if not profiles.exists(): - logger.error(f"No profile found for stream {stream_id}") - return {'error': 'No profile found for stream'} - profile_id = profiles.first().id - else: - stream_id, profile_id = channel.get_stream() - if stream_id is None or profile_id is None: - return {'error': 'No stream assigned to channel'} + # Use the target stream if specified, otherwise use current stream + if target_stream_id: + stream_id = target_stream_id - # Generate the URL using our utility - stream_url, user_agent, transcode, profile_value = generate_stream_url(channel_id) + # Get the stream object + stream = get_object_or_404(Stream, pk=stream_id) - return { - 'url': stream_url, - 'user_agent': user_agent, - 'transcode': transcode, - 'profile': profile_value, - 'stream_id': stream_id, - 'profile_id': profile_id - } + # Find compatible profile for this stream + profiles = M3UAccountProfile.objects.filter(m3u_account=stream.m3u_account) + + if not profiles.exists(): + # Try to get default profile + default_profile = M3UAccountProfile.objects.filter( + m3u_account=stream.m3u_account, + is_default=True + ).first() + + if default_profile: + profile_id = default_profile.id + else: + logger.error(f"No profile found for stream {stream_id}") + return {'error': 'No profile found for stream'} + else: + # Use first available profile + profile_id = profiles.first().id + else: + stream_id, profile_id = channel.get_stream() + if stream_id is None or profile_id is None: + return {'error': 'No stream assigned to channel'} + + # Get the stream and profile objects directly + stream = get_object_or_404(Stream, pk=stream_id) + profile = get_object_or_404(M3UAccountProfile, pk=profile_id) + + # Get the user agent from the M3U account + m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id) + user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent + if not user_agent: + user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id()).user_agent + + # Generate URL using the transform function directly + stream_url = transform_url(stream.url, profile.search_pattern, profile.replace_pattern) + + # Get transcode info from the channel's stream profile + stream_profile = channel.get_stream_profile() + transcode = not (stream_profile.is_proxy() or stream_profile is None) + profile_value = str(stream_profile) + + return { + 'url': stream_url, + 'user_agent': user_agent, + 'transcode': transcode, + 'profile': profile_value, + 'stream_id': stream_id, + 'profile_id': profile_id + } + except Exception as e: + logger.error(f"Error getting stream info for switch: {e}", exc_info=True) + return {'error': f'Error: {str(e)}'} + +def get_alternate_streams(channel_id: str, current_stream_id: Optional[int] = None) -> List[dict]: + """ + Get alternative streams for a channel when the current stream fails. + + Args: + channel_id: The UUID of the channel + current_stream_id: The currently failing stream ID to exclude + + Returns: + List[dict]: List of stream information dictionaries with stream_id and profile_id + """ + try: + # Get channel object + channel = get_object_or_404(Channel, uuid=channel_id) + logger.debug(f"Looking for alternate streams for channel {channel_id}, current stream ID: {current_stream_id}") + + # Get all assigned streams for this channel + streams = channel.streams.all() + logger.debug(f"Channel {channel_id} has {streams.count()} total assigned streams") + + if not streams.exists(): + logger.warning(f"No streams assigned to channel {channel_id}") + return [] + + alternate_streams = [] + + # Process each stream + for stream in streams: + # Log each stream we're checking + logger.debug(f"Checking stream ID {stream.id} ({stream.name}) for channel {channel_id}") + + # Skip the current failing stream + if current_stream_id and stream.id == current_stream_id: + logger.debug(f"Skipping current stream ID {current_stream_id}") + continue + + # Find compatible profiles for this stream + # FIX: Looking at the error message, M3UAccountProfile doesn't have a 'stream' field + # We need to find which field relates M3UAccountProfile to Stream + try: + # Check if we can find profiles via m3u_account + profiles = M3UAccountProfile.objects.filter(m3u_account=stream.m3u_account) + if not profiles.exists(): + logger.debug(f"No profiles found via m3u_account for stream {stream.id}") + # Fallback to the default profile of the account + default_profile = M3UAccountProfile.objects.filter( + m3u_account=stream.m3u_account, + is_default=True + ).first() + if default_profile: + profiles = [default_profile] + else: + logger.warning(f"No default profile found for m3u_account {stream.m3u_account.id}") + continue + + # Get first compatible profile + profile = profiles.first() + if profile: + logger.debug(f"Found compatible profile ID {profile.id} for stream ID {stream.id}") + + alternate_streams.append({ + 'stream_id': stream.id, + 'profile_id': profile.id, + 'name': stream.name + }) + else: + logger.debug(f"No compatible profile found for stream ID {stream.id}") + except Exception as inner_e: + logger.error(f"Error finding profiles for stream {stream.id}: {inner_e}") + continue + + if alternate_streams: + stream_ids = ', '.join([str(s['stream_id']) for s in alternate_streams]) + logger.info(f"Found {len(alternate_streams)} alternate streams for channel {channel_id}: [{stream_ids}]") + else: + logger.warning(f"No alternate streams found for channel {channel_id}") + + return alternate_streams + except Exception as e: + logger.error(f"Error getting alternate streams for channel {channel_id}: {e}", exc_info=True) + return [] diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index f84963bb..3b4a966c 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -58,6 +58,10 @@ def stream_ts(request, channel_id): if stream_url is None: return JsonResponse({'error': 'Channel not available'}, status=404) + # Get the stream ID from the channel + stream_id, profile_id = channel.get_stream() + logger.info(f"Channel {channel_id} using stream ID {stream_id}, profile ID {profile_id}") + # Generate transcode command if needed stream_profile = channel.get_stream_profile() if stream_profile.is_redirect(): @@ -65,7 +69,7 @@ def stream_ts(request, channel_id): # Initialize channel with the stream's user agent (not the client's) success = ChannelService.initialize_channel( - channel_id, stream_url, stream_user_agent, transcode, profile_value + channel_id, stream_url, stream_user_agent, transcode, profile_value, stream_id ) if not success: