nixos-configs/packages/app-launcher-server/app-launcher-server.py

#!/usr/bin/env python3

import json
import logging
import os
import subprocess
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse
import psutil

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Allowlisted applications that can be launched
ALLOWED_APPS = {
    'firefox': 'firefox',
    'kodi': 'kodi'
}

def is_app_running(app_name):
    """Check if an application is already running, returns (is_running, pid)"""
    command = ALLOWED_APPS.get(app_name)
    if not command:
        return False, None

    logger.debug(f"Looking for processes related to app '{app_name}' (command: '{command}')")

    for proc in psutil.process_iter(['name', 'cmdline', 'pid']):
        try:
            proc_name = proc.info['name']
            cmdline = proc.info['cmdline'] or []

            logger.debug(f"Checking process PID {proc.info['pid']}: name='{proc_name}', cmdline={cmdline}")

            # Check multiple patterns for the application:
            # 1. Process name exactly matches command
            # 2. Process name contains the command (e.g., "kodi.bin" contains "kodi")
            # 3. Command line starts with the command
            # 4. Command line contains the wrapped version (e.g., ".kodi-wrapped")
            # 5. Any command line argument ends with the command executable

            matches = False
            match_reason = ""

            if proc_name == command:
                matches = True
                match_reason = f"exact process name match: '{proc_name}'"
            elif command in proc_name:
                matches = True
                match_reason = f"process name contains command: '{proc_name}' contains '{command}'"
            elif cmdline and cmdline[0] == command:
                matches = True
                match_reason = f"exact cmdline match: '{cmdline[0]}'"
            elif cmdline and cmdline[0].endswith('/' + command):
                matches = True
                match_reason = f"cmdline path ends with command: '{cmdline[0]}'"
            elif cmdline and any(f'.{command}-wrapped' in arg for arg in cmdline):
                matches = True
                match_reason = f"wrapped command in cmdline: {cmdline}"
            elif cmdline and any(f'{command}.bin' in arg for arg in cmdline):
                matches = True
                match_reason = f"binary command in cmdline: {cmdline}"

            if matches:
                logger.info(f"Found running {app_name} process: PID {proc.info['pid']} ({match_reason})")
                return True, proc.info['pid']

        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            continue

    logger.debug(f"No running process found for {app_name}")
    return False, None

class AppLauncherHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        logger.info(format % args)

    def do_GET(self):
        if self.path == '/':
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {
                'status': 'running',
                'available_apps': list(ALLOWED_APPS.keys()),
                'usage': 'POST /launch/<app_name> to launch an application'
            }
            self.wfile.write(json.dumps(response, indent=2).encode())
        else:
            self.send_error(404)

    def do_POST(self):
        parsed_path = urlparse(self.path)
        path_parts = parsed_path.path.strip('/').split('/')

        if len(path_parts) == 2 and path_parts[0] == 'launch':
            app_name = path_parts[1]
            self.launch_app(app_name)
        else:
            self.send_error(404, "Invalid endpoint. Use /launch/<app_name>")

    def launch_app(self, app_name):
        if app_name not in ALLOWED_APPS:
            self.send_error(400, f"Application '{app_name}' not allowed. Available apps: {list(ALLOWED_APPS.keys())}")
            return

        command = ALLOWED_APPS[app_name]

        # Check if app is already running
        is_running, existing_pid = is_app_running(app_name)
        if is_running:
            logger.info(f"Application {app_name} is already running (PID: {existing_pid}), skipping launch")
            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {
                'status': 'success',
                'message': f'{app_name} is already running',
                'pid': existing_pid,
                'already_running': True
            }
            self.wfile.write(json.dumps(response).encode())
            return

        try:
            # Launch the application in the background
            # Ensure we have the proper environment for GUI apps
            env = os.environ.copy()

            logger.info(f"Launching application: {command}")
            process = subprocess.Popen(
                [command],
                env=env,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True
            )

            self.send_response(200)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            response = {
                'status': 'success',
                'message': f'Successfully launched {app_name}',
                'pid': process.pid,
                'already_running': False
            }
            self.wfile.write(json.dumps(response).encode())

        except FileNotFoundError:
            logger.error(f"Application not found: {command}")
            self.send_error(500, f"Application '{app_name}' not found on system")
        except Exception as e:
            logger.error(f"Error launching {command}: {e}")
            self.send_error(500, f"Failed to launch {app_name}: {str(e)}")

def main():
    port = int(sys.argv[1]) if len(sys.argv) > 1 else 8081

    server = HTTPServer(('0.0.0.0', port), AppLauncherHandler)
    logger.info(f"App launcher server starting on port {port}")
    logger.info(f"Available applications: {list(ALLOWED_APPS.keys())}")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("Server shutting down...")
        server.server_close()

if __name__ == '__main__':
    main()