gaming-image-files/home/stormux/.local/upload_server/uploader.py

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