Files
techdj/server.py
3nd3r 81120ac7ea Add remote stream relay feature: relay remote DJ streams to listeners
- Server-side: Added remote URL support in ffmpeg transcoder
- UI: Added relay controls in streaming panel with URL input
- Client: Added start/stop relay functions with socket communication
- Listener: Shows remote relay status in stream indicator
2026-01-03 10:29:10 -06:00

622 lines
21 KiB
Python

# 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
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()
import downloader
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=200)
_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
_remote_stream_url = None # For relaying remote streams
def _start_transcoder_if_needed():
global _ffmpeg_proc, _transcode_threads_started
if _ffmpeg_proc is not None and _ffmpeg_proc.poll() is None:
return
if _remote_stream_url:
# Remote relay mode: input from URL
cmd = [
'ffmpeg',
'-hide_banner',
'-loglevel', 'error',
'-i', _remote_stream_url,
'-vn',
'-acodec', 'libmp3lame',
'-b:a', '192k',
'-f', 'mp3',
'pipe:1',
]
else:
# Local broadcast mode: input from pipe
cmd = [
'ffmpeg',
'-hide_banner',
'-loglevel', 'error',
'-i', 'pipe:0',
'-vn',
'-acodec', 'libmp3lame',
'-b:a', '192k',
'-f', 'mp3',
'pipe:1',
]
try:
if _remote_stream_url:
_ffmpeg_proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
bufsize=0,
)
else:
_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
print(f'🎛️ ffmpeg transcoder started for /stream.mp3 ({ "remote relay" if _remote_stream_url else "local broadcast" })')
def _writer():
global _transcoder_last_error
while True:
chunk = _ffmpeg_in_q.get()
if chunk is None:
break
proc = _ffmpeg_proc
if proc is None or proc.stdin is None:
continue
try:
proc.stdin.write(chunk)
except Exception:
# If ffmpeg dies or pipe breaks, just stop writing.
_transcoder_last_error = 'stdin write failed'
break
def _reader():
global _transcoder_bytes_out, _transcoder_last_error
proc = _ffmpeg_proc
if proc is None or proc.stdout is None:
return
while True:
try:
data = proc.stdout.read(4096)
except Exception:
_transcoder_last_error = 'stdout read failed'
break
if not data:
break
_transcoder_bytes_out += len(data)
with _mp3_lock:
clients = list(_mp3_clients)
for q in clients:
try:
q.put_nowait(data)
except Exception:
# Drop if client queue is full or gone.
pass
if not _transcode_threads_started:
threading.Thread(target=_writer, daemon=True).start()
threading.Thread(target=_reader, daemon=True).start()
_transcode_threads_started = True
def _stop_transcoder():
global _ffmpeg_proc
try:
_ffmpeg_in_q.put_nowait(None)
except Exception:
pass
proc = _ffmpeg_proc
_ffmpeg_proc = None
if proc is None:
return
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 or _remote_stream_url:
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('/download', methods=['POST'])
def download():
data = request.get_json(silent=True) or {}
url = data.get('url')
quality = data.get('quality', '320')
if not url:
return jsonify({"success": False, "error": "No URL provided"}), 400
result = downloader.download_mp3(url, quality)
return jsonify(result)
@app.route('/search_youtube', methods=['GET'])
def search_youtube():
query = request.args.get('q', '')
if not query:
return jsonify({"success": False, "error": "No query provided"}), 400
# Get API key from environment variable
api_key = os.environ.get('YOUTUBE_API_KEY', '')
if not api_key:
return jsonify({
"success": False,
"error": "YouTube API key not configured. Set YOUTUBE_API_KEY environment variable."
}), 500
try:
import requests
# Search YouTube using Data API v3
url = 'https://www.googleapis.com/youtube/v3/search'
params = {
'part': 'snippet',
'q': query,
'type': 'video',
'videoCategoryId': '10', # Music category
'maxResults': 20,
'key': api_key
}
response = requests.get(url, params=params)
data = response.json()
if 'error' in data:
return jsonify({
"success": False,
"error": data['error'].get('message', 'YouTube API error')
}), 400
# Format results
results = []
for item in data.get('items', []):
results.append({
'videoId': item['id']['videoId'],
'title': item['snippet']['title'],
'channel': item['snippet']['channelTitle'],
'thumbnail': item['snippet']['thumbnails']['medium']['url'],
'url': f"https://www.youtube.com/watch?v={item['id']['videoId']}"
})
return jsonify({"success": True, "results": results})
except Exception as e:
return jsonify({"success": False, "error": str(e)}), 500
@app.route('/<path:filename>')
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('/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
client_q: queue.Queue = queue.Queue(maxsize=200)
with _mp3_lock:
_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()),
mimetype='audio/mpeg',
headers={
'Cache-Control': 'no-store, no-cache, must-revalidate, max-age=0',
'Connection': 'keep-alive',
},
)
@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 (
"<!doctype html><html><head><meta http-equiv='refresh' content='0; url=/login' /></head>"
"<body>Redirecting to <a href='/login'>/login</a>...</body></html>",
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 (
"<!doctype html><html><head><meta http-equiv='refresh' content='0; url=/' /></head>"
"<body>Auth disabled. Redirecting...</body></html>",
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 (
"<!doctype html><html><head><meta http-equiv='refresh' content='0; url=/' /></head>"
"<body>Logged in. Redirecting...</body></html>",
302,
{'Location': '/'}
)
error = 'Invalid password'
# Minimal inline login page (no new assets)
return f"""<!doctype html>
<html lang=\"en\">
<head>
<meta charset=\"utf-8\" />
<meta name=\"viewport\" content=\"width=device-width, initial-scale=1\" />
<title>TechDJ - DJ Login</title>
<style>
body {{ background:#0a0a12; color:#eee; font-family: system-ui, -apple-system, Segoe UI, Roboto, Arial; margin:0; }}
.wrap {{ min-height:100vh; display:flex; align-items:center; justify-content:center; padding:24px; }}
.card {{ width:100%; max-width:420px; background:rgba(10,10,20,0.85); border:2px solid #bc13fe; border-radius:16px; padding:24px; box-shadow:0 0 40px rgba(188,19,254,0.25); }}
h1 {{ margin:0 0 16px 0; font-size:22px; }}
label {{ display:block; margin:12px 0 8px; opacity:0.9; }}
input {{ width:100%; padding:12px; border-radius:10px; border:1px solid rgba(255,255,255,0.15); background:rgba(0,0,0,0.35); color:#fff; }}
button {{ width:100%; margin-top:14px; padding:12px; border-radius:10px; border:2px solid #bc13fe; background:rgba(188,19,254,0.15); color:#fff; font-weight:700; cursor:pointer; }}
.err {{ margin-top:12px; color:#ffb3ff; }}
.hint {{ margin-top:10px; font-size:12px; opacity:0.7; }}
</style>
</head>
<body>
<div class=\"wrap\">
<div class=\"card\">
<h1>DJ Panel Locked</h1>
<form method=\"post\" action=\"/login\">
<label for=\"password\">Password</label>
<input id=\"password\" name=\"password\" type=\"password\" autocomplete=\"current-password\" autofocus />
<button type=\"submit\">Unlock DJ Panel</button>
{f"<div class='err'>{error}</div>" if error else ""}
<div class=\"hint\">Set/disable this in config.json (dj_panel_password).</div>
</form>
</div>
</div>
</body>
</html>"""
@dj_app.route('/logout')
def dj_logout():
session.pop('dj_authed', None)
return (
"<!doctype html><html><head><meta http-equiv='refresh' content='0; url=/login' /></head>"
"<body>Logged out. Redirecting...</body></html>",
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")
_start_transcoder_if_needed()
listener_socketio.emit('broadcast_started', namespace='/')
listener_socketio.emit('stream_status', {'active': True}, namespace='/')
@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='/')
@dj_socketio.on('start_remote_relay')
def dj_start_remote_relay(data):
global _remote_stream_url
url = data.get('url', '').strip()
if not url:
dj_socketio.emit('error', {'message': 'No URL provided for remote relay'})
return
# Stop any existing broadcast/relay
if broadcast_state['active']:
dj_stop()
_remote_stream_url = url
broadcast_state['active'] = True
broadcast_state['remote_relay'] = True
session['is_dj'] = True
print(f"🔗 Starting remote relay from: {url}")
_start_transcoder_if_needed()
listener_socketio.emit('broadcast_started', namespace='/')
listener_socketio.emit('stream_status', {'active': True, 'remote_relay': True}, namespace='/')
@dj_socketio.on('stop_remote_relay')
def dj_stop_remote_relay():
global _remote_stream_url
_remote_stream_url = None
broadcast_state['active'] = False
broadcast_state['remote_relay'] = False
session['is_dj'] = False
print("🛑 Remote relay stopped")
_stop_transcoder()
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:
_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():
if request.sid in listener_sids:
listener_sids.discard(request.sid)
count = len(listener_sids)
print(f"❌ Listener left. Total: {count}")
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
if __name__ == '__main__':
print("=" * 50)
print("🎧 TECHDJ PRO - DUAL PORT ARCHITECTURE")
print("=" * 50)
print("👉 DJ PANEL: http://localhost:5000")
print("👉 LISTEN PAGE: http://localhost:5001")
print("=" * 50)
# Audio engine DISABLED
print("✅ Local Radio server ready")
# Run both servers using eventlet's spawn
eventlet.spawn(dj_socketio.run, dj_app, host='0.0.0.0', port=5000, debug=False)
listener_socketio.run(listener_app, host='0.0.0.0', port=5001, debug=False)