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 datetime import date
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'
}

# Workout card base URL
WORKOUT_CARD_BASE_URL = 'https://ogle.fyi/ash/workout'

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()),
                'endpoints': {
                    'POST /launch/<app_name>': 'Launch an application (optional JSON body: {"args": ["url"]})',
                    'POST /workout': 'Open today\'s workout card in Firefox'
                }
            }
            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)
        elif len(path_parts) == 1 and path_parts[0] == 'workout':
            self.open_workout_card()
        else:
            self.send_error(404, "Invalid endpoint. Use /launch/<app_name> or /workout")

    def read_post_body(self):
        """Read and parse JSON body from POST request, return dict or empty dict."""
        content_length = int(self.headers.get('Content-Length', 0))
        if content_length > 0:
            try:
                body = self.rfile.read(content_length)
                return json.loads(body.decode('utf-8'))
            except (json.JSONDecodeError, UnicodeDecodeError) as e:
                logger.warning(f"Failed to parse POST body as JSON: {e}")
        return {}

    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]

        # Read optional args from POST body
        body = self.read_post_body()
        extra_args = body.get('args', [])
        # Validate args are strings
        if not isinstance(extra_args, list) or not all(isinstance(a, str) for a in extra_args):
            self.send_error(400, "'args' must be a list of strings")
            return

        full_command = [command] + extra_args

        # Check if app is already running
        is_running, existing_pid = is_app_running(app_name)
        if is_running:
            # If extra args provided, still launch a new instance (e.g., open a URL)
            if extra_args:
                logger.info(f"Application {app_name} already running (PID: {existing_pid}), but extra args provided — launching new instance")
            else:
                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: {' '.join(full_command)}")
            process = subprocess.Popen(
                full_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 open_workout_card(self):
        """Open today's workout card in Firefox."""
        today = date.today().strftime('%Y-%m-%d')
        url = f"{WORKOUT_CARD_BASE_URL}/{today}.html"
        logger.info(f"Opening workout card for {today}: {url}")

        # Always launch Firefox with the URL, even if already running
        command = ALLOWED_APPS['firefox']
        env = os.environ.copy()

        try:
            process = subprocess.Popen(
                [command, url],
                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'Opened workout card in Firefox',
                'url': url,
                'pid': process.pid
            }
            self.wfile.write(json.dumps(response).encode())

        except FileNotFoundError:
            logger.error(f"Firefox not found: {command}")
            self.send_error(500, "Firefox not found on system")
        except Exception as e:
            logger.error(f"Error launching Firefox with workout card: {e}")
            self.send_error(500, f"Failed to open workout card: {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())}")
    logger.info(f"Workout card URL: {WORKOUT_CARD_BASE_URL}/<date>.html")

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

if __name__ == '__main__':
    main()