flaskboard.py

#!/usr/bin/env python3
"""
Flask-based local service status dashboard (interactive setup)
- On first run (or when invoked with --setup), prompts the user to select which local services to monitor.
- Detects init system (systemd or OpenRC).
- Dynamically saves selected services and ports to a simple JSON configuration file.
- Displays status in a Flask web UI.

Usage:
    pip install flask
    python3 flask_service_dashboard.py  # will prompt for setup if config missing
    python3 flask_service_dashboard.py --setup  # force setup
"""

import os
import json
import shutil
import subprocess
import platform
import time
from datetime import timedelta
from flask import Flask, jsonify, render_template_string, request

CONFIG_FILE = os.path.expanduser("~/.service_dashboard_config.json")
CMD_TIMEOUT = 2

app = Flask(__name__)

def which(cmd):
    return shutil.which(cmd) is not None

def detect_init_system():
    if os.path.isdir("/run/systemd/system") or which("systemctl"):
        return "systemd"
    if which("rc-service") or os.path.exists("/etc/init.d/"):
        return "openrc"
    return "unknown"

INIT_SYSTEM = detect_init_system()

def run_cmd(cmd):
    try:
        out = subprocess.check_output(cmd, stderr=subprocess.STDOUT, shell=True, timeout=CMD_TIMEOUT, universal_newlines=True)
        return out.strip()
    except subprocess.CalledProcessError as e:
        return e.output.strip() if e.output else ""
    except subprocess.TimeoutExpired:
        return "<timeout>"

def discover_services():
    candidates = set()
    if INIT_SYSTEM == "systemd":
        out = run_cmd("systemctl list-units --type=service --all --no-legend")
        for line in out.splitlines():
            name = line.split()[0]
            if name.endswith('.service'):
                candidates.add(name.replace('.service',''))
    elif INIT_SYSTEM == "openrc":
        out = run_cmd("rc-service -l")
        for line in out.split():
            candidates.add(line.strip())
    return sorted(candidates)

def setup_interactive():
    print("Detected init system:", INIT_SYSTEM)
    print("Scanning for services...\n")
    svc_list = discover_services()
    if not svc_list:
        print("No services detected.")
        return {}

    print("Select services to monitor (comma-separated indices):")
    for i, s in enumerate(svc_list, 1):
        print(f"[{i}] {s}")
    sel = input("Enter numbers: ").strip()
    selected = []
    try:
        for x in sel.split(','):
            idx = int(x.strip())
            if 1 <= idx <= len(svc_list):
                selected.append(svc_list[idx-1])
    except Exception:
        pass

    print("\nOptionally enter port mappings (e.g., ssh:22, nginx:80,443). Press Enter to skip.")
    ports_in = input("Ports: ").strip()
    port_map = {}
    if ports_in:
        for pair in ports_in.split(','):
            if ':' in pair:
                name, ports = pair.split(':',1)
                port_map[name.strip()] = [int(p) for p in ports.split(',') if p.isdigit()]

    cfg = {"services": selected, "ports": port_map}
    with open(CONFIG_FILE, 'w') as f:
        json.dump(cfg, f, indent=2)
    print(f"\nConfiguration saved to {CONFIG_FILE}")
    return cfg

def load_config():
    if not os.path.exists(CONFIG_FILE):
        return setup_interactive()
    with open(CONFIG_FILE) as f:
        return json.load(f)

CFG = load_config()
SERVICES = CFG.get("services", [])
COMMON_PORTS = CFG.get("ports", {})

def check_service_systemd(name):
    res = {"name": name, "active": None, "enabled": None, "mainpid": None, "raw": {}}
    active = run_cmd(f"systemctl is-active {name}")
    res["active"] = active or "unknown"
    enabled = run_cmd(f"systemctl is-enabled {name} 2>/dev/null || true")
    res["enabled"] = enabled or "unknown"
    mainpid = run_cmd(f"systemctl show -p MainPID --value {name} 2>/dev/null || true")
    res["mainpid"] = mainpid if mainpid and mainpid != '0' else None
    res["raw"]["status"] = run_cmd(f"systemctl status {name} --no-pager --full 2>/dev/null || true")
    return res

def check_service_openrc(name):
    res = {"name": name, "active": None, "enabled": None, "mainpid": None, "raw": {}}
    status_out = run_cmd(f"rc-service {name} status 2>&1 || true")
    if "started" in status_out.lower() or "running" in status_out.lower():
        res["active"] = "active"
    elif "stopped" in status_out.lower() or "not started" in status_out.lower():
        res["active"] = "inactive"
    else:
        res["active"] = status_out.strip() or "unknown"
    rl = run_cmd(f"rc-update show | grep -w {name} || true")
    res["enabled"] = "enabled" if name in rl else "disabled"
    res["raw"]["status"] = status_out
    return res

def check_service(name):
    if INIT_SYSTEM == "systemd":
        for candidate in (name, f"{name}.service"):
            info = check_service_systemd(candidate)
            if info["active"] != "unknown":
                info['queried_name'] = candidate
                return info
        return {"name": name, "active": "unknown", "enabled": "unknown", "mainpid": None, "raw": {}}
    elif INIT_SYSTEM == "openrc":
        return check_service_openrc(name)
    return {"name": name, "active": "unknown", "enabled": "unknown", "mainpid": None, "raw": {}}

def scan_listening_sockets():
    ports = set()
    if which("ss"):
        out = run_cmd("ss -tuln")
        for line in out.splitlines():
            parts = line.split()
            for p in parts:
                if ':' in p:
                    port = p.rsplit(':',1)[-1]
                    if port.isdigit():
                        ports.add(int(port))
    elif which("netstat"):
        out = run_cmd("netstat -tuln")
        for line in out.splitlines():
            parts = line.split()
            if len(parts) >= 4:
                addr = parts[3]
                if ':' in addr:
                    port = addr.rsplit(':',1)[-1]
                    if port.isdigit():
                        ports.add(int(port))
    return ports

def get_uptime():
    try:
        with open('/proc/uptime', 'r') as f:
            s = float(f.read().split()[0])
            return str(timedelta(seconds=int(s)))
    except Exception:
        return run_cmd('uptime -p')

INDEX_HTML = '''<html><head><meta charset="utf-8"><title>Service Dashboard</title>
<link rel="stylesheet" href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.2/dist/css/bootstrap.min.css">
</head><body class="p-3">
<h3>Local Service Dashboard</h3>
<p><strong>Host:</strong> {{ host }} &nbsp; <strong>Init:</strong> {{ init_system }} &nbsp; <strong>Uptime:</strong> <span id="uptime">{{ uptime }}</span></p>
<button id="refreshBtn" class="btn btn-sm btn-primary mb-2">Refresh</button>
<div id="services" class="row gy-2"></div>
<script>
async function fetchStatus(){const r=await fetch('/api/status');const j=await r.json();const c=document.getElementById('services');c.innerHTML='';for(const s of j.services){const up=s.active && s.active.startsWith('active');let ports='';if(s.ports){const p=s.ports.map(pt=>j.listening_ports.includes(pt)?`<span class=\"text-success\">${pt}</span>`:`<span class=\"text-danger\">${pt}</span>`);ports='Ports: '+p.join(', ');}c.innerHTML+=`<div class=\"col-md-4\"><div class=\"card\"><div class=\"card-body\"><h6>${s.name}</h6><div>Status: <span class=\"${up?'text-success':'text-danger'}\">${s.active}</span></div><div>Enabled: ${s.enabled}</div><div>${ports}</div></div></div></div>`;}}
document.getElementById('refreshBtn').addEventListener('click',fetchStatus);fetchStatus();setInterval(fetchStatus,10000);
</script></body></html>'''

@app.route('/')
def index():
    return render_template_string(INDEX_HTML, host=platform.node(), init_system=INIT_SYSTEM, uptime=get_uptime())

@app.route('/api/status')
def api_status():
    listening = scan_listening_sockets()
    services_out = []
    for s in SERVICES:
        info = check_service(s)
        info['ports'] = COMMON_PORTS.get(s, [])
        services_out.append(info)
    return {
        'server_time': int(time.time()),
        'host': platform.node(),
        'init_system': INIT_SYSTEM,
        'uptime': get_uptime(),
        'listening_ports': sorted(list(listening)),
        'services': services_out
    }

if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument('--host', default='0.0.0.0')
    parser.add_argument('--port', type=int, default=5000)
    parser.add_argument('--setup', action='store_true', help='run interactive setup')
    args = parser.parse_args()
    if args.setup:
        setup_interactive()
        exit(0)
    print(f"Starting dashboard on http://{args.host}:{args.port} (init: {INIT_SYSTEM})")
    app.run(host=args.host, port=args.port)