#!/myapps/venvs/portainer/bin/python3 import os import sys import requests import json import uuid import argparse import shutil import time from tabulate import tabulate from git import Repo # pip install gitpython from port import Portainer import logging 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="portainer", help="Site") parser.add_argument("--endpoint-id", "-e", type=str, default="all", 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 stacls") parser.add_argument("--stop-containers", "-O", action="store_true", help="Stop containers") parser.add_argument("--start-containers", "-X", action="store_true", help="Stop containers") parser.add_argument("--delete-stack", "-d", action="store_true", help="Delete stack") 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("--autostart", "-a", 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("--swarm","-w", action="store_true", help="Swarm mode") args = parser.parse_args() _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 loging') 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 loging') 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 error loging') 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 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") token = portainer_api_key # Example: list endpoints por = Portainer(base, token) if args.secrets: secrets = { "gitea_runner_registration_token": "8nmKqJhkvYwltmNfF2o9vs0tzo70ufHSQpVg6ymb", } for key, value in secrets.items(): por.create_secret(key, value, args.endpoint_id, args.timeout) sys.exit() if args.delete_stack: por.delete_stack(args.endpoint_id, args.stack,) sys.exit() if args.create_stack: por.create_stack(args.endpoint_id,args.stack, args.deploy_mode, args.autostart, args.swarm) sys.exit() if args.stop_stack: por.stop_stack(args.stack,args.endpoint_id) sys.exit() if args.start_stack: por.start_stack(args.stack,args.endpoint_id) sys.exit() if args.list_stacks: por.print_stacks(args.endpoint_id) print(json.dumps(por.all_data,indent=2)) sys.exit() if args.list_containers: print("Getting containers") por.get_containers(args.endpoint_id,args.stack) sys.exit() if args.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.print_all_data: print(json.dumps(por.all_data,indent=2)) sys.exit() if args.update_status: por.update_status(args.endpoint_id,args.stack) sys.exit() if args.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.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.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.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.refresh_environment: cont = por.refresh() sys.exit() if args.refresh_status: if args.stack== "all": print("Stopping all stacks...") stcks = por.get_stacks(base, token, endpoint_id=args.endpoint_id) # stcks = get_stack(base, sta, token, endpoint_id=install_endpoint_id) else: por.refresh_status(base, args.stack_id, token)