# Monkey patch MUST be first - before any other imports! import eventlet eventlet.monkey_patch() import os import json import subprocess import threading import queue import time import collections from flask import Flask, send_from_directory, jsonify, request, session, Response, stream_with_context from flask_socketio import SocketIO, emit from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() def _load_config(): """Loads optional config.json from the project root. If missing or invalid, returns an empty dict. """ try: with open('config.json', 'r', encoding='utf-8') as f: data = json.load(f) return data if isinstance(data, dict) else {} except FileNotFoundError: return {} except Exception: return {} CONFIG = _load_config() DJ_PANEL_PASSWORD = (CONFIG.get('dj_panel_password') or '').strip() DJ_AUTH_ENABLED = bool(DJ_PANEL_PASSWORD) # Relay State broadcast_state = { 'active': False, } listener_sids = set() dj_sids = set() # === Optional MP3 fallback stream (server-side transcoding) === # This allows listeners on browsers that don't support WebM/Opus via MediaSource # (notably some Safari / locked-down environments) to still hear the stream. _ffmpeg_proc = None _ffmpeg_in_q = queue.Queue(maxsize=40) _current_bitrate = "192k" _mp3_clients = set() # set[queue.Queue] _mp3_lock = threading.Lock() _transcode_threads_started = False _transcoder_bytes_out = 0 _transcoder_last_error = None _last_audio_chunk_ts = 0.0 _mp3_preroll = collections.deque(maxlen=256) # Pre-roll (~10s at 192k with 1KB chunks) def _start_transcoder_if_needed(is_mp3_input=False): global _ffmpeg_proc, _transcode_threads_started, _transcoder_last_error # If already running, check if we need to restart for mode change if _ffmpeg_proc is not None and _ffmpeg_proc.poll() is None: return # Local broadcast mode: input from pipe (Relay mode removed) # If input is already MP3, we just use 'copy' to avoid double-encoding codec = 'copy' if is_mp3_input else 'libmp3lame' cmd = [ 'ffmpeg', '-hide_banner', '-loglevel', 'error', '-fflags', 'nobuffer', '-flags', 'low_delay', '-probesize', '32', '-analyzeduration', '0' ] if is_mp3_input: cmd.extend(['-f', 'mp3']) cmd.extend([ '-i', 'pipe:0', '-vn', '-acodec', codec, ]) if not is_mp3_input: cmd.extend(['-b:a', _current_bitrate]) cmd.extend([ '-tune', 'zerolatency', '-flush_packets', '1', '-f', 'mp3', 'pipe:1', ]) try: _ffmpeg_proc = subprocess.Popen( cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=0, ) except FileNotFoundError: _ffmpeg_proc = None print('โš ๏ธ ffmpeg not found; /stream.mp3 fallback disabled') return mode_str = "PASSTHROUGH (copy)" if is_mp3_input else f"TRANSCODE ({_current_bitrate})" print(f'๐ŸŽ›๏ธ ffmpeg transcoder started for /stream.mp3 ({mode_str})') # Reset error state _transcoder_last_error = None # Always ensure threads are running if we just started/restarted the process # Clear the input queue to avoid old data being sent to new process while not _ffmpeg_in_q.empty(): try: _ffmpeg_in_q.get_nowait() except: break def _writer(): global _transcoder_last_error, _transcode_threads_started print("๐Ÿงต Transcoder writer thread started") while True: try: chunk = _ffmpeg_in_q.get(timeout=2) if chunk is None: break proc = _ffmpeg_proc if proc is None or proc.stdin is None or proc.poll() is not None: continue proc.stdin.write(chunk) proc.stdin.flush() except queue.Empty: if _ffmpeg_proc is None or _ffmpeg_proc.poll() is not None: break continue except Exception as e: if _ffmpeg_proc is not None: print(f"โš ๏ธ Transcoder writer error: {e}") _transcoder_last_error = f'stdin write failed: {e}' break _transcode_threads_started = False print("๐Ÿงต Transcoder writer thread exiting") def _reader(): global _transcoder_bytes_out, _transcoder_last_error, _transcode_threads_started print("๐Ÿงต Transcoder reader thread started") proc = _ffmpeg_proc while proc and proc.poll() is None: try: # Smaller read for smoother delivery (1KB) # This prevents buffering delays at lower bitrates data = proc.stdout.read(1024) if not data: break _transcoder_bytes_out += len(data) # Store in pre-roll with _mp3_lock: _mp3_preroll.append(data) clients = list(_mp3_clients) for q in clients: try: q.put_nowait(data) except Exception: pass except Exception as e: print(f"โš ๏ธ Transcoder reader error: {e}") _transcoder_last_error = f'stdout read failed: {e}' break _transcode_threads_started = False print("๐Ÿงต Transcoder reader thread exiting") if not _transcode_threads_started: _transcode_threads_started = True threading.Thread(target=_writer, daemon=True).start() threading.Thread(target=_reader, daemon=True).start() def _stop_transcoder(): global _ffmpeg_proc, _transcode_threads_started try: _ffmpeg_in_q.put_nowait(None) except Exception: pass proc = _ffmpeg_proc _ffmpeg_proc = None # Reset thread flag so they can be re-launched if needed # (The existing threads will exit cleanly on None/EOF) _transcode_threads_started = False _mp3_preroll.clear() if proc is None: return # Signal all listening clients to finish their stream with _mp3_lock: clients = list(_mp3_clients) for q in clients: try: q.put_nowait(None) except: pass _mp3_clients.clear() try: proc.terminate() except Exception: pass def _feed_transcoder(data: bytes): global _last_audio_chunk_ts if _ffmpeg_proc is None or _ffmpeg_proc.poll() is not None: return _last_audio_chunk_ts = time.time() try: _ffmpeg_in_q.put_nowait(data) except Exception: # Queue full; drop to keep latency bounded. pass MUSIC_FOLDER = "music" # Ensure music folder exists if not os.path.exists(MUSIC_FOLDER): os.makedirs(MUSIC_FOLDER) # Helper for shared routes def setup_shared_routes(app): @app.route('/library.json') def get_library(): library = [] if os.path.exists(MUSIC_FOLDER): for filename in sorted(os.listdir(MUSIC_FOLDER)): if filename.lower().endswith(('.mp3', '.m4a', '.wav', '.flac', '.ogg')): library.append({ "title": os.path.splitext(filename)[0], "file": f"music/{filename}" }) return jsonify(library) @app.route('/') def serve_static(filename): response = send_from_directory('.', filename) if filename.endswith(('.css', '.js', '.html')): response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0' return response @app.route('/') def index(): return send_from_directory('.', 'index.html') @app.route('/upload', methods=['POST']) def upload_file(): if 'file' not in request.files: return jsonify({"success": False, "error": "No file provided"}), 400 file = request.files['file'] if file.filename == '': return jsonify({"success": False, "error": "No file selected"}), 400 if not file.filename.endswith('.mp3'): return jsonify({"success": False, "error": "Only MP3 files are allowed"}), 400 # Sanitize filename (keep extension) import re name_without_ext = os.path.splitext(file.filename)[0] name_without_ext = re.sub(r'[^\w\s-]', '', name_without_ext) name_without_ext = re.sub(r'[-\s]+', '-', name_without_ext) filename = f"{name_without_ext}.mp3" filepath = os.path.join(MUSIC_FOLDER, filename) try: file.save(filepath) print(f"โœ… Uploaded: {filename}") return jsonify({"success": True, "filename": filename}) except Exception as e: print(f"โŒ Upload error: {e}") return jsonify({"success": False, "error": str(e)}), 500 @app.route('/save_keymaps', methods=['POST']) def save_keymaps(): try: data = request.get_json() with open('keymaps.json', 'w', encoding='utf-8') as f: json.dump(data, f, indent=4) print("๐Ÿ’พ Keymaps saved to keymaps.json") return jsonify({"success": True}) except Exception as e: print(f"โŒ Save keymaps error: {e}") return jsonify({"success": False, "error": str(e)}), 500 @app.route('/load_keymaps', methods=['GET']) def load_keymaps(): try: if os.path.exists('keymaps.json'): with open('keymaps.json', 'r', encoding='utf-8') as f: data = json.load(f) return jsonify({"success": True, "keymaps": data}) else: return jsonify({"success": True, "keymaps": None}) except Exception as e: print(f"โŒ Load keymaps error: {e}") return jsonify({"success": False, "error": str(e)}), 500 @app.route('/stream.mp3') def stream_mp3(): # Streaming response from the ffmpeg transcoder output. # If ffmpeg isn't available, return 503. if _ffmpeg_proc is None or _ffmpeg_proc.poll() is not None: return jsonify({"success": False, "error": "MP3 stream not available"}), 503 print(f"๐Ÿ‘‚ New listener joined stream (Bursting {_mp3_preroll.maxlen} frames)") client_q: queue.Queue = queue.Queue(maxsize=100) with _mp3_lock: # Burst pre-roll to new client so they start playing instantly for chunk in _mp3_preroll: try: client_q.put_nowait(chunk) except Exception: break _mp3_clients.add(client_q) def gen(): try: while True: chunk = client_q.get() if chunk is None: break yield chunk finally: with _mp3_lock: _mp3_clients.discard(client_q) return Response(stream_with_context(gen()), content_type='audio/mpeg', headers={ 'Cache-Control': 'no-cache, no-store, must-revalidate', 'Connection': 'keep-alive', 'Pragma': 'no-cache', 'Expires': '0', 'X-Content-Type-Options': 'nosniff' }) @app.route('/stream_debug') def stream_debug(): proc = _ffmpeg_proc running = proc is not None and proc.poll() is None return jsonify({ 'broadcast_active': broadcast_state.get('active', False), 'broadcast_mimeType': broadcast_state.get('mimeType'), 'ffmpeg_running': running, 'ffmpeg_found': (proc is not None), 'mp3_clients': len(_mp3_clients), 'transcoder_bytes_out': _transcoder_bytes_out, 'transcoder_last_error': _transcoder_last_error, 'last_audio_chunk_ts': _last_audio_chunk_ts, }) # === DJ SERVER (Port 5000) === dj_app = Flask(__name__, static_folder='.', static_url_path='') dj_app.config['SECRET_KEY'] = 'dj_panel_secret' setup_shared_routes(dj_app) @dj_app.before_request def _protect_dj_panel(): """Optionally require a password for the DJ panel only (port 5000). This does not affect the listener server (port 5001). """ if not DJ_AUTH_ENABLED: return None # Allow login/logout endpoints if request.path in ('/login', '/logout'): return None # If already authenticated, allow if session.get('dj_authed') is True: return None # Redirect everything else to login return ( "" "Redirecting to /login...", 302, {'Location': '/login'} ) @dj_app.route('/login', methods=['GET', 'POST']) def dj_login(): if not DJ_AUTH_ENABLED: # If auth is disabled, just go to the panel. session['dj_authed'] = True return ( "" "Auth disabled. Redirecting...", 302, {'Location': '/'} ) error = None if request.method == 'POST': pw = (request.form.get('password') or '').strip() if pw == DJ_PANEL_PASSWORD: session['dj_authed'] = True return ( "" "Logged in. Redirecting...", 302, {'Location': '/'} ) error = 'Invalid password' # Minimal inline login page (no new assets) return f""" TechDJ - DJ Login

DJ Panel Locked

{f"
{error}
" if error else ""}
Set/disable this in config.json (dj_panel_password).
""" @dj_app.route('/logout') def dj_logout(): session.pop('dj_authed', None) return ( "" "Logged out. Redirecting...", 302, {'Location': '/login'} ) dj_socketio = SocketIO( dj_app, cors_allowed_origins="*", async_mode='eventlet', max_http_buffer_size=1e8, # 100MB buffer ping_timeout=10, ping_interval=5, logger=False, engineio_logger=False ) @dj_socketio.on('connect') def dj_connect(): if DJ_AUTH_ENABLED and session.get('dj_authed') is not True: print(f"โ›” DJ socket rejected (unauthorized): {request.sid}") return False print(f"๐ŸŽง DJ connected: {request.sid}") dj_sids.add(request.sid) @dj_socketio.on('disconnect') def dj_disconnect(): dj_sids.discard(request.sid) print("โš ๏ธ DJ disconnected - broadcast will continue until manually stopped") def stop_broadcast_after_timeout(): """No longer used - broadcasts don't auto-stop""" pass @dj_socketio.on('start_broadcast') def dj_start(data=None): broadcast_state['active'] = True session['is_dj'] = True print("๐ŸŽ™๏ธ Broadcast -> ACTIVE") is_mp3_input = False if data: if 'bitrate' in data: global _current_bitrate _current_bitrate = data['bitrate'] print(f"๐Ÿ“ก Setting stream bitrate to: {_current_bitrate}") if data.get('format') == 'mp3': is_mp3_input = True # Clear pre-roll for fresh start _mp3_preroll.clear() _start_transcoder_if_needed(is_mp3_input=is_mp3_input) listener_socketio.emit('broadcast_started', namespace='/') listener_socketio.emit('stream_status', {'active': True}, namespace='/') @dj_socketio.on('get_listener_count') def dj_get_listener_count(): emit('listener_count', {'count': len(listener_sids)}) @dj_socketio.on('stop_broadcast') def dj_stop(): broadcast_state['active'] = False session['is_dj'] = False print("๐Ÿ›‘ DJ stopped broadcasting") _stop_transcoder() listener_socketio.emit('broadcast_stopped', namespace='/') listener_socketio.emit('stream_status', {'active': False}, namespace='/') listener_socketio.emit('broadcast_stopped', namespace='/') listener_socketio.emit('stream_status', {'active': False}, namespace='/') @dj_socketio.on('audio_chunk') def dj_audio(data): # MP3-only mode: do not relay raw chunks to listeners; feed transcoder only. if broadcast_state['active']: # Ensure MP3 fallback transcoder is running (if ffmpeg is installed) if _ffmpeg_proc is None or _ffmpeg_proc.poll() is not None: # If we don't know the format, default to transcode, # but usually start_broadcast handles this _start_transcoder_if_needed() if isinstance(data, (bytes, bytearray)): _feed_transcoder(bytes(data)) # === LISTENER SERVER (Port 5001) === listener_app = Flask(__name__, static_folder='.', static_url_path='') listener_app.config['SECRET_KEY'] = 'listener_secret' setup_shared_routes(listener_app) listener_socketio = SocketIO( listener_app, cors_allowed_origins="*", async_mode='eventlet', max_http_buffer_size=1e8, # 100MB buffer ping_timeout=10, ping_interval=5, logger=False, engineio_logger=False ) @listener_socketio.on('connect') def listener_connect(): print(f"๐Ÿ‘‚ Listener Socket Connected: {request.sid}") @listener_socketio.on('disconnect') def listener_disconnect(): listener_sids.discard(request.sid) count = len(listener_sids) print(f"โŒ Listener left. Total: {count}") # Notify BOTH namespaces listener_socketio.emit('listener_count', {'count': count}, namespace='/') dj_socketio.emit('listener_count', {'count': count}, namespace='/') @listener_socketio.on('join_listener') def listener_join(): if request.sid not in listener_sids: listener_sids.add(request.sid) count = len(listener_sids) print(f"๐Ÿ‘‚ New listener joined. Total: {count}") listener_socketio.emit('listener_count', {'count': count}, namespace='/') dj_socketio.emit('listener_count', {'count': count}, namespace='/') emit('stream_status', {'active': broadcast_state['active']}) @listener_socketio.on('get_listener_count') def listener_get_count(): emit('listener_count', {'count': len(listener_sids)}) # DJ Panel Routes (No engine commands needed in local mode) @dj_socketio.on('get_mixer_status') def get_mixer_status(): pass @dj_socketio.on('audio_sync_queue') def audio_sync_queue(data): pass def _listener_count_sync_loop(): """Periodic background sync to ensure listener count is always accurate.""" while True: count = len(listener_sids) listener_socketio.emit('listener_count', {'count': count}, namespace='/') dj_socketio.emit('listener_count', {'count': count}, namespace='/') eventlet.sleep(5) if __name__ == '__main__': print("=" * 50) print("๐ŸŽง TECHDJ PRO - DUAL PORT ARCHITECTURE") print("=" * 50) # Ports from environment or defaults dj_port = int(os.environ.get('DJ_PORT', 5000)) listen_port = int(os.environ.get('LISTEN_PORT', 5001)) print(f"๐Ÿ‘‰ DJ PANEL API: http://0.0.0.0:{dj_port}") print(f"๐Ÿ‘‰ LISTEN PAGE: http://0.0.0.0:{listen_port}") print("=" * 50) # Audio engine DISABLED print(f"โœ… Local Radio server ready on ports {dj_port} & {listen_port}") # Run both servers using eventlet's spawn eventlet.spawn(_listener_count_sync_loop) eventlet.spawn(dj_socketio.run, dj_app, host='0.0.0.0', port=dj_port, debug=False) listener_socketio.run(listener_app, host='0.0.0.0', port=listen_port, debug=False)