mixapps

serve.py (6.0 KB)

#!/usr/bin/env python3
"""
Starts HTTP server for local testing
Automatically manages a virtual environment for dependencies
"""

import http.server
import socketserver
import socket
import sys
import os
import json
import subprocess
from pathlib import Path

DEFAULT_PORT = 8000
SCRIPT_DIR = Path(__file__).parent.absolute()
VENV_DIR = SCRIPT_DIR / "venv"


def setup_venv():
	"""Create and setup virtual environment if it doesn't exist"""
	# Determine the path to pip and python in the venv
	if sys.platform == "win32":
		pip_path = VENV_DIR / "Scripts" / "pip"
		python_path = VENV_DIR / "Scripts" / "python"
	else:
		pip_path = VENV_DIR / "bin" / "pip"
		python_path = VENV_DIR / "bin" / "python3"

	# Check if venv needs to be created or recreated
	if not VENV_DIR.exists() or not python_path.exists():
		if VENV_DIR.exists():
			print("Virtual environment incomplete, recreating...")
			import shutil
			shutil.rmtree(VENV_DIR)
		else:
			print("Creating virtual environment...")

		try:
			subprocess.check_call([sys.executable, "-m", "venv", str(VENV_DIR)])
			print("Virtual environment created successfully.")
		except subprocess.CalledProcessError as e:
			print(f"Error creating virtual environment: {e}")
			sys.exit(1)

	# Ensure pip is available (sometimes venv doesn't include it)
	if not pip_path.exists():
		print("Installing pip in virtual environment...")
		try:
			subprocess.check_call([str(python_path), "-m", "ensurepip", "--upgrade"])
		except subprocess.CalledProcessError as e:
			print(f"Error ensuring pip: {e}")
			sys.exit(1)

	check = subprocess.run(
		[str(python_path), "-c", "import qrcode"],
		capture_output=True
	)
	if check.returncode != 0:
		try:
			subprocess.check_call([str(python_path), "-m", "pip", "install", "-q", "qrcode"])
		except subprocess.CalledProcessError:
			print("Note: Could not install qrcode (offline?). QR codes will be unavailable.\n")

	return python_path


def run_in_venv():
	"""Re-run this script in the virtual environment"""
	python_path = setup_venv()

	# Re-run this script with the venv Python
	try:
		subprocess.check_call([str(python_path), __file__, "--in-venv"])
	except (KeyboardInterrupt, subprocess.CalledProcessError):
		pass
	sys.exit(0)


def get_local_ip():
	"""Get the local IP address for network access"""
	try:
		# Create a socket to determine the local IP
		s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
		# Connect to a public DNS server (doesn't actually send data)
		s.connect(("8.8.8.8", 80))
		local_ip = s.getsockname()[0]
		s.close()
		return local_ip
	except Exception:
		return "Unable to determine"


def find_available_port(start_port=DEFAULT_PORT, max_attempts=10):
	"""Find an available port starting from start_port"""
	for port in range(start_port, start_port + max_attempts):
		try:
			with socketserver.TCPServer(("", port), None) as s:
				return port
		except OSError:
			continue
	return None


def print_qr_code(url):
	"""Generate and print a QR code using block characters"""
	try:
		import qrcode

		qr = qrcode.QRCode(
			version=1,
			error_correction=qrcode.constants.ERROR_CORRECT_L,
			box_size=1,
			border=1,
		)
		qr.add_data(url)
		qr.make(fit=True)

		# Get the QR code matrix
		matrix = qr.get_matrix()

		print("\nScan to connect:")
		for y in range(0, len(matrix), 2):
			line = ""
			for x in range(len(matrix[y])):
				top = matrix[y][x]
				bottom = matrix[y + 1][x] if y + 1 < len(matrix) else False
				if top and bottom:
					line += "█"
				elif top:
					line += "▀"
				elif bottom:
					line += "▄"
				else:
					line += " "
			print(line)
		print()
	except ImportError:
		print("\nQR code generation unavailable (qrcode library not installed)")
	except Exception as e:
		print(f"\nCould not generate QR code: {e}")

def start_server():
	"""Start the HTTP server (runs after venv is set up)"""
	# Change to script directory
	os.chdir(SCRIPT_DIR)

	# Find an available port
	port = find_available_port(DEFAULT_PORT)

	if port is None:
		print(f"Error: Could not find an available port (tried {DEFAULT_PORT}-{DEFAULT_PORT + 9})")
		sys.exit(1)

	# Get local IP for network access
	local_ip = get_local_ip()

	# Create server
	Handler = http.server.SimpleHTTPRequestHandler
	tracks_path = SCRIPT_DIR / "mix" / "tracks.json"

	# Suppress default logging and broken pipe errors
	class QuietHandler(Handler):
		def end_headers(self):
			self.send_header('Cache-Control', 'no-cache')
			super().end_headers()

		def log_message(self, format, *args):
			pass

		def handle(self):
			"""Handle requests and suppress broken pipe errors"""
			try:
				super().handle()
			except (BrokenPipeError, ConnectionResetError):
				# Browser cancelled the request (normal for media streaming/preloading)
				pass

		def do_POST(self):
			# Local-only: receive a reordered tracks array and overwrite tracks.json
			if self.path != '/tracks':
				self.send_response(404)
				self.end_headers()
				return
			length = int(self.headers.get('Content-Length', '0'))
			payload = json.loads(self.rfile.read(length))
			tracks_path.write_text(
				json.dumps(payload, indent='\t', ensure_ascii=False) + '\n',
				encoding='utf-8',
			)
			self.send_response(204)
			self.end_headers()

	try:
		with socketserver.ThreadingTCPServer(("", port), QuietHandler) as httpd:
			httpd.daemon_threads = True
			local_url = f"http://localhost:{port}"
			network_url = f"http://{local_ip}:{port}"

			print("=" * 60)
			print("💿 mixapps")
			print("=" * 60)
			print(f"\nServer running on port {port}")

			# Print QR code for easy mobile access
			print_qr_code(network_url)

			print(f"Local access:   {local_url}")
			print(f"Network access: {network_url}")
			print("\nPress Ctrl+C to stop the server")

			# Serve forever
			httpd.serve_forever()

	except KeyboardInterrupt:
		print("\n\nShutting down server...")
		sys.exit(0)
	except Exception as e:
		print(f"\nError starting server: {e}")
		sys.exit(1)


def main():
	"""Main entry point"""
	# Check if we're already running in venv
	if "--in-venv" not in sys.argv:
		run_in_venv()
	else:
		start_server()


if __name__ == "__main__":
	main()