From 9b6f6073e0bcc6a16a38c2bc6c48f12d8e787c4e Mon Sep 17 00:00:00 2001 From: SergeantPanda Date: Fri, 28 Feb 2025 11:04:05 -0600 Subject: [PATCH] Initial Proxy release --- apps/proxy/hls_proxy | 627 +++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + 2 files changed, 628 insertions(+) create mode 100644 apps/proxy/hls_proxy diff --git a/apps/proxy/hls_proxy b/apps/proxy/hls_proxy new file mode 100644 index 00000000..34dfb1ea --- /dev/null +++ b/apps/proxy/hls_proxy @@ -0,0 +1,627 @@ +""" +HLS Proxy Server with Advanced Stream Switching Support +This proxy handles HLS live streams with support for: +- Stream switching with proper discontinuity handling +- Buffer management +- Segment validation +- Connection pooling and reuse +""" + +from flask import Flask, Response, request, jsonify +import requests +import threading +import logging +import m3u8 +import time +from urllib.parse import urlparse, urljoin +import argparse +from typing import Optional +import sys + +# Initialize Flask app +app = Flask(__name__) + +# Global state management +manifest_buffer = None # Stores current manifest content +segment_buffers = {} # Maps sequence numbers to segment data +buffer_lock = threading.Lock() # Synchronizes access to buffers + +class Config: + """Configuration settings for stream handling and buffering""" + # Buffer size settings + MIN_SEGMENTS = 12 # Minimum segments to maintain + MAX_SEGMENTS = 16 # Maximum segments to store + WINDOW_SIZE = 12 # Number of segments in manifest window + INITIAL_SEGMENTS = 3 # Initial segments to buffer before playback + +class StreamFetcher: + """Handles HTTP requests for stream segments with connection pooling""" + def __init__(self, stream_url): + self.stream_url = stream_url + self.session = requests.Session() + + # Set up connection pooling + adapter = requests.adapters.HTTPAdapter( + pool_connections=2, # Number of connection pools + pool_maxsize=4, # Connections per pool + max_retries=3, # Auto-retry failed requests + pool_block=False # Don't block when pool is full + ) + + # Apply adapter to both HTTP and HTTPS + self.session.mount('http://', adapter) + self.session.mount('https://', adapter) + + # Request optimization + self.last_request_time = 0 + self.min_request_interval = 0.05 # Minimum time between requests + self.last_host = None # Cache last successful host + self.redirect_cache = {} # Cache redirect responses + + def get_base_host(self, url): + """Extract base host from URL using urlparse""" + try: + parsed = urlparse(url) + return f"{parsed.scheme}://{parsed.netloc}" + except Exception as e: + logging.error(f"Error extracting base host: {e}") + return url + + def download(self, url): + """Download content with connection reuse""" + now = time.time() + wait_time = self.last_request_time + self.min_request_interval - now + if (wait_time > 0): + time.sleep(wait_time) + + try: + # Use cached redirect if available + if url in self.redirect_cache: + logging.debug(f"Using cached redirect for {url}") + final_url = self.redirect_cache[url] + response = self.session.get(final_url, timeout=10) + else: + response = self.session.get(url, allow_redirects=True, timeout=10) + if response.history: # Cache redirects + logging.debug(f"Caching redirect for {url} -> {response.url}") + self.redirect_cache[url] = response.url + + self.last_request_time = time.time() + + if response.status_code == 200: + self.last_host = self.get_base_host(response.url) + + return response.content, response.url + + except Exception as e: + logging.error(f"Download error: {e}") + if self.last_host and not url.startswith(self.last_host): + # Use urljoin to handle path resolution + new_url = urljoin(self.last_host + '/', url.split('://')[-1].split('/', 1)[-1]) + logging.debug(f"Retrying with last host: {new_url}") + return self.download(new_url) + raise + +def analyze_ts_packet(data: bytes) -> dict: + """ + Analyze a single MPEG-TS packet (188 bytes) + + Args: + data: Raw packet bytes + + Returns: + dict with packet analysis: + - sync_valid: True if sync byte is 0x47 + - transport_error: True if TEI bit is set + - payload_start: True if PUSI bit is set + - pid: Packet ID (13 bits) + - hex_dump: First 16 bytes for debugging + """ + # Validate minimum packet size + if len(data) < 188: + return { + 'sync_valid': False, + 'transport_error': True, + 'payload_start': False, + 'pid': 0, + 'error': 'Packet too short' + } + + # Verify sync byte (0x47) + if data[0] != 0x47: + return { + 'sync_valid': False, + 'transport_error': True, + 'payload_start': False, + 'pid': 0, + 'error': 'Invalid sync byte' + } + + # Extract packet header fields + transport_error = (data[1] & 0x80) != 0 # Transport Error Indicator + payload_start = (data[1] & 0x40) != 0 # Payload Unit Start Indicator + pid = ((data[1] & 0x1F) << 8) | data[2] # Packet ID (13 bits) + + # Create hex dump for debugging + hex_dump = ' '.join(f'{b:02x}' for b in data[:16]) + + return { + 'sync_valid': True, + 'transport_error': transport_error, + 'payload_start': payload_start, + 'pid': pid, + 'hex_dump': hex_dump, + 'packet_size': len(data) + } + +def get_segment_sequence(segment_uri: str) -> Optional[int]: + """ + Extract sequence number from segment URI + + Args: + segment_uri: Segment filename or path + + Returns: + int: Sequence number if found + None: If no sequence can be extracted + """ + try: + # Try numerical sequence (e.g., 1038_3693.ts) + if '_' in segment_uri: + return int(segment_uri.split('_')[-1].split('.')[0]) + return None + except ValueError: + return None + +# Update verify_segment with more thorough checks +def verify_segment(data: bytes) -> dict: + """ + Verify MPEG-TS segment integrity + + Args: + data: Raw segment bytes + + Returns: + dict with verification results: + - valid: True if segment passes all checks + - packets: Number of valid packets found + - size: Total segment size in bytes + - error: Description if validation fails + """ + # Check minimum size + if len(data) < 188: + return {'valid': False, 'error': 'Segment too short'} + + # Verify segment size is multiple of packet size + if len(data) % 188 != 0: + return {'valid': False, 'error': 'Invalid segment size'} + + valid_packets = 0 + total_packets = len(data) // 188 + + # Scan all packets in segment + for i in range(0, len(data), 188): + packet = data[i:i+188] + + # Check packet completeness + if len(packet) != 188: + return {'valid': False, 'error': 'Incomplete packet'} + + # Verify sync byte + if packet[0] != 0x47: + return {'valid': False, 'error': f'Invalid sync byte at offset {i}'} + + # Check transport error indicator + if packet[1] & 0x80: + return {'valid': False, 'error': 'Transport error indicator set'} + + valid_packets += 1 + + return { + 'valid': True, + 'packets': valid_packets, + 'size': len(data) + } + +class StreamManager: + """Manages HLS stream state and switching logic""" + def __init__(self, initial_url: str): + # Stream state + self.current_url = initial_url + self.running = True + self.switching_stream = False + + # Sequence tracking + self.next_sequence = 0 + self.highest_sequence = 0 + self.buffered_sequences = set() + self.downloaded_sources = {} + self.segment_durations = {} + + # Source tracking + self.current_source = None + self.source_changes = set() + self.stream_switch_count = 0 + + # Threading + self.fetcher = None + self.fetch_thread = None + self.url_changed = threading.Event() + + def update_url(self, new_url: str) -> bool: + """Handle stream URL changes with proper discontinuity marking""" + if new_url != self.current_url: + with buffer_lock: + self.switching_stream = True + self.current_url = new_url + + # Set sequence numbers for stream switch + if segment_buffers: + self.highest_sequence = max(segment_buffers.keys()) + self.next_sequence = self.highest_sequence + 1 + # Mark discontinuity at first segment of new stream + self.source_changes = {self.next_sequence + 1} + else: + self.stream_switch_count += 1 + self.next_sequence = self.stream_switch_count * 1000 + self.source_changes = {self.next_sequence} + + logging.info(f"Stream switch - next sequence will start at {self.next_sequence}") + + # Clear state but maintain sequence numbers + self.downloaded_sources.clear() + self.segment_durations.clear() + self.current_source = None + + # Signal thread to switch URL + self.url_changed.set() + + return True + return False + + def get_next_sequence(self, source_id): + """Assign sequence numbers to segments with source change detection""" + if source_id in self.downloaded_sources: + return None + + seq = self.next_sequence + while seq in self.buffered_sequences: + seq += 1 + + # Track source changes for discontinuity markers + source_prefix = source_id.split('_')[0] + if not self.switching_stream and self.current_source and self.current_source != source_prefix: + self.source_changes.add(seq) + logging.debug(f"Source change detected at sequence {seq}") + self.current_source = source_prefix + + # Update tracking + self.downloaded_sources[source_id] = seq + self.buffered_sequences.add(seq) + self.next_sequence = seq + 1 + self.highest_sequence = max(self.highest_sequence, seq) + + return seq + + def _fetch_loop(self): + """Background thread for continuous stream fetching""" + while self.running: + try: + self.fetcher = StreamFetcher(self.current_url) + fetch_stream(self.fetcher, self.url_changed, self.next_sequence) + except Exception as e: + logging.error(f"Stream error: {e}") + time.sleep(5) # Wait before retry + + self.url_changed.clear() + + def start(self): + """Start the background fetch thread""" + if not self.fetch_thread or not self.fetch_thread.is_alive(): + self.running = True + self.fetch_thread = threading.Thread( + target=self._fetch_loop, + name="StreamFetcher", + daemon=True # Thread will exit when main program does + ) + self.fetch_thread.start() + logging.info("Stream manager started") + + def stop(self): + """Stop the background fetch thread""" + self.running = False + if self.fetch_thread and self.fetch_thread.is_alive(): + self.url_changed.set() # Signal thread to exit + self.fetch_thread.join(timeout=5) # Wait up to 5 seconds + logging.info("Stream manager stopped") + +def fetch_stream(fetcher: StreamFetcher, stop_event: threading.Event, start_sequence: int = 0): + global manifest_buffer, segment_buffers + retry_delay = 1 + max_retry_delay = 8 + last_segment_time = 0 + buffer_initialized = False + manifest_update_needed = True + segment_duration = None + + while not stop_event.is_set(): + try: + now = time.time() + + # Only update manifest when it's time for next segment + should_update = ( + manifest_update_needed or + not segment_duration or + (last_segment_time and (now - last_segment_time) >= segment_duration * 0.8) + ) + + if should_update: + manifest_data, final_url = fetcher.download(fetcher.stream_url) + manifest = m3u8.loads(manifest_data.decode()) + + if not manifest.segments: + continue + + with buffer_lock: + manifest_content = manifest_data.decode() + new_segments = {} + + if stream_manager.switching_stream: + # Stream switch - only get latest segment + manifest_segments = [manifest.segments[-1]] + seq_start = stream_manager.next_sequence + max_segments = 1 + logging.debug(f"Processing stream switch - getting latest segment at sequence {seq_start}") + elif not buffer_initialized: + # Initial buffer + manifest_segments = manifest.segments[-Config.INITIAL_SEGMENTS:] + seq_start = stream_manager.next_sequence + max_segments = Config.INITIAL_SEGMENTS + logging.debug(f"Starting initial buffer at sequence {seq_start}") + else: + # Normal operation + manifest_segments = [manifest.segments[-1]] + seq_start = stream_manager.next_sequence + max_segments = 1 + + # Map segments + segments_mapped = 0 + for segment in manifest_segments: + if segments_mapped >= max_segments: + break + + source_id = segment.uri.split('/')[-1].split('.')[0] + next_seq = stream_manager.get_next_sequence(source_id) + + if next_seq is not None: + duration = float(segment.duration) + new_segments[next_seq] = { + 'uri': segment.uri, + 'duration': duration, + 'source_id': source_id + } + stream_manager.segment_durations[next_seq] = duration + segments_mapped += 1 + + manifest_buffer = manifest_content + + # Download segments + for sequence_id, segment_info in new_segments.items(): + try: + segment_url = f"{fetcher.last_host}{segment_info['uri']}" + logging.debug(f"Downloading {segment_info['uri']} as segment {sequence_id}.ts " + f"(source: {segment_info['source_id']}, duration: {segment_info['duration']:.3f}s)") + + segment_data, _ = fetcher.download(segment_url) + validation = verify_segment(segment_data) + + if validation.get('valid', False): + with buffer_lock: + segment_buffers[sequence_id] = segment_data + logging.debug(f"Downloaded and verified segment {sequence_id} (packets: {validation['packets']})") + + if stream_manager.switching_stream: + stream_manager.switching_stream = False + stop_event.set() # Force fetcher restart with new URL + break + elif not buffer_initialized and len(segment_buffers) >= Config.INITIAL_SEGMENTS: + buffer_initialized = True + manifest_update_needed = True + break + except Exception as e: + logging.error(f"Segment download error: {e}") + continue + + else: + # Short sleep to prevent CPU spinning + threading.Event().wait(0.1) + + except Exception as e: + logging.error(f"Manifest error: {e}") + threading.Event().wait(retry_delay) + retry_delay = min(retry_delay * 2, max_retry_delay) + manifest_update_needed = True + +# Flask Routes for HLS Proxy Server +@app.route('/stream.m3u8') +def master_playlist(): + """ + Serve the HLS master playlist + + Handles: + - Initial buffering state + - Ongoing playback with sliding window + - Discontinuity markers for stream switches + - Dynamic segment durations + """ + with buffer_lock: + # Verify buffer state + if not manifest_buffer or not segment_buffers: + logging.warning("No manifest or segments available yet") + return '', 404 + + available = sorted(segment_buffers.keys()) + if not available: + logging.warning("No segments available") + return '', 404 + + manifest = m3u8.loads(manifest_buffer) + max_seq = max(available) + + # Calculate window bounds + if len(available) <= Config.INITIAL_SEGMENTS: + # During initial buffering, show all segments + min_seq = min(available) + else: + # For ongoing playback, maintain sliding window + min_seq = max( + min(available), + max_seq - Config.WINDOW_SIZE + 1 + ) + + # Build manifest with proper tags + new_manifest = ['#EXTM3U'] + new_manifest.append('#EXT-X-VERSION:3') + new_manifest.append(f'#EXT-X-MEDIA-SEQUENCE:{min_seq}') + new_manifest.append(f'#EXT-X-TARGETDURATION:{int(manifest.target_duration)}') + + # Filter segments within window + window_segments = [s for s in available if min_seq <= s <= max_seq] + + # Add segments with discontinuity handling + for seq in window_segments: + # Mark stream switches with discontinuity + if seq in stream_manager.source_changes: + new_manifest.append('#EXT-X-DISCONTINUITY') + logging.debug(f"Added discontinuity marker before segment {seq}") + + # Use actual segment duration or fallback to target + duration = stream_manager.segment_durations.get(seq, manifest.target_duration) + new_manifest.append(f'#EXTINF:{duration:.3f},') + new_manifest.append(f'/segments/{seq}.ts') + + manifest_content = '\n'.join(new_manifest) + logging.debug(f"Serving manifest with segments {min_seq}-{max_seq} (window: {len(window_segments)})") + return Response(manifest_content, content_type='application/vnd.apple.mpegurl') + +@app.route('/segments/') +def get_segment(segment_name): + """ + Serve individual MPEG-TS segments + + Args: + segment_name: Segment filename (e.g., '123.ts') + + Returns: + MPEG-TS segment data or 404 if not found + """ + try: + segment_id = int(segment_name.split('.')[0]) + with buffer_lock: + if segment_id in segment_buffers: + available = sorted(segment_buffers.keys()) + logging.debug(f"Client requested segment {segment_id} (buffer: {min(available)}-{max(available)})") + return Response(segment_buffers[segment_id], content_type='video/MP2T') + + logging.warning( + f"Segment {segment_id} not found. " + f"Available: {min(segment_buffers.keys()) if segment_buffers else 'none'}" + f"-{max(segment_buffers.keys()) if segment_buffers else 'none'}" + ) + except Exception as e: + logging.error(f"Error serving segment {segment_name}: {e}") + return '', 404 + +@app.route('/change_stream', methods=['POST']) +def change_stream(): + """ + Handle stream URL changes via HTTP POST + + Expected JSON body: + {"url": "new_stream_url"} + + Returns: + JSON response indicating success/failure + """ + new_url = request.json.get('url') + if not new_url: + return jsonify({'error': 'No URL provided'}), 400 + + if stream_manager.update_url(new_url): + return jsonify({'message': 'Stream URL updated', 'url': new_url}) + return jsonify({'message': 'URL unchanged', 'url': new_url}) + +@app.before_request +def log_request_info(): + """Log client connections and important requests""" + if request.path == '/stream.m3u8' and not segment_buffers: + # First manifest request from a client + logging.info(f"New client connected from {request.remote_addr}") + elif request.path.startswith('/change_stream'): + # Keep stream switch requests as INFO + logging.info(f"Stream switch requested from {request.remote_addr}") + else: + # Move routine requests to DEBUG + logging.debug(f"{request.remote_addr} - {request.method} {request.path}") + +# Configure Werkzeug logger to DEBUG +logging.getLogger('werkzeug').setLevel(logging.DEBUG) + +# Main Application Setup +if __name__ == '__main__': + # Command line argument parsing + parser = argparse.ArgumentParser(description='HLS Proxy Server with Stream Switching') + parser.add_argument( + '--url', '-u', + default='http://example.com/stream.m3u8', + help='Initial HLS stream URL to proxy' + ) + parser.add_argument( + '--port', '-p', + type=int, + default=5000, + help='Local port to serve proxy on (default: 5000)' + ) + parser.add_argument( + '--host', '-H', + default='0.0.0.0', + help='Interface to bind server to (default: all interfaces)' + ) + parser.add_argument( + '--debug', + action='store_true', + help='Enable debug logging' + ) + args = parser.parse_args() + + # Configure logging with separate format for access logs + logging.basicConfig( + level=logging.DEBUG if args.debug else logging.INFO, + format='%(asctime)s - %(levelname)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' + ) + + # Initialize proxy components + try: + # Create and start stream manager + stream_manager = StreamManager(args.url) + stream_manager.start() + + logging.info(f"Starting HLS proxy server on {args.host}:{args.port}") + logging.info(f"Initial stream URL: {args.url}") + + # Run Flask development server + # Note: For production, use a proper WSGI server like gunicorn + app.run( + host=args.host, + port=args.port, + threaded=True, # Enable multi-threading for segment handling + debug=args.debug + ) + except Exception as e: + logging.error(f"Failed to start server: {e}") + if stream_manager: + stream_manager.running = False + if stream_manager.fetch_thread: + stream_manager.fetch_thread.join() + sys.exit(1) diff --git a/requirements.txt b/requirements.txt index d479a65c..a2fbbc7a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,3 +14,4 @@ yt-dlp gevent==24.11.1 django-cors-headers djangorestframework-simplejwt +m3u8 \ No newline at end of file