mirror of
https://gitlab.sectorq.eu/jaydee/portainer.git
synced 2025-12-14 10:44:52 +01:00
198 lines
8.5 KiB
Python
Executable File
198 lines
8.5 KiB
Python
Executable File
#!/home/jd/projects/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("--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.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)
|
|
|