#!/usr/bin/env python3 from flask import Flask, request, redirect, url_for, render_template, flash, jsonify import os import mimetypes import shutil import subprocess import socket import sys import re import logging import threading import termios import tty import select import tempfile from threading import Timer from werkzeug.utils import secure_filename app = Flask(__name__) app.secret_key = os.urandom(24) # For flash messages # Configure logging def setup_logging(): """Configure logging for the upload server""" logDir = os.path.expanduser("~/.local/log/stormux") os.makedirs(logDir, exist_ok=True) logFile = os.path.join(logDir, "upload_server.log") logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(logFile), logging.StreamHandler() ] ) return logging.getLogger(__name__) logger = setup_logging() # Configuration UPLOAD_FOLDER = '/tmp/uploads' # Temporary storage FINAL_DESTINATIONS = { 'apple': '/home/stormux/.local/games/apple2e/disks', 'NES': '/home/stormux/Roms/Nintendo Entertainment System', 'SNES': '/home/stormux/Roms/Super Nintendo Entertainment System', 'SMS': '/home/stormux/Roms/Sega Master System', 'music': '/home/stormux/Music', 'audiobook': '/home/stormux/Library', 'voxin': '/home/stormux/Downloads', 'other': '/home/stormux/Downloads' } # Allowed file extensions by type (can be easily expanded) FILE_TYPES = { 'apple': ['.do', '.dsk'], 'NES': ['.nes'], 'SNES': ['.fig', '.smc'], 'SMS': ['.sms'], 'music': ['.mp3', '.ogg', '.flac', '.wav', '.m4a', '.opus'], 'audiobook': ['.m4b', '.epub', '.zip'], 'voxin': ['.tgz'], 'other': [] # Empty list to represent any other file type } # 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 # Global variable to store server info for repeat announcement server_info = {'ip': '', 'port': 0} # 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"Upload server running on {formatted_ip} port {port}" # Using spd-say with -Cw flag for better speech synthesis subprocess.run(['spd-say', '-Cw', message]) logger.info(f"Announced server start: {message}") except Exception as e: logger.error(f"Could not announce server start: {e}") print(f"Could not announce server start: {e}") # 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(): logger.info("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) # Any key press will trigger the announcement if key: logger.info(f"Key pressed: {repr(key)}, repeating server announcement") repeat_announcement() except KeyboardInterrupt: # Handle Ctrl+C gracefully logger.info("Keyboard listener interrupted") break except Exception as e: logger.error(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: logger.error(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() logger.info("Keyboard listener started") except Exception as e: logger.error(f"Could not start keyboard listener: {e}") # Set file permissions def set_file_permissions(filepath): try: # Set permissions to 644 (rw-r--r--) os.chmod(filepath, 0o644) return True except OSError as e: print(f"Warning: Could not set permissions for {filepath}: {e}") return False # Ensure all directories exist and have correct permissions def setup_directories(): logger.info("Setting up upload directories") # Create temp directory if not os.path.exists(UPLOAD_FOLDER): try: os.makedirs(UPLOAD_FOLDER) # Set permissions to 755 os.chmod(UPLOAD_FOLDER, 0o755) logger.info(f"Created temp directory: {UPLOAD_FOLDER}") except OSError as e: logger.error(f"Could not create or set permissions for {UPLOAD_FOLDER}: {e}") print(f"Warning: Could not create or set permissions for {UPLOAD_FOLDER}: {e}") # Create all destination directories if they don't exist for directory in FINAL_DESTINATIONS.values(): if not os.path.exists(directory): try: os.makedirs(directory) # Set permissions to 755 os.chmod(directory, 0o755) logger.info(f"Created destination directory: {directory}") except OSError as e: logger.error(f"Could not create or set permissions for {directory}: {e}") print(f"Warning: Could not create or set permissions for {directory}: {e}") # Create directory if it doesn't exist def ensure_directory_exists(directory): if not os.path.exists(directory): try: os.makedirs(directory) # Set permissions to 755 os.chmod(directory, 0o755) return True except OSError as e: print(f"Warning: Could not create or set permissions for {directory}: {e}") return False return True # Get file type from extension def get_file_type(filename): ext = os.path.splitext(filename.lower())[1] for file_type, extensions in FILE_TYPES.items(): if ext in extensions: return file_type return None @app.route('/') def index(): return render_template('index.html', file_types=FILE_TYPES.keys()) @app.route('/upload', methods=['POST']) def upload_file(): client_ip = request.remote_addr logger.info(f"File upload request from {client_ip}") if 'file' not in request.files: logger.warning(f"Upload request from {client_ip} missing file part") flash('No file part', 'error') return redirect(request.url) files = request.files.getlist('file') logger.info(f"Received {len(files)} files from {client_ip}") uploaded_files = [] unknown_files = [] voxin_results = [] for file in files: if file.filename == '': continue filename = secure_filename(file.filename) logger.info(f"Processing file: {filename} from {client_ip}") file_path = os.path.join(UPLOAD_FOLDER, filename) file.save(file_path) # Set file permissions set_file_permissions(file_path) # Check if it's a voxin package by filename pattern if is_voxin_package(filename): logger.info(f"Detected voxin package: {filename}") # Move to Downloads directory dest_dir = FINAL_DESTINATIONS['voxin'] dest_path = os.path.join(dest_dir, filename) shutil.move(file_path, dest_path) set_file_permissions(dest_path) # Process the voxin package result = handle_voxin_package(dest_path) logger.info(f"Voxin package processing result for {filename}: {result['success']}") voxin_results.append({ 'name': filename, 'type': 'voxin', 'success': result['success'], 'message': result['message'], 'details': result.get('details', '') }) else: # Handle regular files file_type = get_file_type(filename) if file_type: logger.info(f"File {filename} categorized as: {file_type}") # Move to appropriate directory dest_dir = FINAL_DESTINATIONS.get(file_type, FINAL_DESTINATIONS['other']) dest_path = os.path.join(dest_dir, filename) shutil.move(file_path, dest_path) set_file_permissions(dest_path) uploaded_files.append({'name': filename, 'type': file_type}) else: logger.info(f"File {filename} requires manual classification") # Keep in temporary folder for classification unknown_files.append({'name': filename, 'path': file_path}) # Handle results if voxin_results: # If there were voxin packages, show their installation results return render_template('voxin_results.html', results=voxin_results) elif unknown_files: # If there are files needing classification return render_template('classify.html', unknown_files=unknown_files, file_types=FILE_TYPES.keys()) else: # All other files flash('All files uploaded successfully!', 'success') return render_template('success.html', uploaded_files=uploaded_files) @app.route('/upload_music', methods=['POST']) def upload_music(): if 'folder' not in request.files: flash('No folder selected', 'error') return redirect('/') files = request.files.getlist('folder') if len(files) == 0 or files[0].filename == '': flash('No files in selected folder', 'error') return redirect('/') # Process each file processed_files = [] skipped_files = [] # Get the root folder name from the first file first_file = files[0] if '/' not in first_file.filename: flash('Invalid folder structure in uploaded files', 'error') return redirect('/') # Extract the root folder name root_folder_name = secure_filename(first_file.filename.split('/')[0]) # Create the main folder in Music directory music_root_dir = os.path.join(FINAL_DESTINATIONS['music'], root_folder_name) ensure_directory_exists(music_root_dir) for file in files: if file.filename == '': continue # Get the relative path within the folder rel_path = file.filename # Skip if there's no path structure if '/' not in rel_path: skipped_files.append(rel_path) continue # Extract folder structure: "FolderName/SubFolder/file.mp3" path_parts = rel_path.split('/') # Create secure path components secure_parts = [secure_filename(part) for part in path_parts] # We keep ALL parts including the root folder for the final path # BUT we need to recreate the structure inside the Music directory rel_dest_path = '/'.join(secure_parts) # Create the full destination path - put directly in Music/root_folder_name/... dest_path = os.path.join(FINAL_DESTINATIONS['music'], rel_dest_path) dest_dir = os.path.dirname(dest_path) # Ensure the destination directory exists ensure_directory_exists(dest_dir) # Save to temp location first temp_path = os.path.join(UPLOAD_FOLDER, secure_parts[-1]) file.save(temp_path) set_file_permissions(temp_path) # Move to final location try: shutil.move(temp_path, dest_path) set_file_permissions(dest_path) processed_files.append({ 'name': rel_dest_path, 'type': 'music', 'success': True }) except Exception as e: print(f"Error moving file {temp_path} to {dest_path}: {e}") processed_files.append({ 'name': rel_dest_path, 'type': 'music', 'success': False, 'error': str(e) }) # Provide feedback based on results success_count = sum(1 for file in processed_files if file.get('success', False)) if success_count > 0: flash(f'Music folder "{root_folder_name}" uploaded successfully! {success_count} files processed.', 'success') else: flash('No files were successfully processed. Please check the folder structure.', 'error') return render_template('success.html', uploaded_files=processed_files) # Add a function to check if a file is a voxin package def is_voxin_package(filename): """Check if the filename starts with 'voxin-' and ends with '.tgz'.""" return filename.lower().startswith('voxin-') and filename.lower().endswith('.tgz') def find_voxin_installer(extract_dir): """Return the Voxin installer path from an extracted package, if present.""" for root, dirs, files in os.walk(extract_dir): if 'voxin-installer.sh' in files: return os.path.join(root, 'voxin-installer.sh') return None # Add function to handle voxin installation def handle_voxin_package(file_path): """ Process a voxin package: 1. Extract the package 2. Run the installer script with -l option 3. Test the new voice Returns: dict with status and message """ extract_dir = None try: filename = os.path.basename(file_path) logger.info(f"Starting voxin package processing: {filename}") extract_dir = tempfile.mkdtemp(prefix='voxin_extract_') # Extract the tarball logger.info(f"Extracting voxin package: {filename}") subprocess.run(['tar', 'xf', file_path, '-C', extract_dir], check=True) # Find the installer script installer_path = find_voxin_installer(extract_dir) if not installer_path: logger.error(f"Voxin installer script not found in package: {filename}") return { 'success': False, 'message': 'Voxin installer script not found in package.' } # Make the installer executable os.chmod(installer_path, 0o755) # Run the installer with -l option. /etc/speech-dispatcher/speechd.conf # is a symlink into /home/stormux/.local/etc on Stormux images. logger.info(f"Running voxin installer for: {filename}") install_result = subprocess.run( ['sudo', str(installer_path), '-l'], capture_output=True, text=True, timeout=180 # 3 minute timeout ) if install_result.returncode != 0: logger.error(f"Voxin installation failed for {filename}: {install_result.stderr}") return { 'success': False, 'message': 'Voxin installation failed.', 'details': install_result.stderr } # Restart speech-dispatcher subprocess.run(['killall', 'speech-dispatcher'], stderr=subprocess.DEVNULL) # Ignore errors if not running # Wait a moment for service to fully terminate subprocess.run(['sleep', '2']) # Test the voice test_message = "If you hear this message, Voxin installation was successful, and you may now change the voice from the system menu." test_result = subprocess.run( ['spd-say', '-o', 'voxin', test_message], capture_output=True, text=True ) logger.info(f"Voxin package installation completed successfully: {filename}") return { 'success': True, 'message': 'Voice package installed successfully!', 'details': 'The test message should have been spoken using the voxin voice system.' } except subprocess.TimeoutExpired: logger.error(f"Voxin installation timed out for: {filename}") return { 'success': False, 'message': 'Voice installation timed out.', 'details': 'The installation process took too long and was terminated.' } except Exception as e: logger.error(f"Voxin installation error for {filename}: {e}") return { 'success': False, 'message': 'Voice installation error.', 'details': str(e) } finally: if extract_dir and os.path.exists(extract_dir): shutil.rmtree(extract_dir) @app.route('/classify', methods=['POST']) def classify_files(): results = [] for filename, file_type in request.form.items(): if filename == 'csrf_token': continue file_path = os.path.join(UPLOAD_FOLDER, filename) if os.path.exists(file_path): # Move to selected directory dest_dir = FINAL_DESTINATIONS.get(file_type, FINAL_DESTINATIONS['other']) dest_path = os.path.join(dest_dir, filename) shutil.move(file_path, dest_path) set_file_permissions(dest_path) results.append({'name': filename, 'type': file_type, 'success': True}) else: results.append({'name': filename, 'success': False}) flash('Files classified and moved successfully!', 'success') return render_template('success.html', uploaded_files=results) @app.route('/check_local', methods=['GET']) def check_local(): # Get client IP client_ip = request.remote_addr # Define local networks local_networks = ['127.0.0.1', '192.168.', '10.', '172.16.', '172.17.', '172.18.', '172.19.', '172.20.', '172.21.', '172.22.', '172.23.', '172.24.', '172.25.', '172.26.', '172.27.', '172.28.', '172.29.', '172.30.', '172.31.'] # Check if IP is local is_local = any(client_ip.startswith(network) for network in local_networks) if is_local: return jsonify({'status': 'success', 'local': True}) else: return jsonify({'status': 'error', 'local': False}), 403 # Add a shutdown route to close the application @app.route('/shutdown', methods=['GET', 'POST']) def shutdown(): # Function to shutdown the server def shutdown_server(): # Announce shutdown try: # Using spd-say with -Cw flag for better speech synthesis subprocess.run(['spd-say', '-Cw', 'Server shutting down']) except Exception: pass # Force exit with os._exit which is more aggressive than sys.exit os._exit(0) # Render confirmation page on GET if request.method == 'GET': return render_template('shutdown.html') # Process shutdown on POST flash('Server is shutting down...', 'info') # Use a shorter timer to give time for the response to be sent Timer(0.5, shutdown_server).start() return render_template('shutdown_confirm.html') # Middleware to check if request is from local network @app.before_request def check_local_network(): # Skip for certain routes if request.endpoint == 'check_local': return # Get client IP client_ip = request.remote_addr # Define local networks local_networks = ['127.0.0.1', '192.168.', '10.', '172.16.', '172.17.', '172.18.', '172.19.', '172.20.', '172.21.', '172.22.', '172.23.', '172.24.', '172.25.', '172.26.', '172.27.', '172.28.', '172.29.', '172.30.', '172.31.'] # Check if IP is local is_local = any(client_ip.startswith(network) for network in local_networks) if not is_local: return render_template('error.html', message="Access denied: Local network only"), 403 if __name__ == '__main__': logger.info("Starting upload server") # Set up directories setup_directories() # Get local IP and port local_ip = get_local_ip() port = 5000 # Store server info for repeat announcements server_info['ip'] = local_ip server_info['port'] = port logger.info(f"Server will run on {local_ip}:{port}") # Start keyboard listener for repeat announcements start_keyboard_listener() # Announce server start announce_server_start(local_ip, port) # Run the server logger.info("Upload server started successfully") app.run(host='0.0.0.0', port=port, debug=True)