diff --git a/apps/proxy/ts_proxy/channel_status.py b/apps/proxy/ts_proxy/channel_status.py index 9338fb7f..799da605 100644 --- a/apps/proxy/ts_proxy/channel_status.py +++ b/apps/proxy/ts_proxy/channel_status.py @@ -3,7 +3,7 @@ import time import re from . import proxy_server from .redis_keys import RedisKeys -from .constants import TS_PACKET_SIZE +from .constants import TS_PACKET_SIZE, ChannelMetadataField from redis.exceptions import ConnectionError, TimeoutError from .utils import get_logger @@ -35,28 +35,32 @@ class ChannelStatus: info = { 'channel_id': channel_id, - 'state': metadata.get(b'state', b'unknown').decode('utf-8'), - 'url': metadata.get(b'url', b'').decode('utf-8'), - 'profile': metadata.get(b'profile', b'unknown').decode('utf-8'), - 'started_at': metadata.get(b'init_time', b'0').decode('utf-8'), - 'owner': metadata.get(b'owner', b'unknown').decode('utf-8'), + 'state': metadata.get(ChannelMetadataField.STATE.encode('utf-8'), b'unknown').decode('utf-8'), + 'url': metadata.get(ChannelMetadataField.URL.encode('utf-8'), b'').decode('utf-8'), + 'profile': metadata.get(ChannelMetadataField.STREAM_PROFILE.encode('utf-8'), + metadata.get(ChannelMetadataField.PROFILE.encode('utf-8'), b'unknown')).decode('utf-8'), + 'started_at': metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0').decode('utf-8'), + 'owner': metadata.get(ChannelMetadataField.OWNER.encode('utf-8'), b'unknown').decode('utf-8'), 'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0, } # Add timing information - if b'state_changed_at' in metadata: - state_changed_at = float(metadata[b'state_changed_at'].decode('utf-8')) + state_changed_field = ChannelMetadataField.STATE_CHANGED_AT.encode('utf-8') + if state_changed_field in metadata: + state_changed_at = float(metadata[state_changed_field].decode('utf-8')) info['state_changed_at'] = state_changed_at info['state_duration'] = time.time() - state_changed_at - if b'init_time' in metadata: - created_at = float(metadata[b'init_time'].decode('utf-8')) + init_time_field = ChannelMetadataField.INIT_TIME.encode('utf-8') + if init_time_field in metadata: + created_at = float(metadata[init_time_field].decode('utf-8')) info['started_at'] = created_at info['uptime'] = time.time() - created_at # Add data throughput information - if b'total_bytes' in metadata: - total_bytes = int(metadata[b'total_bytes'].decode('utf-8')) + total_bytes_field = ChannelMetadataField.TOTAL_BYTES.encode('utf-8') + if total_bytes_field in metadata: + total_bytes = int(metadata[total_bytes_field].decode('utf-8')) info['total_bytes'] = total_bytes # Format total bytes in human-readable form @@ -87,7 +91,7 @@ class ChannelStatus: for client_id in client_ids: client_id_str = client_id.decode('utf-8') - client_key = f"ts_proxy:channel:{channel_id}:clients:{client_id_str}" + client_key = RedisKeys.client_metadata(channel_id, client_id_str) client_data = proxy_server.redis_client.hgetall(client_key) if client_data: @@ -261,23 +265,23 @@ class ChannelStatus: client_count = proxy_server.redis_client.scard(client_set_key) or 0 # Calculate uptime - created_at = float(metadata.get(b'init_time', b'0').decode('utf-8')) + created_at = float(metadata.get(ChannelMetadataField.INIT_TIME.encode('utf-8'), b'0').decode('utf-8')) uptime = time.time() - created_at if created_at > 0 else 0 # Simplified info info = { 'channel_id': channel_id, - 'state': metadata.get(b'state', b'unknown').decode('utf-8'), - 'url': metadata.get(b'url', b'').decode('utf-8'), - 'profile': metadata.get(b'profile', b'unknown').decode('utf-8'), - 'owner': metadata.get(b'owner', b'unknown').decode('utf-8'), + 'state': metadata.get(ChannelMetadataField.STATE.encode('utf-8'), b'unknown').decode('utf-8'), + 'url': metadata.get(ChannelMetadataField.URL.encode('utf-8'), b'').decode('utf-8'), + 'profile': metadata.get(ChannelMetadataField.PROFILE.encode('utf-8'), b'unknown').decode('utf-8'), + 'owner': metadata.get(ChannelMetadataField.OWNER.encode('utf-8'), b'unknown').decode('utf-8'), 'buffer_index': int(buffer_index_value.decode('utf-8')) if buffer_index_value else 0, 'client_count': client_count, 'uptime': uptime } # Add data throughput information to basic info - total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, 'total_bytes') + total_bytes_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.TOTAL_BYTES.encode('utf-8')) if total_bytes_bytes: total_bytes = int(total_bytes_bytes.decode('utf-8')) info['total_bytes'] = total_bytes @@ -307,7 +311,7 @@ class ChannelStatus: # Get up to 10 clients for the basic view for client_id in list(client_ids)[:10]: client_id_str = client_id.decode('utf-8') - client_key = f"ts_proxy:channel:{channel_id}:clients:{client_id_str}" + client_key = RedisKeys.client_metadata(channel_id, client_id_str) # Efficient way - just retrieve the essentials client_info = { diff --git a/apps/proxy/ts_proxy/constants.py b/apps/proxy/ts_proxy/constants.py index 56caacbe..4827b24b 100644 --- a/apps/proxy/ts_proxy/constants.py +++ b/apps/proxy/ts_proxy/constants.py @@ -35,6 +35,45 @@ class StreamType: TS = "ts" UNKNOWN = "unknown" +# Channel metadata field names stored in Redis +class ChannelMetadataField: + # Basic fields + URL = "url" + USER_AGENT = "user_agent" + STATE = "state" + OWNER = "owner" + STREAM_ID = "stream_id" + + # Profile fields + STREAM_PROFILE = "stream_profile" + M3U_PROFILE = "m3u_profile" + + # Status and error fields + ERROR_MESSAGE = "error_message" + ERROR_TIME = "error_time" + STATE_CHANGED_AT = "state_changed_at" + INIT_TIME = "init_time" + CONNECTION_READY_TIME = "connection_ready_time" + + # Buffer and data tracking + BUFFER_CHUNKS = "buffer_chunks" + TOTAL_BYTES = "total_bytes" + + # Stream switching + STREAM_SWITCH_TIME = "stream_switch_time" + STREAM_SWITCH_REASON = "stream_switch_reason" + + # Client metadata fields + CONNECTED_AT = "connected_at" + LAST_ACTIVE = "last_active" + BYTES_SENT = "bytes_sent" + AVG_RATE_KBPS = "avg_rate_KBps" + CURRENT_RATE_KBPS = "current_rate_KBps" + IP_ADDRESS = "ip_address" + WORKER_ID = "worker_id" + CHUNKS_SENT = "chunks_sent" + STATS_UPDATED_AT = "stats_updated_at" + # TS packet constants TS_PACKET_SIZE = 188 TS_SYNC_BYTE = 0x47 diff --git a/apps/proxy/ts_proxy/services/channel_service.py b/apps/proxy/ts_proxy/services/channel_service.py index 210e4b0f..2b7363dc 100644 --- a/apps/proxy/ts_proxy/services/channel_service.py +++ b/apps/proxy/ts_proxy/services/channel_service.py @@ -11,7 +11,7 @@ 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, ChannelState +from ..constants import EventType, ChannelState, ChannelMetadataField from ..url_utils import get_stream_info_for_switch logger = logging.getLogger("ts_proxy") @@ -42,19 +42,19 @@ class ChannelService: # 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)) + proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.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), + ChannelMetadataField.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") + stream_id_value = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_ID) if stream_id_value: logger.info(f"Verified stream_id {stream_id_value.decode('utf-8')} is now set in Redis") else: @@ -68,9 +68,9 @@ class ChannelService: metadata_key = RedisKeys.channel_metadata(channel_id) update_data = {} if profile_value: - update_data["profile"] = profile_value + update_data[ChannelMetadataField.STREAM_PROFILE] = profile_value if stream_id: - update_data["stream_id"] = str(stream_id) + update_data[ChannelMetadataField.STREAM_ID] = str(stream_id) if update_data: proxy_server.redis_client.hset(metadata_key, mapping=update_data) @@ -220,8 +220,8 @@ class ChannelService: channel_info = {"state": state} # Immediately mark as stopping in metadata so clients detect it faster - proxy_server.redis_client.hset(metadata_key, "state", ChannelState.STOPPING) - proxy_server.redis_client.hset(metadata_key, "state_changed_at", str(time.time())) + proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE, ChannelState.STOPPING) + proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.STATE_CHANGED_AT, str(time.time())) except Exception as e: logger.error(f"Error fetching channel state: {e}") @@ -350,8 +350,8 @@ class ChannelService: 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') + state = metadata.get(ChannelMetadataField.STATE.encode(), b'unknown').decode('utf-8') + owner = metadata.get(ChannelMetadataField.OWNER.encode(), b'unknown').decode('utf-8') # Valid states indicate channel is running properly valid_states = [ChannelState.ACTIVE, ChannelState.WAITING_FOR_CLIENTS, ChannelState.CONNECTING] @@ -360,7 +360,7 @@ class ChannelService: 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_heartbeat_key = RedisKeys.worker_heartbeat(owner) owner_alive = proxy_server.redis_client.exists(owner_heartbeat_key) if not owner_alive: @@ -407,21 +407,21 @@ class ChannelService: # Use the appropriate method based on the key type if key_type == 'hash': - proxy_server.redis_client.hset(metadata_key, "url", url) + proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.URL, url) if user_agent: - proxy_server.redis_client.hset(metadata_key, "user_agent", user_agent) + proxy_server.redis_client.hset(metadata_key, ChannelMetadataField.USER_AGENT, user_agent) elif key_type == 'none': # Key doesn't exist yet # Create new hash with all required fields - metadata = {"url": url} + metadata = {ChannelMetadataField.URL: url} if user_agent: - metadata["user_agent"] = user_agent + metadata[ChannelMetadataField.USER_AGENT] = user_agent proxy_server.redis_client.hset(metadata_key, mapping=metadata) else: # If key exists with wrong type, delete it and recreate proxy_server.redis_client.delete(metadata_key) - metadata = {"url": url} + metadata = {ChannelMetadataField.URL: url} if user_agent: - metadata["user_agent"] = user_agent + metadata[ChannelMetadataField.USER_AGENT] = user_agent proxy_server.redis_client.hset(metadata_key, mapping=metadata) # Set switch request flag to ensure all workers see it @@ -438,7 +438,7 @@ class ChannelService: return False switch_request = { - "event": EventType.STREAM_SWITCH, # Use constant instead of string + "event": EventType.STREAM_SWITCH, "channel_id": channel_id, "url": new_url, "user_agent": user_agent, @@ -459,7 +459,7 @@ class ChannelService: return False stop_request = { - "event": EventType.CHANNEL_STOP, # Use constant instead of string + "event": EventType.CHANNEL_STOP, "channel_id": channel_id, "requester_worker_id": proxy_server.worker_id, "timestamp": time.time() @@ -480,7 +480,7 @@ class ChannelService: return False stop_request = { - "event": EventType.CLIENT_STOP, # Use constant instead of string + "event": EventType.CLIENT_STOP, "channel_id": channel_id, "client_id": client_id, "requester_worker_id": proxy_server.worker_id, diff --git a/apps/proxy/ts_proxy/stream_generator.py b/apps/proxy/ts_proxy/stream_generator.py index dc6c2fc2..abf52d3a 100644 --- a/apps/proxy/ts_proxy/stream_generator.py +++ b/apps/proxy/ts_proxy/stream_generator.py @@ -11,6 +11,7 @@ from . import proxy_server from .utils import create_ts_packet, get_logger from .redis_keys import RedisKeys from .utils import get_logger +from .constants import ChannelMetadataField logger = get_logger() @@ -298,11 +299,11 @@ class StreamGenerator: try: client_key = RedisKeys.client_metadata(self.channel_id, self.client_id) stats = { - "chunks_sent": str(self.chunks_sent), - "bytes_sent": str(self.bytes_sent), - "avg_rate_KBps": str(round(avg_rate, 1)), - "current_rate_KBps": str(round(self.current_rate, 1)), - "stats_updated_at": str(current_time) + ChannelMetadataField.CHUNKS_SENT: str(self.chunks_sent), + ChannelMetadataField.BYTES_SENT: str(self.bytes_sent), + ChannelMetadataField.AVG_RATE_KBPS: str(round(avg_rate, 1)), + ChannelMetadataField.CURRENT_RATE_KBPS: str(round(self.current_rate, 1)), + ChannelMetadataField.STATS_UPDATED_AT: str(current_time) } proxy_server.redis_client.hset(client_key, mapping=stats) # No need to set expiration as client heartbeat will refresh this key diff --git a/apps/proxy/ts_proxy/stream_manager.py b/apps/proxy/ts_proxy/stream_manager.py index db0d6bb9..0f08df97 100644 --- a/apps/proxy/ts_proxy/stream_manager.py +++ b/apps/proxy/ts_proxy/stream_manager.py @@ -15,7 +15,7 @@ from core.models import UserAgent, CoreSettings from .stream_buffer import StreamBuffer from .utils import detect_stream_type, get_logger from .redis_keys import RedisKeys -from .constants import ChannelState, EventType, StreamType, TS_PACKET_SIZE +from .constants import ChannelState, EventType, StreamType, ChannelMetadataField, TS_PACKET_SIZE from .config_helper import ConfigHelper from .url_utils import get_alternate_streams, get_stream_info_for_switch @@ -284,10 +284,10 @@ class StreamManager: # Update metadata to indicate error state update_data = { - "state": ChannelState.ERROR, - "state_changed_at": str(time.time()), - "error_message": error_message, - "error_time": str(time.time()) + ChannelMetadataField.STATE: ChannelState.ERROR, + ChannelMetadataField.STATE_CHANGED_AT: str(time.time()), + ChannelMetadataField.ERROR_MESSAGE: error_message, + ChannelMetadataField.ERROR_TIME: str(time.time()) } self.buffer.redis_client.hset(metadata_key, mapping=update_data) logger.info(f"Updated channel {self.channel_id} state to ERROR in Redis after stream failure") @@ -398,13 +398,13 @@ class StreamManager: metadata_key = RedisKeys.channel_metadata(self.channel_id) # Use hincrby to atomically increment the total_bytes field - self.buffer.redis_client.hincrby(metadata_key, "total_bytes", self.bytes_processed) + self.buffer.redis_client.hincrby(metadata_key, ChannelMetadataField.TOTAL_BYTES, self.bytes_processed) # Reset local counter after updating Redis self.bytes_processed = 0 self.last_bytes_update = now - logger.debug(f"Updated total_bytes in Redis for channel {self.channel_id}") + logger.debug(f"Updated {ChannelMetadataField.TOTAL_BYTES} in Redis for channel {self.channel_id}") except Exception as e: logger.error(f"Error updating bytes processed: {e}") @@ -743,8 +743,9 @@ class StreamManager: current_state = None try: metadata = redis_client.hgetall(metadata_key) - if metadata and b'state' in metadata: - current_state = metadata[b'state'].decode('utf-8') + state_field = ChannelMetadataField.STATE.encode('utf-8') + if metadata and state_field in metadata: + current_state = metadata[state_field].decode('utf-8') except Exception as e: logger.error(f"Error checking current state: {e}") @@ -758,8 +759,8 @@ class StreamManager: # Not enough buffer yet - set to connecting state if not already if current_state != ChannelState.CONNECTING: update_data = { - "state": ChannelState.CONNECTING, - "state_changed_at": current_time + ChannelMetadataField.STATE: ChannelState.CONNECTING, + ChannelMetadataField.STATE_CHANGED_AT: current_time } redis_client.hset(metadata_key, mapping=update_data) logger.info(f"Channel {channel_id} connected but waiting for buffer to fill: {current_buffer_index}/{initial_chunks_needed} chunks") @@ -772,10 +773,10 @@ class StreamManager: # We have enough buffer, proceed with state change update_data = { - "state": ChannelState.WAITING_FOR_CLIENTS, - "connection_ready_time": current_time, - "state_changed_at": current_time, - "buffer_chunks": str(current_buffer_index) + ChannelMetadataField.STATE: ChannelState.WAITING_FOR_CLIENTS, + ChannelMetadataField.CONNECTION_READY_TIME: current_time, + ChannelMetadataField.STATE_CHANGED_AT: current_time, + ChannelMetadataField.BUFFER_CHUNKS: str(current_buffer_index) } redis_client.hset(metadata_key, mapping=update_data) @@ -885,12 +886,12 @@ class StreamManager: 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" + ChannelMetadataField.URL: new_url, + ChannelMetadataField.USER_AGENT: new_user_agent, + ChannelMetadataField.STREAM_PROFILE: stream_info['profile'], + ChannelMetadataField.STREAM_ID: str(stream_id), + ChannelMetadataField.STREAM_SWITCH_TIME: str(time.time()), + ChannelMetadataField.STREAM_SWITCH_REASON: "max_retries_exceeded" }) # Log the switch diff --git a/apps/proxy/ts_proxy/url_utils.py b/apps/proxy/ts_proxy/url_utils.py index a0ed4476..8183c4f2 100644 --- a/apps/proxy/ts_proxy/url_utils.py +++ b/apps/proxy/ts_proxy/url_utils.py @@ -33,10 +33,10 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: # 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) + m3u_profile = get_object_or_404(M3UAccountProfile, pk=profile_id) # Get the appropriate user agent - m3u_account = M3UAccount.objects.get(id=profile.m3u_account.id) + m3u_account = M3UAccount.objects.get(id=m3u_profile.m3u_account.id) stream_user_agent = UserAgent.objects.get(id=m3u_account.user_agent.id).user_agent if stream_user_agent is None: @@ -45,7 +45,7 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: # Generate stream URL based on the selected profile input_url = stream.url - stream_url = transform_url(input_url, profile.search_pattern, profile.replace_pattern) + stream_url = transform_url(input_url, m3u_profile.search_pattern, m3u_profile.replace_pattern) # Check if transcoding is needed stream_profile = channel.get_stream_profile() @@ -54,7 +54,8 @@ def generate_stream_url(channel_id: str) -> Tuple[str, str, bool]: else: transcode = True - # Get profile name as string + # Get profile name as string - use id for backward compatibility + # but we'll store it in the STREAM_PROFILE field profile_value = stream_profile.id return stream_url, stream_user_agent, transcode, profile_value diff --git a/apps/proxy/ts_proxy/views.py b/apps/proxy/ts_proxy/views.py index fe87e677..504fa72c 100644 --- a/apps/proxy/ts_proxy/views.py +++ b/apps/proxy/ts_proxy/views.py @@ -18,7 +18,7 @@ from apps.m3u.models import M3UAccount, M3UAccountProfile from core.models import UserAgent, CoreSettings, PROXY_PROFILE_NAME from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated -from .constants import ChannelState, EventType, StreamType +from .constants import ChannelState, EventType, StreamType, ChannelMetadataField 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 @@ -56,15 +56,17 @@ def stream_ts(request, channel_id): metadata_key = RedisKeys.channel_metadata(channel_id) if proxy_server.redis_client.exists(metadata_key): metadata = proxy_server.redis_client.hgetall(metadata_key) - if b'state' in metadata: - channel_state = metadata[b'state'].decode('utf-8') + state_field = ChannelMetadataField.STATE.encode('utf-8') + if state_field in metadata: + channel_state = metadata[state_field].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_field = ChannelMetadataField.OWNER.encode('utf-8') + if owner_field in metadata: + owner = metadata[owner_field].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 @@ -134,9 +136,9 @@ def stream_ts(request, channel_id): if proxy_server.redis_client: metadata_key = RedisKeys.channel_metadata(channel_id) - url_bytes = proxy_server.redis_client.hget(metadata_key, "url") - ua_bytes = proxy_server.redis_client.hget(metadata_key, "user_agent") - profile_bytes = proxy_server.redis_client.hget(metadata_key, "profile") + url_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.URL) + ua_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.USER_AGENT) + profile_bytes = proxy_server.redis_client.hget(metadata_key, ChannelMetadataField.STREAM_PROFILE) if url_bytes: url = url_bytes.decode('utf-8')