shell/docker-manager/docker-manager.py

#!/usr/bin/env python3
"""
Docker Manager Application

This script manages Docker containers defined in subdirectories of `~/docker/`.
Each subdirectory is treated as a project and must contain a `docker-compose.yml`, `docker-compose.yaml`, `compose.yml`, or `compose.yaml`.

Usage Examples:
    1. List all running containers:
       $ dm list

    2. Update a specific project (pulls latest images and recreates containers):
       $ dm update media-server

    3. Stop all projects (with confirmation prompt):
       $ dm stop

    4. Restart all projects without confirmation:
       $ dm restart --all

    5. Stop a specific project:
       $ dm stop web-app

Purpose:
    - Simplifies management of multiple Docker Compose projects.
    - Provides a unified interface for common operations (list, stop, update, restart).
    - Eliminates the need to navigate to specific directories to run docker commands.

To make the command available as dm from any directory, run:
  chmod +x ~/shell/docker-manager/docker-manager.py
  mkdir -p ~/.local/bin
  ln -s ~/shell/docker-manager/docker-manager.py ~/.local/bin/dm
"""

import os
import argparse
import subprocess
import json
from pathlib import Path

# Configuration
DOCKER_ROOT = Path(os.path.expanduser("~/docker"))


def get_projects():
    """Scan DOCKER_ROOT for subdirectories with compose files."""
    projects = {}
    if not DOCKER_ROOT.exists():
        print(f"Error: Directory {DOCKER_ROOT} does not exist.")
        return projects

    for item in DOCKER_ROOT.iterdir():
        if item.is_dir():
            if (item / "docker-compose.yml").exists() or \
               (item / "docker-compose.yaml").exists() or \
               (item / "compose.yml").exists() or \
               (item / "compose.yaml").exists():
                projects[item.name] = item
    return projects


def run_command(cmd, cwd, capture_output=False):
    """Run a shell command in a specific directory."""
    try:
        result = subprocess.run(
            cmd,
            cwd=cwd,
            check=True,
            text=True,
            capture_output=capture_output
        )
        return result
    except subprocess.CalledProcessError as e:
        if capture_output:
            return e
        print(f"Error running command: {' '.join(cmd)}")
        print(f"cwd: {cwd}")
        return None


def list_containers(projects):
    """List running containers for all projects."""
    print(f"{'Project':<20} | {'Container Name':<40} | {'State':<10}")
    print("-" * 75)

    found_any = False
    for name, path in sorted(projects.items()):
        # Get running container names
        cmd = ["docker", "compose", "ps", "--format",
               "{{.Names}}", "--filter", "status=running"]
        res = run_command(cmd, path, capture_output=True)

        if res and res.returncode == 0:
            containers = res.stdout.strip().split('\n')
            containers = [c for c in containers if c]  # filter empty
            if containers:
                found_any = True
                for container in containers:
                    print(f"{name:<20} | {container:<40} | Running")

    if not found_any:
        print("No running containers found in managed projects.")


def is_project_running(path):
    """Check if any containers in the project are running."""
    cmd = ["docker", "compose", "ps", "--format",
           "json", "--filter", "status=running"]
    res = run_command(cmd, path, capture_output=True)
    if res and res.returncode == 0:
        # If output is not empty/just brackets, something is running
        output = res.stdout.strip()
        return output and output != "[]"
    return False


def describe_project(projects, target):
    """Show detailed information about containers in a project."""
    if target not in projects:
        print(f"Error: Project '{target}' not found.")
        return

    path = projects[target]
    print(f"Describing project: {target}")

    cmd = ["docker", "compose", "ps", "--format", "json"]
    res = run_command(cmd, path, capture_output=True)

    if res and res.returncode == 0:
        lines = res.stdout.strip().split('\n')
        found_any = False
        for line in lines:
            if not line:
                continue
            try:
                container = json.loads(line)
                found_any = True
                name = container.get('Name', 'N/A')
                print("-" * 60)
                print(f"Service:     {container.get('Service', 'N/A')}")
                print(f"Container:   {name}")
                print(f"Image:       {container.get('Image', 'N/A')}")
                print(f"State:       {container.get('State', 'N/A')}")
                print(f"Status:      {container.get('Status', 'N/A')}")

                ports = container.get('Ports', '')
                if ports:
                    print(f"Ports:       {ports}")

                # Get details via inspect for better parsing (Labels and Mounts)
                inspect_cmd = ["docker", "inspect", name, "--format", "{{json .}}"]
                inspect_res = run_command(inspect_cmd, path, capture_output=True)
                if inspect_res and inspect_res.returncode == 0:
                    try:
                        details = json.loads(inspect_res.stdout.strip())
                        
                        # Description from labels
                        labels = details.get("Config", {}).get("Labels", {})
                        description = labels.get("org.opencontainers.image.description")
                        if description:
                            print(f"Description: {description}")

                        # Volumes/Mounts
                        mounts = details.get("Mounts", [])
                        if mounts:
                            print("Volumes:")
                            print(f"  {'Name/Type':<20} | {'Source (Local Path)':<50} | {'Destination'}")
                            print(f"  {'-'*20} | {'-'*50} | {'-'*20}")
                            for mount in mounts:
                                m_type = mount.get("Type")
                                vol_name = mount.get("Name", "")
                                source = mount.get("Source", "")
                                dest = mount.get("Destination", "")
                                
                                # If it's a bind mount, use "Bind" as name, otherwise use volume name
                                display_name = vol_name if vol_name else f"[{m_type}]"
                                
                                # Truncate source if too long for cleaner display, or just let it wrap?
                                # Let's just print it.
                                print(f"  {display_name:<20} | {source:<50} | {dest}")

                    except json.JSONDecodeError:
                        pass

            except json.JSONDecodeError:
                pass
        
        if not found_any:
            print("No containers found for this project.")
    else:
        print("Failed to get container info.")


def list_project_volumes(projects, target):
    """List volumes used by containers in a project."""
    if target not in projects:
        print(f"Error: Project '{target}' not found.")
        return

    path = projects[target]
    print(f"Volumes for project: {target}")
    print("-" * 100)
    print(f"{'Service':<20} | {'Type':<10} | {'Name/Source':<40} | {'Destination'}")
    print("-" * 100)

    cmd = ["docker", "compose", "ps", "--format", "json"]
    res = run_command(cmd, path, capture_output=True)

    if res and res.returncode == 0:
        lines = res.stdout.strip().split('\n')
        found_any = False
        for line in lines:
            if not line:
                continue
            try:
                container = json.loads(line)
                name = container.get('Name', 'N/A')
                service = container.get('Service', 'N/A')

                # Get details via inspect
                inspect_cmd = ["docker", "inspect", name, "--format", "{{json .Mounts}}"]
                inspect_res = run_command(inspect_cmd, path, capture_output=True)
                
                if inspect_res and inspect_res.returncode == 0:
                    try:
                        mounts = json.loads(inspect_res.stdout.strip())
                        if mounts:
                            found_any = True
                            for mount in mounts:
                                m_type = mount.get("Type", "unknown")
                                source = mount.get("Source", "")
                                dest = mount.get("Destination", "")
                                name_or_source = mount.get("Name", "")
                                
                                if m_type == "bind":
                                    name_or_source = source
                                elif not name_or_source:
                                    name_or_source = source

                                print(f"{service:<20} | {m_type:<10} | {name_or_source:<40} | {dest}")
                    except json.JSONDecodeError:
                        pass

            except json.JSONDecodeError:
                pass
        
        if not found_any:
            print("No volumes found for this project.")
    else:
        print("Failed to get container info.")


def manage_project(projects, action, target, extra_args=None):
    """
    Execute the specified action (stop, update, restart, logs) on target project(s).

    Args:
        projects (dict): Dictionary of project names to their paths.
        action (str): The action to perform ('stop', 'update', 'restart', 'logs').
        target (str): The target project name or 'all'.
        extra_args (list): Additional arguments for the command (e.g. for logs).
    """

    targets = []
    # Determine which projects to target
    if target == "all":
        targets = sorted(projects.items())
    elif target in projects:
        targets = [(target, projects[target])]
    else:
        print(f"Error: Project '{target}' not found in {DOCKER_ROOT}")
        print("Available projects:", ", ".join(sorted(projects.keys())))
        return

    # Logs are special, usually run on one project interactively
    if action == "logs":
        if len(targets) > 1:
            print("Error: Logs can only be viewed for one project at a time.")
            return
        name, path = targets[0]
        print(f"Viewing logs for {name}...")
        cmd = ["docker", "compose", "logs"] + (extra_args or [])
        # For logs, we want to stream output directly to stdout, so no capture_output
        # and we might want to allow Ctrl+C to exit gracefully
        try:
            subprocess.run(cmd, cwd=path, check=False)
        except KeyboardInterrupt:
            print("\nLog view stopped.")
        return

    print(f"Performing '{action}' on {len(targets)} project(s)...")

    for name, path in targets:
        print(f"\n[{name}] -> {action}...")

        # Execute the requested action
        if action == "stop":
            # Stop containers without removing them
            run_command(["docker", "compose", "stop"], path)
        elif action == "restart":
            # Restart containers
            run_command(["docker", "compose", "restart"], path)
        elif action == "update":
            # Update process: Check running -> Stop -> Pull -> Start if was running
            was_running = is_project_running(path)

            if was_running:
                print(f"  Stopping running containers for {name}...")
                run_command(["docker", "compose", "stop"], path)

            print(f"  Pulling latest images for {name}...")
            pull_res = run_command(["docker", "compose", "pull"], path)

            if pull_res and pull_res.returncode == 0:
                if was_running:
                    print(f"  Restarting containers for {name}...")
                    run_command(["docker", "compose", "up", "-d"], path)
                else:
                    print(
                        f"  Project {name} was not running. Images updated, but not started.")
            else:
                print(f"  Failed to pull images for {name}. Skipping update.")


def main():
    """Main entry point for the application."""
    parser = argparse.ArgumentParser(
        description="Manage Docker containers in ~/docker/")
    subparsers = parser.add_subparsers(
        dest="command", help="Command to execute")

    # List command configuration
    subparsers.add_parser("list", help="List running containers")

    # Describe command configuration
    describe_parser = subparsers.add_parser("describe", help="Show details of a project's containers")
    describe_parser.add_argument("project", help="Project name to describe")

    # Volumes command configuration
    volumes_parser = subparsers.add_parser("volumes", help="List volumes used by a project")
    volumes_parser.add_argument("project", help="Project name to list volumes for")

    # Stop command configuration
    stop_parser = subparsers.add_parser("stop", help="Stop containers")
    stop_parser.add_argument("project", nargs="?", default=None,
                             help="Project name (optional). If omitted, asks for confirmation to stop ALL.")
    stop_parser.add_argument("--all", action="store_true",
                             help="Stop all projects without confirmation prompt if specified")

    # Update command configuration
    update_parser = subparsers.add_parser(
        "update", help="Pull and update containers")
    update_parser.add_argument("project", nargs="?", default=None,
                               help="Project name (optional). If omitted, asks for confirmation to update ALL.")
    update_parser.add_argument("--all", action="store_true",
                               help="Update all projects without confirmation prompt if specified")

    # Restart command configuration
    restart_parser = subparsers.add_parser(
        "restart", help="Restart containers")
    restart_parser.add_argument("project", nargs="?", default=None,
                                help="Project name (optional). If omitted, asks for confirmation to restart ALL.")
    restart_parser.add_argument("--all", action="store_true",
                                help="Restart all projects without confirmation prompt if specified")

    # Logs command configuration
    logs_parser = subparsers.add_parser("logs", help="View container logs")
    logs_parser.add_argument("project", help="Project name to view logs for")
    logs_parser.add_argument(
        "-f", "--follow", action="store_true", help="Follow log output")
    logs_parser.add_argument(
        "--tail", default="all", help="Number of lines to show from the end of the logs (default: all)")

    args = parser.parse_args()

    # Scan for projects
    projects = get_projects()
    if not projects:
        print(f"No docker projects found in {DOCKER_ROOT}")
        return

    # Dispatch commands
    if args.command == "list":
        list_containers(projects)
    elif args.command == "describe":
        describe_project(projects, args.project)
    elif args.command == "volumes":
        list_project_volumes(projects, args.project)
    elif args.command == "logs":
        # Prepare extra args for logs
        extra_args = []
        if args.follow:
            extra_args.append("-f")
        if args.tail:
            extra_args.extend(["--tail", args.tail])

        manage_project(projects, "logs", args.project, extra_args=extra_args)

    elif args.command in ["stop", "update", "restart"]:
        target = args.project

        # Handle "all" logic safely with user confirmation
        if target is None:
            if args.all:
                target = "all"
            else:
                # Interactive check if user wants to apply to all
                print(
                    f"No project specified. Do you want to {args.command} ALL projects? (y/N)")
                choice = input().lower()
                if choice == 'y':
                    target = "all"
                else:
                    print("Operation cancelled. Specify a project name or use --all.")
                    return

        manage_project(projects, args.command, target)
    else:
        # Show help if no command provided
        parser.print_help()


if __name__ == "__main__":
    main()