gaming-image-files/home/stormux/.local/download_server.py

#!/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)