portainer/portainer.py

#!/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
VERSION="0.0.1"

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=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 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("--action", "-a", type=str, default=None, help="Action to perform")


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")
print("Running version:", VERSION)
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")
    if args.action == None:
        actions = ["get-vms", "get-vms-all", "start", "stop", "delete", "rollback", "snapshot", "clone", "update_hosts"]
        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.secrets:
        if args.endpoint_id == 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.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)