#!/myapps/venvs/portainer/bin/python3 import os import sys import requests import json import argparse from tabulate import tabulate from port import Portainer import logging VERSION = "0.0.1" 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): 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 get_portainer_token(base_url, username=None, password=None, timeout=10): """ Authenticate to Portainer and return a JWT token. Reads PORTAINER_USER / PORTAINER_PASS from environment if username/password are not provided. """ username = username or os.getenv("PORTAINER_USER") password = password or os.getenv("PORTAINER_PASS") if not username or not password: raise ValueError("Username and password must be provided (or set PORTAINER_USER / PORTAINER_PASS).") url = f"{base_url.rstrip('/')}/api/auth" resp = requests.post(url, json={"Username": username, "Password": password}, timeout=timeout) resp.raise_for_status() data = resp.json() token = data.get("jwt") or data.get("JWT") or data.get("token") if not token: raise ValueError(f"No token found in response: {data}") return token def prompt_missing_args(args, defaults, fields): """ fields = [("arg_name", "Prompt text")] """ for field, text in fields: value = getattr(args, field) default = defaults.get(field) if value is None: if default is not None: prompt = f"{text} (default={default}) : " value = input(prompt) or default else: value = input(f"{text}: ") setattr(args, field, value) 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] token = portainer_api_key # Example: list endpoints por = Portainer(base, token) 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") autostart = True if args.autostart else False por.update_stack(args.endpoint_id, args.stack, 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() data = [] for i in eps["by_id"]: data.append([i, eps["by_id"][i]]) headers = ["EndpointId", "Name"] print(tabulate(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 == c or args.stack == "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 == c or args.stack == "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 == c or args.stack == "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(base, token, endpoint_id=args.endpoint_id) else: por.refresh_status(base, args.stack_id, token)