#!/usr/bin/env python3 import os import sys import socket import subprocess import threading import termios import tty import select import signal import http.server import socketserver from threading import Timer # Global variable to store server info for repeat announcement server_info = {'ip': '', 'port': 0, 'httpd': None} # Get local IP address def get_local_ip(): try: # Create a socket connection to determine the local IP s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # Doesn't even have to be reachable s.connect(('10.255.255.255', 1)) IP = s.getsockname()[0] s.close() except Exception: IP = '127.0.0.1' return IP # Announce server start with speech def announce_server_start(ip, port): formatted_ip = "" segments = ip.split(".") for i, segment in enumerate(segments): # Add each digit with space between formatted_ip += " ".join(segment) # Add " dot " between segments (but not after the last segment) if i < len(segments) - 1: formatted_ip += " dot " try: message = f"Download server running on {formatted_ip} port {port}" # Using spd-say with -Cw flag for better speech synthesis subprocess.run(['spd-say', '-Cw', message]) print(f"Download server running on {ip}:{port}") print("Press any key to repeat server information") print("Press 'q' to quit server") except Exception as e: print(f"Could not announce server start: {e}") print(f"Download server running on {ip}:{port}") # Function to repeat the server announcement def repeat_announcement(): if server_info['ip'] and server_info['port']: announce_server_start(server_info['ip'], server_info['port']) # Keyboard listener function def keyboard_listener(): """Listen for key presses and repeat server announcement""" if not sys.stdin.isatty(): print("Not running in a terminal, keyboard listener disabled") return try: # Save original terminal settings old_settings = termios.tcgetattr(sys.stdin) while True: try: # Set terminal to raw mode to capture individual key presses tty.setraw(sys.stdin.fileno()) # Use select to check if input is available (non-blocking) if select.select([sys.stdin], [], [], 1) == ([sys.stdin], [], []): # Read a single character key = sys.stdin.read(1) # Check for specific keys if key: if key.lower() == 'q': # Quit the server print("\nShutting down server...") try: subprocess.run(['spd-say', '-Cw', 'Download server shutting down']) except Exception: pass if server_info['httpd']: server_info['httpd'].shutdown() os._exit(0) else: # Any other key repeats the announcement repeat_announcement() except KeyboardInterrupt: # Handle Ctrl+C gracefully print("\nKeyboard listener interrupted") break except Exception as e: print(f"Error in keyboard listener: {e}") break finally: # Always restore terminal settings try: termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_settings) except: pass except Exception as e: print(f"Could not initialize keyboard listener: {e}") # Start keyboard listener in background thread def start_keyboard_listener(): """Start the keyboard listener in a daemon thread""" try: listener_thread = threading.Thread(target=keyboard_listener, daemon=True) listener_thread.start() except Exception as e: print(f"Could not start keyboard listener: {e}") # Signal handler for graceful shutdown def signal_handler(signum, frame): try: subprocess.run(['spd-say', '-Cw', 'Download server shutting down']) except Exception: pass print("\nDownload server shutting down...") if server_info['httpd']: server_info['httpd'].shutdown() sys.exit(0) # Custom HTTP handler to serve files from /home/stormux class DownloadHandler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): # Set the directory to serve files from super().__init__(*args, directory='/home/stormux', **kwargs) def log_message(self, format, *args): # Log downloads with speech announcement - be less chatty client_ip = self.client_address[0] if 'GET' in format % args: try: path = args[1] if len(args) > 1 else "unknown" # Only announce actual file downloads (not directories, not common web files) if (not path.endswith('/') and not any(x in path.lower() for x in ['.ico', '.css', '.js', '.png', '.jpg', '.gif', '.svg']) and path != '/' and path != '' and '?' not in path): filename = os.path.basename(path) # Only announce files with actual content extensions if any(filename.lower().endswith(ext) for ext in ['.mp3', '.ogg', '.flac', '.wav', '.nes', '.smc', '.do', '.dsk', '.pdf', '.txt', '.zip', '.tgz', '.tar']): subprocess.run(['spd-say', '-Cw', f'File {filename} downloaded'], timeout=3) except Exception: pass # Still print to console print(format % args) if __name__ == '__main__': print("Starting Download Server...") # Change to the home directory to serve files from there os.chdir('/home/stormux') # Get local IP and set port local_ip = get_local_ip() port = 8000 # Store server info for repeat announcements server_info['ip'] = local_ip server_info['port'] = port # Set up signal handlers for graceful shutdown signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) try: # Create HTTP server with socketserver.TCPServer(("", port), DownloadHandler) as httpd: server_info['httpd'] = httpd # Start keyboard listener for repeat announcements start_keyboard_listener() # Announce server start announce_server_start(local_ip, port) # Start serving files print(f"Serving files from /home/stormux at http://{local_ip}:{port}") httpd.serve_forever() except OSError as e: if e.errno == 98: # Address already in use try: subprocess.run(['spd-say', '-Cw', f'Port {port} is already in use']) except Exception: pass print(f"Error: Port {port} is already in use. Please stop any existing servers and try again.") else: try: subprocess.run(['spd-say', '-Cw', 'Download server failed to start']) except Exception: pass print(f"Error starting server: {e}") sys.exit(1) except Exception as e: try: subprocess.run(['spd-say', '-Cw', 'Download server error']) except Exception: pass print(f"Unexpected error: {e}") sys.exit(1)