merged in main

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
+

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)

apps/proxy/ts_proxy/redis_keys.py

@@ -73,3 +73,8 @@ class RedisKeys:
     def switch_status(channel_id):
         """Key for stream switch status"""
         return f"ts_proxy:channel:{channel_id}:switch_status"
+
+    @staticmethod
+    def worker_heartbeat(worker_id):
+        """Key for worker heartbeat"""
+        return f"ts_proxy:worker:{worker_id}:heartbeat"

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
@@ -468,7 +498,10 @@ class ProxyServer:
             return False
 
     def check_if_channel_exists(self, channel_id):
-        """Check if a channel exists using standardized key structure"""
+        """
+        Check if a channel exists and is in a valid state.
+        Enhanced to detect zombie channels after server restarts.
+        """
         # Check local memory first
         if channel_id in self.stream_managers or channel_id in self.stream_buffers:
             return True
@@ -478,9 +511,48 @@ class ProxyServer:
             # Primary check - look for channel metadata
             metadata_key = RedisKeys.channel_metadata(channel_id)
 
-            # If metadata exists, return true
+            # If metadata exists, validate it's in a healthy state
             if self.redis_client.exists(metadata_key):
-                return True
+                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')
+
+                # States that indicate the channel is running properly
+                valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING]
+
+                # If the channel is in a valid state, check if the owner is still active
+                if state in valid_states:
+                    # Check if owner still exists by checking heartbeat
+                    owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
+                    owner_alive = self.redis_client.exists(owner_heartbeat_key)
+
+                    if owner_alive:
+                        return True
+                    else:
+                        # This is a zombie channel - owner is gone but metadata still exists
+                        logger.warning(f"Detected zombie channel {channel_id} - owner {owner} is no longer active")
+                        self._clean_zombie_channel(channel_id, metadata)
+                        return False
+                elif state in [ChannelState.STOPPING, ChannelState.STOPPED, ChannelState.ERROR]:
+                    # These states indicate the channel should be reinitialized
+                    logger.info(f"Channel {channel_id} exists but in terminal state: {state}")
+                    return False
+                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'))
+                        state_age = time.time() - state_changed_at
+
+                        # If in initializing state for too long, consider it stale
+                        if state_age > 60:  # 60 seconds threshold
+                            logger.warning(f"Channel {channel_id} stuck in {state} state for {state_age:.1f}s - treating as zombie")
+                            self._clean_zombie_channel(channel_id, metadata)
+                            return False
+
+                    # Otherwise assume it's still in progress
+                    return True
 
             # Additional checks if metadata doesn't exist
             additional_keys = [
@@ -491,10 +563,41 @@ class ProxyServer:
 
             for key in additional_keys:
                 if self.redis_client.exists(key):
-                    return True
+                    # Found orphaned keys without metadata - clean them up
+                    logger.warning(f"Found orphaned keys for channel {channel_id} without metadata - cleaning up")
+                    self._clean_redis_keys(channel_id)
+                    return False
 
         return False
 
+    def _clean_zombie_channel(self, channel_id, metadata=None):
+        """Clean up a zombie channel (channel with Redis keys but no active owner)"""
+        try:
+            logger.info(f"Cleaning up zombie channel {channel_id}")
+
+            # 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')
+                logger.info(f"Zombie channel details - state: {state}, owner: {owner}")
+
+            # Clean up Redis keys
+            self._clean_redis_keys(channel_id)
+
+            # Force release resources in the Channel model
+            try:
+                from apps.channels.models import Channel
+                channel = Channel.objects.get(uuid=channel_id)
+                channel.release_stream()
+                logger.info(f"Released stream allocation for zombie channel {channel_id}")
+            except Exception as e:
+                logger.error(f"Error releasing stream for zombie channel {channel_id}: {e}")
+
+            return True
+        except Exception as e:
+            logger.error(f"Error cleaning zombie channel {channel_id}: {e}", exc_info=True)
+            return False
+
     def stop_channel(self, channel_id):
         """Stop a channel with proper ownership handling"""
         try:
@@ -610,6 +713,11 @@ class ProxyServer:
         def cleanup_task():
             while True:
                 try:
+                    # Send worker heartbeat first
+                    if self.redis_client:
+                        worker_heartbeat_key = f"ts_proxy:worker:{self.worker_id}:heartbeat"
+                        self.redis_client.setex(worker_heartbeat_key, 30, str(time.time()))
+
                     # Refresh channel registry
                     self.refresh_channel_registry()
 

apps/proxy/ts_proxy/services/channel_service.py

@@ -11,7 +11,8 @@ from apps.channels.models import Channel
 from apps.proxy.config import TSConfig as Config
 from .. import proxy_server
 from ..redis_keys import RedisKeys
-from ..constants import EventType
+from ..constants import EventType, ChannelState
+from ..url_utils import get_stream_info_for_switch
 
 logger = logging.getLogger("ts_proxy")
 
@@ -19,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.
 
@@ -29,32 +30,78 @@ 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
 
     @staticmethod
-    def change_stream_url(channel_id, new_url, user_agent=None):
+    def change_stream_url(channel_id, new_url=None, user_agent=None, target_stream_id=None):
         """
         Change the URL of an existing stream.
 
         Args:
             channel_id: UUID of the channel
-            new_url: New stream URL
+            new_url: New stream URL (optional if target_stream_id is provided)
             user_agent: Optional user agent to update
+            target_stream_id: Optional target stream ID to switch to
 
         Returns:
             dict: Result information including success status and diagnostics
         """
+        # If no direct URL is provided but a target stream is, get URL from target stream
+        if not new_url and target_stream_id:
+            stream_info = get_stream_info_for_switch(channel_id, target_stream_id)
+            if 'error' in stream_info:
+                return {
+                    'status': 'error',
+                    'message': stream_info['error']
+                }
+            new_url = stream_info['url']
+            user_agent = stream_info['user_agent']
+
         # Check if channel exists
         in_local_managers = channel_id in proxy_server.stream_managers
         in_local_buffers = channel_id in proxy_server.stream_buffers
@@ -271,6 +318,69 @@ class ChannelService:
             'event_published': event_published
         }
 
+    @staticmethod
+    def validate_channel_state(channel_id):
+        """
+        Validate if a channel is in a healthy state and has an active owner.
+
+        Args:
+            channel_id: UUID of the channel
+
+        Returns:
+            tuple: (valid, state, owner, details) - validity status, current state, owner, and diagnostic info
+        """
+        if not proxy_server.redis_client:
+            return False, None, None, {"error": "Redis not available"}
+
+        try:
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            if not proxy_server.redis_client.exists(metadata_key):
+                return False, None, None, {"error": "No channel metadata"}
+
+            metadata = proxy_server.redis_client.hgetall(metadata_key)
+
+            # Extract state and owner
+            state = metadata.get(b'state', b'unknown').decode('utf-8')
+            owner = metadata.get(b'owner', b'unknown').decode('utf-8')
+
+            # Valid states indicate channel is running properly
+            valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING]
+
+            if state not in valid_states:
+                return False, state, owner, {"error": f"Invalid state: {state}"}
+
+            # Check if owner is still active
+            owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
+            owner_alive = proxy_server.redis_client.exists(owner_heartbeat_key)
+
+            if not owner_alive:
+                return False, state, owner, {"error": "Owner not active"}
+
+            # Check for recent activity
+            last_data_key = RedisKeys.last_data(channel_id)
+            last_data = proxy_server.redis_client.get(last_data_key)
+
+            details = {
+                "state": state,
+                "owner": owner,
+                "owner_alive": owner_alive
+            }
+
+            if last_data:
+                last_data_time = float(last_data.decode('utf-8'))
+                data_age = time.time() - last_data_time
+                details["last_data_age"] = data_age
+
+                # If no data for too long, consider invalid
+                if data_age > 30:  # 30 seconds threshold
+                    return False, state, owner, {"error": f"No data for {data_age:.1f}s", **details}
+
+            return True, state, owner, details
+
+        except Exception as e:
+            logger.error(f"Error validating channel state: {e}", exc_info=True)
+            return False, None, None, {"error": f"Exception: {str(e)}"}
+
     # Helper methods for Redis operations
 
     @staticmethod

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,20 +114,16 @@ 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
-            stream_type = detect_stream_type(self.url)
-            if self.transcode == False and stream_type == StreamType.HLS:
-                logger.info(f"Detected HLS stream: {self.url}")
-                logger.info(f"HLS streams will be handled with FFmpeg for now - future version will support HLS natively")
-                # Enable transcoding for HLS streams
-                self.transcode = True
-                # We'll override the stream profile selection with ffmpeg in the transcoding section
-                self.force_ffmpeg = True
+
 
             # Start health monitor thread
             health_thread = threading.Thread(target=self._monitor_health, daemon=True)
@@ -102,200 +131,265 @@ 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:
+                # Check stream type before connecting
+                stream_type = detect_stream_type(self.url)
+                if self.transcode == False and stream_type == StreamType.HLS:
+                    logger.info(f"Detected HLS stream: {self.url}")
+                    logger.info(f"HLS streams will be handled with FFmpeg for now - future version will support HLS natively")
+                    # Enable transcoding for HLS streams
+                    self.transcode = True
+                    # We'll override the stream profile selection with ffmpeg in the transcoding section
+                    self.force_ffmpeg = True
+                # 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 +453,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 +679,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
+
+

apps/proxy/ts_proxy/url_utils.py

@@ -0,0 +1,246 @@
+"""
+Utilities for handling stream URLs and transformations.
+"""
+
+import logging
+import re
+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
+from core.models import UserAgent, CoreSettings
+
+logger = logging.getLogger("ts_proxy")
+
+def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]:
+    """
+    Generate the appropriate stream URL for a channel based on its profile settings.
+
+    Args:
+        channel_id: The UUID of the channel
+
+    Returns:
+        Tuple[str, str, bool]: (stream_url, user_agent, transcode_flag)
+    """
+    # Get channel and related objects
+    channel = get_object_or_404(Channel, uuid=channel_id)
+    stream_id, profile_id = channel.get_stream()
+
+    if stream_id is None or profile_id is None:
+        logger.error(f"No stream assigned to channel {channel_id}")
+        return None, None, False
+
+    # Get the M3U account profile for URL pattern
+    stream = get_object_or_404(Stream, pk=stream_id)
+    profile = get_object_or_404(M3UAccountProfile, pk=profile_id)
+
+    # Get the appropriate user agent
+    m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id)
+    stream_user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent
+
+    if stream_user_agent is None:
+        stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
+        logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
+
+    # Generate stream URL based on the selected profile
+    input_url = stream.url
+    stream_url = transform_url(input_url, profile.search_pattern, profile.replace_pattern)
+
+    # Check if transcoding is needed
+    stream_profile = channel.get_stream_profile()
+    if stream_profile.is_proxy() or stream_profile is None:
+        transcode = False
+    else:
+        transcode = True
+
+    # Get profile name as string
+    profile_value = str(stream_profile)
+
+    return stream_url, stream_user_agent, transcode, profile_value
+
+def transform_url(input_url: str, search_pattern: str, replace_pattern: str) -> str:
+    """
+    Transform a URL using regex pattern replacement.
+
+    Args:
+        input_url: The base URL to transform
+        search_pattern: The regex search pattern
+        replace_pattern: The replacement pattern
+
+    Returns:
+        str: The transformed URL
+    """
+    try:
+        logger.debug("Executing URL pattern replacement:")
+        logger.debug(f"  base URL: {input_url}")
+        logger.debug(f"  search: {search_pattern}")
+
+        # Handle backreferences in the replacement pattern
+        safe_replace_pattern = re.sub(r'\$(\d+)', r'\\\1', replace_pattern)
+        logger.debug(f"  replace: {replace_pattern}")
+        logger.debug(f"  safe replace: {safe_replace_pattern}")
+
+        # Apply the transformation
+        stream_url = re.sub(search_pattern, safe_replace_pattern, input_url)
+        logger.debug(f"Generated stream url: {stream_url}")
+
+        return stream_url
+    except Exception as e:
+        logger.error(f"Error transforming URL: {e}")
+        return input_url  # Return original URL on error
+
+def get_stream_info_for_switch(channel_id: str, target_stream_id: Optional[int] = None) -> dict:
+    """
+    Get stream information for a channel switch, optionally to a specific stream ID.
+
+    Args:
+        channel_id: The UUID of the channel
+        target_stream_id: Optional specific stream ID to switch to
+
+    Returns:
+        dict: Stream information including URL, user agent and transcode flag
+    """
+    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
+
+            # Get the stream object
+            stream = get_object_or_404(Stream, pk=stream_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 []

apps/proxy/ts_proxy/views.py

@@ -21,6 +21,7 @@ from rest_framework.permissions import IsAuthenticated
 from .constants import ChannelState, EventType, StreamType
 from .config_helper import ConfigHelper
 from .services.channel_service import ChannelService
+from .url_utils import generate_stream_url, transform_url, get_stream_info_for_switch
 
 # Configure logging properly
 logger = logging.getLogger("ts_proxy")
@@ -46,58 +47,58 @@ def stream_ts(request, channel_id):
                 logger.debug(f"[{client_id}] Client connected with user agent: {client_user_agent}")
                 break
 
+        # Check if we need to reinitialize the channel
+        needs_initialization = True
+        channel_state = None
+
+        # Get current channel state from Redis if available
+        if proxy_server.redis_client:
+            metadata_key = RedisKeys.channel_metadata(channel_id)
+            if proxy_server.redis_client.exists(metadata_key):
+                metadata = proxy_server.redis_client.hgetall(metadata_key)
+                if b'state' in metadata:
+                    channel_state = metadata[b'state'].decode('utf-8')
+
+                    # Only skip initialization if channel is in a healthy state
+                    valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS]
+                    if channel_state in valid_states:
+                        # Verify the owner is still active
+                        if b'owner' in metadata:
+                            owner = metadata[b'owner'].decode('utf-8')
+                            owner_heartbeat_key = f"ts_proxy:worker:{owner}:heartbeat"
+                            if proxy_server.redis_client.exists(owner_heartbeat_key):
+                                # Owner is active and channel is in good state
+                                needs_initialization = False
+                                logger.info(f"[{client_id}] Channel {channel_id} in state {channel_state} with active owner {owner}")
+
         # Start initialization if needed
         channel_initializing = False
-        if not proxy_server.check_if_channel_exists(channel_id):
+        if needs_initialization or not proxy_server.check_if_channel_exists(channel_id):
+            # Force cleanup of any previous instance
+            if channel_state in [ChannelState.ERROR, ChannelState.STOPPING, ChannelState.STOPPED]:
+                logger.warning(f"[{client_id}] Channel {channel_id} in state {channel_state}, forcing cleanup")
+                proxy_server.stop_channel(channel_id)
+
             # Initialize the channel (but don't wait for completion)
             logger.info(f"[{client_id}] Starting channel {channel_id} initialization")
 
-            # Get stream details from channel model
-            stream_id, profile_id = channel.get_stream()
-            if stream_id is None or profile_id is None:
+            # Use the utility function to get stream URL and settings
+            stream_url, stream_user_agent, transcode, profile_value = generate_stream_url(channel_id)
+            if stream_url is None:
                 return JsonResponse({'error': 'Channel not available'}, status=404)
 
-            # Load in necessary objects for the stream
-            logger.info(f"Fetching stream ID {stream_id}")
-            stream = get_object_or_404(Stream, pk=stream_id)
-            logger.info(f"Fetching profile ID {profile_id}")
-            profile = get_object_or_404(M3UAccountProfile, pk=profile_id)
-
-            # Load in the user-agent for the STREAM connection (not client)
-            m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id)
-            stream_user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent
-            if stream_user_agent is None:
-                stream_user_agent = UserAgent.objects.get(id=CoreSettings.get_default_user_agent_id())
-                logger.debug(f"No user agent found for account, using default: {stream_user_agent}")
-            else:
-                logger.debug(f"User agent found for account: {stream_user_agent}")
-
-            # Generate stream URL based on the selected profile
-            input_url = stream.url
-            logger.debug("Executing the following pattern replacement:")
-            logger.debug(f"  search: {profile.search_pattern}")
-            safe_replace_pattern = re.sub(r'\$(\d+)', r'\\\1', profile.replace_pattern)
-            logger.debug(f"  replace: {profile.replace_pattern}")
-            logger.debug(f"  safe replace: {safe_replace_pattern}")
-            stream_url = re.sub(profile.search_pattern, safe_replace_pattern, input_url)
-            logger.debug(f"Generated stream url: {stream_url}")
+            # 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():
                 return HttpResponseRedirect(stream_url)
 
-            # Need to check if profile is transcoded
-            logger.debug(f"Using profile {stream_profile} for stream {stream_id}")
-            if stream_profile.is_proxy() or stream_profile is None:
-                transcode = False
-            else:
-                transcode = True
-
-            # Initialize channel using the service
-            profile_value = stream_profile.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: