portainer/portainer.py

"""
Portainer API Client module.

This module provides a wrapper for interacting with the Portainer API
to manage endpoints, stacks, and containers.
"""

# !/myapps/venvs/portainer/bin/python3
import os
import logging
import sys
import json
import argparse
from tabulate import tabulate
from port import Portainer


VERSION = "0.0.4"

defaults = {
    "endpoint_id": "vm01",
    "stack": "my_stack",
    "deploy_mode": "git",
    "autostart": True,
    "stack_mode": "swarm",
    "site": "portainer",
}


parser = argparse.ArgumentParser(
    description="Portainer helper - use env vars or pass credentials."
)
parser.add_argument(
    "--base",
    "-b",
    default=os.getenv("PORTAINER_URL", "https://portainer.example.com"),
    help="Base URL for Portainer (ENV: PORTAINER_URL)",
)
parser.add_argument("--site", "-t", type=str, default=None, help="Site")
parser.add_argument(
    "--endpoint-id",
    "-e",
    type=str,
    default=None,
    help="Endpoint ID to limit stack operations",
)
parser.add_argument(
    "--refresh-environment", "-R", action="store_true", help="List endpoints"
)
parser.add_argument(
    "--list-endpoints", "-E", action="store_true", help="List endpoints"
)
parser.add_argument("--list-stacks", "-l", action="store_true", help="List stacks")
parser.add_argument("--print-all-data", "-A", action="store_true", help="List stacks")
parser.add_argument(
    "--list-containers", "-c", action="store_true", help="List containers"
)
parser.add_argument("--update-stack", "-U", action="store_true", help="Update stacks")
parser.add_argument(
    "--stop-containers", "-O", action="store_true", help="Stop containers"
)
parser.add_argument(
    "--start-containers", "-X", action="store_true", help="Start containers"
)
parser.add_argument("--update-status", "-S", action="store_true", help="Update status")
parser.add_argument(
    "--get-stack", metavar="NAME_OR_ID", help="Get stack by name or numeric id"
)
parser.add_argument("--action", "-a", type=str, default=None, help="Action to perform")
parser.add_argument(
    "--autostart", "-Z", action="store_true", help="Auto-start created stacks"
)
parser.add_argument("--start-stack", "-x", action="store_true")
parser.add_argument("--stop-stack", "-o", action="store_true")
parser.add_argument("--secrets", "-q", action="store_true")
parser.add_argument("--debug", "-D", action="store_true")
parser.add_argument("--create-stack", "-n", action="store_true")
parser.add_argument("--create-stack_new2", "-N", action="store_true")
parser.add_argument("--gpu", "-g", action="store_true")
parser.add_argument("--create-stacks", "-C", action="store_true")
parser.add_argument("--refresh-status", "-r", action="store_true")

parser.add_argument("--stack", "-s", type=str, help="Stack ID for operations")
parser.add_argument(
    "--token-only", action="store_true", help="Print auth token and exit"
)
parser.add_argument("--timeout", type=int, default=10, help="Request timeout seconds")
parser.add_argument("--deploy-mode", "-m", type=str, default="git", help="Deploy mode")
parser.add_argument("--stack-mode", "-w", default=None, help="Stack mode")
args = parser.parse_args()
print("Running version:", VERSION)
print("Environment:", args.site)

_LOG_LEVEL = "INFO"
LOG_FILE = "/tmp/portainer.log"
if _LOG_LEVEL == "DEBUG":
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.DEBUG,
        format="%(asctime)s : %(levelname)s : %(message)s",
        datefmt="%m/%d/%Y %I:%M:%S %p",
    )
    logging.debug("using debug logging")
elif _LOG_LEVEL == "ERROR":
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.ERROR,
        format="%(asctime)s : %(levelname)s : %(message)s",
        datefmt="%m/%d/%Y %I:%M:%S %p",
    )
    logging.info("using error logging")
elif _LOG_LEVEL == "SCAN":
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.DEBUG,
        format="%(asctime)s : %(levelname)s : %(message)s",
        datefmt="%m/%d/%Y %I:%M:%S %p",
    )
    logging.info("using scan logging")
else:
    logging.basicConfig(
        filename=LOG_FILE,
        level=logging.INFO,
        format="%(asctime)s : %(levelname)s : %(message)s",
        datefmt="%m/%d/%Y %I:%M:%S %p",
    )
logging.info("script started")

logger = logging.getLogger(__name__)

if args.site == "portainer":
    base = os.getenv("PORTAINER_URL", "https://portainer.sectorq.eu/api")
    PORTAINER_API_KEY = "ptr_GCNUoFcTOaXm7k8ZxPdQGmrFIamxZPTydbserYofMHc="
else:
    base = os.getenv("PORTAINER_URL", "https://port.sectorq.eu/api")
    PORTAINER_API_KEY = "ptr_/5RkMCT/j3BTaL32vMSDtXFi76yOXRKVFOrUtzMsl5Y="


def wl(msg):
    """Write log message if debug is enabled."""
    if args.debug:
        print(msg)


def is_number(s):
    """Check if the input string is a number."""
    try:
        float(s)
        return True
    except ValueError:
        return False


def prompt_missing_args(args_in, defaults_in, fields):
    """
    fields = [("arg_name", "Prompt text")]
    """
    for field, text in fields:
        value_in = getattr(args_in, field)
        default = defaults_in.get(field)

        if value_in is None:
            if default is not None:
                prompt = f"{text} (default={default}) : "
                value_in = input(prompt) or default
            else:
                value_in = input(f"{text}: ")

            setattr(args, field, value_in)

    return args


if __name__ == "__main__":
    # Example usage: set PORTAINER_USER and PORTAINER_PASS in env, or pass literals below.
    # token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo")  # or get_portainer_token(base, "admin", "secret")
    if args.action is None:
        actions = [
            "delete_stack",
            "create_stack",
            "stop_stack",
            "start_stack",
            "list_stacks",
            "update_stack",
            "secrets",
            "print_all_data",
            "list_endpoints",
            "list_containers",
            "stop_containers",
            "start_containers",
            "refresh_environment",
            "refresh_status",
            "update_status",
        ]
        print("Possible actions: ")
        i = 1
        for a in actions:
            print(f" > {i}. {a}")
            i += 1
        ans = input("\nSelect action to perform: ")
        args.action = actions[int(ans) - 1]

    # Example: list endpoints
    por = Portainer(base, PORTAINER_API_KEY, timeout=args.timeout)

    if args.action == "secrets":
        if args.endpoint_id is None:
            args.endpoint_id = input("Endpoint ID is required for creating secrets : ")

        secrets = {
            "gitea_runner_registration_token": "8nmKqJhkvYwltmNfF2o9vs0tzo70ufHSQpVg6ymb",
            "influxdb2-admin-token": "l4c1j4yd33Du5lo",
            "ha_influxdb2_admin_token": "l4c1j4yd33Du5lo",
            "wordpress_db_password": "wordpress",
            "wordpress_root_db_password": "wordpress",
        }
        for key, value in secrets.items():
            res = por.create_secret(key, value, args.endpoint_id, args.timeout)
            print(res)
        sys.exit()

    if args.action == "delete_stack":
        args = prompt_missing_args(
            args,
            defaults,
            [
                ("site", "Site"),
                ("endpoint_id", "Endpoint ID"),
                ("stack", "Stack name or ID"),
            ],
        )
        por.delete_stack(
            args.endpoint_id,
            args.stack,
        )
        sys.exit()

    if args.action == "create_stack":
        args = prompt_missing_args(
            args,
            defaults,
            [
                ("site", "Site"),
                ("endpoint_id", "Endpoint ID"),
                ("stack", "Stack name or ID"),
                ("stack_mode", "Stack mode (swarm or compose)"),
                ("deploy_mode", "Deploy mode (git or upload)"),
            ],
        )
        por.create_stack(
            args.endpoint_id,
            args.stack,
            args.deploy_mode,
            args.autostart,
            args.stack_mode,
        )
        sys.exit()

    if args.action == "stop_stack":
        if args.endpoint_id is None:
            args.endpoint_id = input("Endpoint ID is required for stopping stacks : ")
        if args.stack is None:
            args.stack = input("Stack name or ID is required for stopping stacks : ")
        por.stop_stack(args.stack, args.endpoint_id)
        sys.exit()

    if args.action == "start_stack":
        if args.endpoint_id is None:
            args.endpoint_id = input("Endpoint ID is required for starting stacks : ")
        if args.stack is None:
            args.stack = input("Stack name or ID is required for starting stacks : ")
        por.start_stack(args.stack, args.endpoint_id)
        sys.exit()

    if args.action == "list_stacks":
        por.print_stacks(args.endpoint_id)
        print(json.dumps(por.all_data, indent=2))
        sys.exit()

    if args.action == "list_containers":
        print("Getting containers")
        por.get_containers(args.endpoint_id, args.stack)
        sys.exit()

    if args.action == "update_stack":
        print("Updating stacks")
        por.update_stack(args.endpoint_id, args.stack, args.autostart)
        sys.exit()
    if args.action == "print_all_data":
        print(json.dumps(por.all_data, indent=2))
        sys.exit()
    if args.action == "update_status":
        por.update_status(args.endpoint_id, args.stack)
        sys.exit()

    if args.action == "list_endpoints":
        eps = por.get_endpoints()
        export_data = []
        for i in eps["by_id"]:
            export_data.append([i, eps["by_id"][i]])
        headers = ["EndpointId", "Name"]
        print(tabulate(export_data, headers=headers, tablefmt="github"))

        sys.exit()

    if args.action == "stop_containers":
        if por.all_data["endpoints_status"][args.endpoint_id] != 1:
            print(f"Endpoint {por.get_endpoint_name(args.endpoint_id)} is offline")
            sys.exit()
        print(f"Stopping containers on {por.get_endpoint_name(args.endpoint_id)}")
        cont = []
        for c in por.all_data["containers"][args.endpoint_id]:
            if args.stack in (c, "all"):
                cont += por.all_data["containers"][args.endpoint_id][c]
        por.stop_containers(args.endpoint_id, cont)
        sys.exit()

    if args.action == "start_containers":
        print("Starting containers")
        cont = []
        # input(json.dumps(por.all_data, indent=2))
        for c in por.all_data["containers"][args.endpoint_id]:
            if args.stack in (c, "all"):
                cont += por.all_data["containers"][args.endpoint_id][c]
        por.start_containers(args.endpoint_id, cont)
        sys.exit()
    if args.action == "start_containers":
        print("Starting containers")
        cont = []
        # input(json.dumps(por.all_data,indent=2))
        for c in por.all_data["containers"][args.endpoint_id]:
            if args.stack in (c, "all"):
                cont += por.all_data["containers"][args.endpoint_id][c]
        por.start_containers(args.endpoint_id, cont)
        sys.exit()
    if args.action == "refresh_environment":
        cont = por.refresh()
        sys.exit()

    if args.action == "refresh_status":
        if args.stack == "all":
            print("Stopping all stacks...")
            stcks = por.get_stacks(endpoint_id=args.endpoint_id)
        else:
            por.refresh_status(args.stack_id)