#!/usr/bin/env python3 """ Gestionnaire d'applications Docker sur VPS Debian. Navigation au clavier (↑↓ + Entrée), sans dépendance externe. """ import os import subprocess import sys import termios import tty from pathlib import Path # ─── Configuration ──────────────────────────────────────────────────────────── APPS_BASE_DIR = "/home" COMPOSE_FILES = ["compose.yml", "compose.yaml", "docker-compose.yml", "docker-compose.yaml"] # ─── ANSI ───────────────────────────────────────────────────────────────────── R = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" CYN = "\033[36m" YEL = "\033[33m" GRN = "\033[32m" RED = "\033[31m" SEL = "\033[44m\033[97m" # fond bleu + texte blanc W = 56 # largeur de la boîte # ─── Gestion du clavier ─────────────────────────────────────────────────────── KEY_UP = "\x1b[A" KEY_DOWN = "\x1b[B" KEY_ENTER = ("\r", "\n") KEY_QUIT = ("q", "Q", "\x1b", "\x03") # q, ESC, Ctrl+C def getch() -> str: """Lit un seul appui clavier sans attendre Entrée.""" fd = sys.stdin.fileno() old = termios.tcgetattr(fd) try: tty.setraw(fd) ch = sys.stdin.read(1) if ch == "\x1b": nxt = sys.stdin.read(1) if nxt == "[": return "\x1b[" + sys.stdin.read(1) return "\x1b" return ch finally: termios.tcsetattr(fd, termios.TCSADRAIN, old) def hide_cursor() -> None: print("\033[?25l", end="", flush=True) def show_cursor() -> None: print("\033[?25h", end="", flush=True) # ─── Rendu ──────────────────────────────────────────────────────────────────── def draw_header(title: str, subtitle: str = "") -> None: print(f"{CYN}╔{'═' * (W - 2)}╗{R}") print(f"{CYN}║{R}{BOLD}{title.center(W - 2)}{R}{CYN}║{R}") print(f"{CYN}╠{'═' * (W - 2)}╣{R}") if subtitle: pad = W - 2 - len(subtitle) print(f"{CYN}║{R} {YEL}{subtitle}{R}{' ' * max(0, pad - 1)}{CYN}║{R}") print(f"{CYN}╠{'─' * (W - 2)}╣{R}") def draw_footer() -> None: print(f"{CYN}╠{'─' * (W - 2)}╣{R}") hint = f" {YEL}↑↓{R} naviguer {YEL}↵{R} sélectionner {YEL}q{R} retour" print(f"{CYN}║{R}{hint}") print(f"{CYN}╚{'═' * (W - 2)}╝{R}") def draw_item(index: int, label: str, selected: bool, badge: str = "") -> None: num = f"[{index + 1}]" badge_str = f" {GRN}({badge}){R}" if badge else "" text = f" {num} {label}" visible_len = len(text) + (len(badge) + 3 if badge else 0) padding = " " * max(0, W - 4 - visible_len) if selected: print(f"{CYN}║{R}{SEL} › {text}{R}{GRN if badge else ''}{badge_str}{R}{SEL}{padding} {R}{CYN}║{R}") else: print(f"{CYN}║{R} {text}{badge_str}{padding} {CYN}║{R}") # ─── Menu générique ─────────────────────────────────────────────────────────── def navigate_menu( title: str, items: list[str], subtitle: str = "", badges: list[str] | None = None, ) -> int | None: """ Menu interactif à navigation clavier (↑↓ + Entrée). Retourne l'index sélectionné, ou None si l'utilisateur quitte. """ current = 0 badges = badges or [""] * len(items) while True: os.system("clear") hide_cursor() draw_header(title, subtitle) for i, item in enumerate(items): draw_item(i, item, i == current, badges[i]) draw_footer() key = getch() if key == KEY_UP: current = (current - 1) % len(items) elif key == KEY_DOWN: current = (current + 1) % len(items) elif key in KEY_ENTER: show_cursor() return current elif key in KEY_QUIT: show_cursor() return None elif key.isdigit(): idx = int(key) - 1 if 0 <= idx < len(items): show_cursor() return idx def pause(message: str = "Appuyez sur une touche pour continuer…") -> None: show_cursor() print(f"\n {DIM}{message}{R}", end="", flush=True) getch() # ─── Détection Docker ───────────────────────────────────────────────────────── def _apps_from_docker_ps() -> list[dict]: """Détecte les projets Compose via les labels des conteneurs docker ps -a.""" apps = [] seen_dirs: set[str] = set() try: result = subprocess.run( [ "docker", "ps", "-a", "--format", "{{.Label \"com.docker.compose.project\"}}\t" "{{.Label \"com.docker.compose.project.working_dir\"}}\t" "{{.Label \"com.docker.compose.project.config_files\"}}", ], capture_output=True, text=True, ) except FileNotFoundError: return [] for line in result.stdout.splitlines(): parts = line.split("\t") if len(parts) != 3: continue project, working_dir, config_files = (p.strip() for p in parts) if not project or not working_dir or working_dir in seen_dirs: continue compose_file = config_files.split(",")[0].strip() if config_files else "" if not compose_file or not Path(compose_file).exists(): compose_file = "" for name in COMPOSE_FILES: candidate = Path(working_dir) / name if candidate.exists(): compose_file = str(candidate) break if not compose_file: continue seen_dirs.add(working_dir) apps.append({"name": project, "path": working_dir, "compose_file": compose_file}) return apps def _apps_from_filesystem(exclude_dirs: set) -> list[dict]: """Fallback : scan récursif /home/*/ (profondeur 3) pour trouver des fichiers compose.""" apps = [] base = Path(APPS_BASE_DIR) def search(directory: Path, app_name: str, depth: int) -> None: if depth == 0: return try: for compose_name in COMPOSE_FILES: compose_path = directory / compose_name if compose_path.exists() and str(directory) not in exclude_dirs: rel = directory.relative_to(base / app_name) label = app_name if rel == Path(".") else f"{app_name}/{rel}" apps.append({"name": label, "path": str(directory), "compose_file": str(compose_path)}) exclude_dirs.add(str(directory)) return for sub in sorted(directory.iterdir()): if sub.is_dir(): search(sub, app_name, depth - 1) except PermissionError: pass try: for entry in sorted(base.iterdir()): if entry.is_dir(): search(entry, entry.name, depth=3) except PermissionError: show_cursor() print(f"Erreur : permission refusée pour lire {APPS_BASE_DIR}") sys.exit(1) return apps def find_docker_apps() -> list[dict]: """Détecte les apps Compose via docker ps (labels) puis complète avec le filesystem.""" apps = _apps_from_docker_ps() seen = {a["path"] for a in apps} apps += _apps_from_filesystem(seen) return sorted(apps, key=lambda a: a["name"].lower()) # ─── Actions Docker ─────────────────────────────────────────────────────────── def get_running_containers(app: dict) -> list[str]: try: result = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "ps", "--services", "--filter", "status=running"], capture_output=True, text=True, cwd=app["path"], ) return [s.strip() for s in result.stdout.splitlines() if s.strip()] except FileNotFoundError: return [] def get_all_services(app: dict) -> list[str]: try: result = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "config", "--services"], capture_output=True, text=True, cwd=app["path"], ) return [s.strip() for s in result.stdout.splitlines() if s.strip()] except FileNotFoundError: return [] def get_container_id(app: dict, service: str) -> str | None: try: result = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "ps", "-q", service], capture_output=True, text=True, cwd=app["path"], ) cid = result.stdout.strip() return cid if cid else None except FileNotFoundError: return None def action_bash(app: dict, service: str) -> None: container_id = get_container_id(app, service) if not container_id: print(f"\n {RED}Conteneur '{service}' introuvable ou arrêté.{R}") pause() return show_cursor() os.system("clear") print(f"{CYN}{'─' * W}{R}") print(f" Shell bash → {BOLD}{app['name']}{R} / {service}") print(f" {DIM}(tapez 'exit' pour revenir au menu){R}") print(f"{CYN}{'─' * W}{R}\n") subprocess.run(["docker", "exec", "-it", container_id, "bash"]) def action_logs(app: dict, service: str) -> None: container_id = get_container_id(app, service) if not container_id: print(f"\n {RED}Conteneur '{service}' introuvable ou arrêté.{R}") pause() return show_cursor() os.system("clear") print(f"{CYN}{'─' * W}{R}") print(f" Logs → {BOLD}{app['name']}{R} / {service}") print(f" {DIM}(Ctrl+C pour arrêter){R}") print(f"{CYN}{'─' * W}{R}\n") try: subprocess.run(["docker", "logs", "--tail", "100", "-f", container_id]) except KeyboardInterrupt: print(f"\n {YEL}Suivi interrompu.{R}") pause() def action_update(app: dict) -> None: show_cursor() os.system("clear") print(f"{CYN}{'─' * W}{R}") print(f" Mise à jour → {BOLD}{app['name']}{R}") print(f"{CYN}{'─' * W}{R}\n") print(f" {YEL}→ docker compose pull{R}\n") result_pull = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "pull"], cwd=app["path"], ) if result_pull.returncode != 0: print(f"\n {RED}Erreur lors du pull.{R}") pause() return print(f"\n {YEL}→ docker compose up -d{R}\n") result_up = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "up", "-d"], cwd=app["path"], ) if result_up.returncode == 0: print(f"\n {GRN}Mise à jour terminée avec succès.{R}") else: print(f"\n {RED}Erreur lors du redémarrage.{R}") pause() # ─── Menus ──────────────────────────────────────────────────────────────────── def service_menu(app: dict, action: str) -> None: """Sélection d'un service puis exécution de l'action (bash ou logs).""" services = get_all_services(app) if not services: print(f"\n {RED}Aucun service trouvé dans le compose.{R}") pause() return running = set(get_running_containers(app)) badges = ["actif" if s in running else "" for s in services] label = "Shell bash" if action == "bash" else "Logs" idx = navigate_menu( title=" Gestionnaire Docker VPS ", subtitle=f"{app['name']} › {label}", items=services, badges=badges, ) if idx is None: return if action == "bash": action_bash(app, services[idx]) else: action_logs(app, services[idx]) def app_menu(app: dict) -> None: """Menu des actions pour une application.""" actions = [ "Shell bash dans un conteneur", "Voir les logs d'un conteneur", "Mettre à jour (pull + up -d)", ] while True: running = get_running_containers(app) if running: status = f"{len(running)} service(s) actif(s)" else: status = "aucun service actif" idx = navigate_menu( title=" Gestionnaire Docker VPS ", subtitle=f"{app['name']} · {status}", items=actions, ) if idx is None: return elif idx == 0: service_menu(app, "bash") elif idx == 1: service_menu(app, "logs") elif idx == 2: action_update(app) def main() -> None: if os.geteuid() != 0: print(f"{YEL}Attention : droits root recommandés (sudo python3 manage_vps.py){R}\n") try: while True: apps = find_docker_apps() if not apps: os.system("clear") print(f"\n {RED}Aucune application Docker trouvée sous {APPS_BASE_DIR}/{R}") print(f" {DIM}(vérifiez que des fichiers compose.yml existent){R}\n") sys.exit(0) idx = navigate_menu( title=" Gestionnaire Docker VPS ", subtitle=f"{len(apps)} application(s) détectée(s)", items=[a["name"] for a in apps], ) if idx is None: os.system("clear") print(f"\n Au revoir !\n") sys.exit(0) app_menu(apps[idx]) except KeyboardInterrupt: show_cursor() os.system("clear") print("\n Au revoir !\n") sys.exit(0) if __name__ == "__main__": main() def _apps_from_docker_ps() -> list[dict]: """Détecte les projets Compose via les labels des conteneurs docker ps -a.""" apps = [] seen_dirs = set() try: result = subprocess.run( [ "docker", "ps", "-a", "--format", "{{.Label \"com.docker.compose.project\"}}\t" "{{.Label \"com.docker.compose.project.working_dir\"}}\t" "{{.Label \"com.docker.compose.project.config_files\"}}", ], capture_output=True, text=True, ) except FileNotFoundError: return [] # docker non installé for line in result.stdout.splitlines(): parts = line.split("\t") if len(parts) != 3: continue project, working_dir, config_files = (p.strip() for p in parts) if not project or not working_dir or working_dir in seen_dirs: continue # Prend le premier fichier de config listé compose_file = config_files.split(",")[0].strip() if config_files else "" if not compose_file or not Path(compose_file).exists(): # Cherche un fichier compose connu dans le working_dir compose_file = "" for name in COMPOSE_FILES: candidate = Path(working_dir) / name if candidate.exists(): compose_file = str(candidate) break if not compose_file: continue seen_dirs.add(working_dir) apps.append({ "name": project, "path": working_dir, "compose_file": compose_file, }) return apps def _apps_from_filesystem(exclude_dirs: set) -> list[dict]: """Fallback : scanne /home/*/ récursivement (profondeur 3) pour trouver des fichiers compose.""" apps = [] base = Path(APPS_BASE_DIR) def search(directory: Path, app_name: str, depth: int): if depth == 0: return try: for compose_name in COMPOSE_FILES: compose_path = directory / compose_name if compose_path.exists() and str(directory) not in exclude_dirs: rel = directory.relative_to(base / app_name) label = app_name if rel == Path(".") else f"{app_name}/{rel}" apps.append({ "name": label, "path": str(directory), "compose_file": str(compose_path), }) exclude_dirs.add(str(directory)) return for sub in sorted(directory.iterdir()): if sub.is_dir(): search(sub, app_name, depth - 1) except PermissionError: pass try: for entry in sorted(base.iterdir()): if entry.is_dir(): search(entry, entry.name, depth=3) except PermissionError: print(f" Erreur : permission refusée pour lire {APPS_BASE_DIR}") sys.exit(1) return apps def find_docker_apps() -> list[dict]: """Détecte les apps Compose via docker ps (labels) puis complète avec le filesystem.""" apps = _apps_from_docker_ps() seen_dirs = {a["path"] for a in apps} apps += _apps_from_filesystem(seen_dirs) return sorted(apps, key=lambda a: a["name"].lower()) def get_running_containers(app: dict) -> list[str]: """Retourne les noms des conteneurs en cours d'exécution pour une app.""" try: result = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "ps", "--services", "--filter", "status=running"], capture_output=True, text=True, cwd=app["path"], ) services = [s.strip() for s in result.stdout.splitlines() if s.strip()] return services except FileNotFoundError: return [] def get_container_id(app: dict, service: str) -> str | None: """Retourne l'ID du conteneur Docker pour un service donné.""" try: result = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "ps", "-q", service], capture_output=True, text=True, cwd=app["path"], ) container_id = result.stdout.strip() return container_id if container_id else None except FileNotFoundError: return None def print_separator(): print("-" * 50) def print_header(): print("=" * 50) print(" Gestionnaire Docker VPS") print("=" * 50) def select_from_list(items: list[str], prompt: str) -> int | None: """Affiche une liste numérotée et retourne l'index choisi, ou None si annulé.""" for i, item in enumerate(items, 1): print(f" [{i}] {item}") print(" [0] Retour") print() while True: try: choice = input(f"{prompt}: ").strip() if choice == "0": return None idx = int(choice) - 1 if 0 <= idx < len(items): return idx print(f" Choix invalide (1-{len(items)} ou 0 pour retour).") except (ValueError, EOFError): print(" Entrée invalide.") def action_bash(app: dict, service: str): """Ouvre un shell bash interactif dans le conteneur.""" container_id = get_container_id(app, service) if not container_id: print(f" Conteneur '{service}' introuvable ou arrêté.") input(" Appuyez sur Entrée pour continuer...") return print(f"\n Connexion bash → {app['name']} / {service}") print(" (tapez 'exit' pour revenir au menu)\n") subprocess.run(["docker", "exec", "-it", container_id, "bash"]) def action_logs(app: dict, service: str): """Affiche les logs du conteneur (100 dernières lignes, avec suivi).""" container_id = get_container_id(app, service) if not container_id: print(f" Conteneur '{service}' introuvable ou arrêté.") input(" Appuyez sur Entrée pour continuer...") return print(f"\n Logs → {app['name']} / {service}") print(" (Ctrl+C pour arrêter le suivi)\n") try: subprocess.run(["docker", "logs", "--tail", "100", "-f", container_id]) except KeyboardInterrupt: print("\n Suivi des logs interrompu.") def action_update(app: dict): """Met à jour l'application avec docker compose pull puis up -d.""" print(f"\n Mise à jour de '{app['name']}'...") print_separator() print(" → docker compose pull") result_pull = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "pull"], cwd=app["path"], ) if result_pull.returncode != 0: print(" Erreur lors du pull.") input(" Appuyez sur Entrée pour continuer...") return print("\n → docker compose up -d") result_up = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "up", "-d"], cwd=app["path"], ) if result_up.returncode == 0: print("\n Mise à jour terminée avec succès.") else: print("\n Erreur lors du redémarrage des conteneurs.") input("\n Appuyez sur Entrée pour continuer...") def app_menu(app: dict): """Menu des actions disponibles pour une application.""" while True: os.system("clear") print_header() print(f" Application : {app['name']}") print(f" Répertoire : {app['path']}") print_separator() running = get_running_containers(app) if running: print(f" Services actifs : {', '.join(running)}") else: print(" Aucun service actif détecté.") print() print(" [1] Ouvrir un shell bash dans un conteneur") print(" [2] Voir les logs d'un conteneur") print(" [3] Mettre à jour (docker compose pull + up -d)") print(" [0] Retour") print() choice = input(" Votre choix : ").strip() if choice == "0": break elif choice in ("1", "2"): os.system("clear") print_header() label = "shell bash" if choice == "1" else "logs" print(f" Choisir le service ({label}) :") print_separator() all_services = get_all_services(app) if not all_services: print(" Aucun service trouvé dans le compose.") input(" Appuyez sur Entrée pour continuer...") continue idx = select_from_list(all_services, " Service") if idx is None: continue service = all_services[idx] if choice == "1": action_bash(app, service) else: action_logs(app, service) elif choice == "3": action_update(app) else: print(" Choix invalide.") def get_all_services(app: dict) -> list[str]: """Retourne tous les services définis dans le docker-compose (actifs ou non).""" try: result = subprocess.run( ["docker", "compose", "-f", app["compose_file"], "config", "--services"], capture_output=True, text=True, cwd=app["path"], ) services = [s.strip() for s in result.stdout.splitlines() if s.strip()] return services except FileNotFoundError: return [] def main(): if os.geteuid() != 0: print("Attention : ce script nécessite des droits root pour interagir avec Docker.") print("Relancez avec : sudo python3 manage_vps.py\n") while True: os.system("clear") print_header() apps = find_docker_apps() if not apps: print(f" Aucune application Docker trouvée sous {APPS_BASE_DIR}/") print(" (recherche de docker-compose.yml dans chaque sous-dossier)") sys.exit(0) print(f" {len(apps)} application(s) détectée(s) :\n") app_names = [a["name"] for a in apps] idx = select_from_list(app_names, " Choisir une application") if idx is None: print("\n Au revoir !") sys.exit(0) app_menu(apps[idx]) if __name__ == "__main__": main()