portainer/portainer.py

import os
import requests
import json
import uuid
import argparse
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("--user", "-u", default=os.getenv("PORTAINER_USER"),
                    help="Portainer username (ENV: PORTAINER_USER)")
parser.add_argument("--password", "-p", default=os.getenv("PORTAINER_PASS"),
                    help="Portainer password (ENV: PORTAINER_PASS)")
parser.add_argument("--endpoint-id", "-e", type=str, help="Endpoint ID to limit stack operations")
parser.add_argument("--list-endpoints", action="store_true", help="List endpoints")
parser.add_argument("--list-stacks", "-l", action="store_true", help="List stacks")
parser.add_argument("--delete-stack", "-d", action="store_true", help="Delete stack")
parser.add_argument("--get-stack", metavar="NAME_OR_ID", help="Get stack by name or numeric id")
parser.add_argument("--create-stack","-c", action='store_true')
parser.add_argument("--gpu","-g", action='store_true')
parser.add_argument("--create-stacks","-C", action='store_true')
parser.add_argument("--stack-id", "-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")
args = parser.parse_args()


nas_stacks = ["pihole","authentik","bitwarden","bookstack","dockermon","gitea","gitlab","grafana","home-assistant","homepage","immich","jupiter","kestra","mailu3","mealie","mediacenter"]


portainer_api_key = "ptr_QoHE/e6kfqIirE3fhzYMRFRxK7eL42wtCxo5P18i1zQ="


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 api_get(base_url, path, token, timeout=10):
    """Example authenticated GET request to Portainer API."""
    url = f"{base_url.rstrip('/')}{path}"
    headers = {"Authorization": f"Bearer {token}"}
    headers = {"X-API-Key": f"{token}"}
    resp = requests.get(url, headers=headers, timeout=timeout)
    resp.raise_for_status()
    return resp.json()

def api_post(base_url, path, token, data, timeout=120):
    """Example authenticated GET request to Portainer API."""
    url = f"{base_url.rstrip('/')}{path}"
    headers = {"Authorization": f"Bearer {token}"}
    headers = {"X-API-Key": f"{token}"}
    resp = requests.post(url, headers=headers, json=data, timeout=timeout)
    resp.raise_for_status()
    return resp.json()

def api_delete(base_url, path, token, timeout=20):
    """Example authenticated DELETE request to Portainer API."""
    url = f"{base_url.rstrip('/')}{path}"
    headers = {"Authorization": f"Bearer {token}"}
    headers = {"X-API-Key": f"{token}"}
    resp = requests.delete(url, headers=headers, timeout=timeout)
    resp.raise_for_status()
    print(resp.status_code)
    return resp.status_code

def get_stacks(base_url, token, endpoint_id=None, timeout=10):
    """
    Return a list of stacks. If endpoint_id is provided, it will be added as a query param.
    """
    path = "/api/stacks"
    if endpoint_id is not None:
        path += f"?endpointId={endpoint_id}"
    stacks = api_get(base_url, path, token, timeout=timeout)
    if stacks is None:
        return []
    return stacks

def get_stack(base_url, identifier, token, endpoint_id=None, timeout=10):
    """
    Retrieve a single stack by numeric Id or by Name.
    Identifier may be an int (Id) or a string (Name). Raises ValueError if not found.
    """
    stacks = get_stacks(base_url, token, endpoint_id=endpoint_id, timeout=timeout)
    # Normalize identifier
    ident_id = None
    try:
        ident_id = int(identifier)
    except (TypeError, ValueError):
        pass

    for s in stacks:
        # Many Portainer responses use 'Id' and 'Name' keys
        if ident_id is not None and s.get("Id") == ident_id:
            return s
        if str(s.get("Name")) == str(identifier):
            return s

    raise ValueError(f"Stack not found: {identifier}")

def create_stack(base_url, token, endpoint_id=None, data={}, timeout=120):
    """
    Return a list of stacks. If endpoint_id is provided, it will be added as a query param.
    """
    path = "/api/stacks/create/standalone/repository"
    if endpoint_id is not None:
        path += f"?endpointId={endpoint_id}"
    try:        
        stacks = api_post(base_url, path, token, data, timeout=timeout)
    except Exception as e:
        print(f"Error creating stack: {e}")
        if "Conflict for url" in str(e):
            print("Stack with this name may already exist.")
        return []
    if stacks is None:
        return []
    return stacks

def delete_stack(base_url, token, endpoint_id=None, stack=None, timeout=30):
    """
    Return a list of stacks. If endpoint_id is provided, it will be added as a query param.
    """
    
    if stack == "all":
        stacks = get_stacks(base_url, token, endpoint_id=endpoint_id, timeout=timeout)
        for s in stacks:
            #print(s['EndpointId'], endpoint_id)
            if int(s['EndpointId']) != int(endpoint_id):
                continue
            print("Deleting stack:", s['Name'])
            path = f"/api/stacks/{s['Id']}"
            if endpoint_id is not None:
                path += f"?endpointId={endpoint_id}&removeVolumes=true"
            #print(path)
            out = api_delete(base_url, path, token, timeout=timeout)
        return "Done"
    else:
        path = f"/api/stacks/{stack}"

        try:
            #print(path)
            stacks = api_delete(base_url, path, token, timeout=timeout)
        except Exception as e:
            #print(f"Error creating stack: {e}")
            if "Conflict for url" in str(e):
                print("Stack with this name may already exist.")
            else:
                print(f"Error deleting stack: {e}")
            print(stacks)
            return []
        if stacks is None:
            return []
    
    return stacks

def print_stacks(base, token):
    stacks = get_stacks(base, token)
    for stack in stacks:    
        try:
            print(f"Stack ID: {stack['Id']}, Name: {stack['Name']}, EndpointName: {eps[stack['EndpointId']]}")
        except KeyError as e:
            print(f"Stack ID: {stack['Id']}, Name: {stack['Name']}, EndpointName: ?")

if __name__ == "__main__":
    # Example usage: set PORTAINER_USER and PORTAINER_PASS in env, or pass literals below.
    base = os.getenv("PORTAINER_URL", "https://portainer.sectorq.eu")
    #token = get_portainer_token(base,"admin","l4c1j4yd33Du5lo")  # or get_portainer_token(base, "admin", "secret")
    token = portainer_api_key
    # Example: list endpoints
    endpoints = api_get(base, "/api/endpoints", token)
    #print(endpoints)
    eps = {}
    install_endpoint_id = None
    for ep in endpoints:
        eps[ep['Id']] = ep['Name']
        print(f"Endpoint ID: {ep['Id']}, Name: {ep['Name']}")
        if args.endpoint_id == str(ep['Name']):
            install_endpoint_id = ep['Id']
    if install_endpoint_id == None:
        install_endpoint_id = args.endpoint_id
        
    if args.list_stacks:
        print_stacks(base, token)


    #print(json.dumps(req, indent=2))
    if args.create_stack:
        stck = get_stack(base, args.stack_id, token)
        print(f"Found stack: ID {stck['Id']}, Name: {stck['Name']}")
        print(json.dumps(stck, indent=2))
            
        for e in stck["Env"]:
            print(f"Env: {e['name']} = {e['value']}")
            HWS = ["HW_MODE","HW_MODE1","HW_MODE2"]
            if e['name'] in HWS:
                print("Found HW_MODE env var.")
                if args.gpu:
                    e['value'] = "hw"
                else:
                    e['value'] = "cpu"
            if e['name'] == "LOGGING":
                    print("Found LOGGING env var.")
                    if args.gpu:
                        e['value'] = "journald"
                    else:
                        e['value'] = "syslog"    
        #print(json.dumps(stck, indent=2))
        uid = str(uuid.uuid4())
        try:
            stck["AutoUpdate"]["Webhook"] = uid
        except:
            stck["AutoUpdate"] = None
        try:
            req = {
                "Name": stck["Name"],
                "Env": stck["Env"],
                "AdditionalFiles": stck["AdditionalFiles"],
                "AutoUpdate": stck["AutoUpdate"],
                "repositoryURL": stck["GitConfig"]["URL"],
                "ReferenceName": "refs/heads/main",
                "composeFile": f"{stck['Name']}/docker-compose.yml",
                "ConfigFilePath": f"{stck['Name']}/docker-compose.yml",
                "repositoryAuthentication": True,
                "repositoryUsername": "jaydee",
                "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS",
                "AuthorizationType": 0,
                "TLSSkipVerify": False,
                "supportRelativePath": True,
                "repositoryAuthentication": True,
                "fromAppTemplate": False,
                "registries": [6,3],
                "FromAppTemplate": False,
                "Namespace": "",
                "CreatedByUserId": "",
                "Webhook": "",
                "filesystemPath": "/share/docker_data/portainer/portainer-data/stacks",
                "RegistryID": 4
            }
        except:
            req = {
            "Name": stck["Name"],
            "Env": stck["Env"],
            "AdditionalFiles": stck["AdditionalFiles"],
            "AutoUpdate": None,
            "repositoryURL": "https://gitlab.sectorq.eu/home/docker-compose.git",
            "ReferenceName": "refs/heads/main",
            "composeFile": f"{stck['Name']}/docker-compose.yml",
            "ConfigFilePath": f"{stck['Name']}/docker-compose.yml",
            "repositoryAuthentication": True,
            "repositoryUsername": "jaydee",
            "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS",
            "AuthorizationType": 0,
            "TLSSkipVerify": False,
            "supportRelativePath": True,
            "repositoryAuthentication": True,
            "fromAppTemplate": False,
            "registries": [6,3],
            "FromAppTemplate": False,
            "Namespace": "",
            "CreatedByUserId": "",
            "Webhook": "",
            "filesystemPath": "/share/docker_data/portainer/portainer-data/stacks",
            "RegistryID": 6
        }
        print(json.dumps(req, indent=2))
        create_stack(base, token, install_endpoint_id, data=req)
    else:
        print("Not creating stack, --create-stack not provided.")        
    if args.create_stacks:
        for ns in nas_stacks:
            stck = get_stack(base, ns, token)
            print(f"Found stack: ID {stck['Id']}, Name: {stck['Name']}")
            #print(json.dumps(stck, indent=2))
            for e in stck["Env"]:
                #print(f"Env: {e['name']} = {e['value']}")
                HWS = ["HW_MODE","HW_MODE1","HW_MODE2"]
                if e['name'] in HWS:
                    print("Found HW_MODE env var.")
                    if args.gpu:
                        e['value'] = "hw"
                    else:
                        e['value'] = "cpu"
                if e['name'] == "LOGGING":
                    print("Found LOGGING env var.")
                    if args.gpu:
                        e['value'] = "journald"
                    else:
                        e['value'] = "syslog"                        
            #print(stck["Env"])
            uid = str(uuid.uuid4())
            try:
                stck["AutoUpdate"]["Webhook"] = uid
            except:
                stck["AutoUpdate"] = {}
            try:
                req = {
                    "Name": stck["Name"],
                    "Env": stck["Env"],
                    "AdditionalFiles": stck["AdditionalFiles"],
                    "AutoUpdate": stck["AutoUpdate"],
                    "repositoryURL": stck["GitConfig"]["URL"],
                    "ReferenceName": "refs/heads/main",
                    "composeFile": f"{stck['Name']}/docker-compose.yml",
                    "ConfigFilePath": f"{stck['Name']}/docker-compose.yml",
                    "repositoryAuthentication": True,
                    "repositoryUsername": "jaydee",
                    "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS",
                    "AuthorizationType": 0,
                    "TLSSkipVerify": False,
                    "supportRelativePath": True,
                    "repositoryAuthentication": True,
                    "fromAppTemplate": False,
                    "registries": [6,3],
                    "FromAppTemplate": False,
                    "Namespace": "",
                    "CreatedByUserId": "",
                    "Webhook": "",
                    "filesystemPath": "/share/docker_data/portainer/portainer-data/stacks",
                    "RegistryID": 4
                }
            except:
                req = {
                "Name": stck["Name"],
                "Env": stck["Env"],
                "AdditionalFiles": stck["AdditionalFiles"],
                "AutoUpdate": None,
                "repositoryURL": "https://gitlab.sectorq.eu/home/docker-compose.git",
                "ReferenceName": "refs/heads/main",
                "composeFile": f"{stck['Name']}/docker-compose.yml",
                "ConfigFilePath": f"{stck['Name']}/docker-compose.yml",
                "repositoryAuthentication": True,
                "repositoryUsername": "jaydee",
                "repositoryPassword": "glpat-uj-n-eEfTY398PE4vKSS",
                "AuthorizationType": 0,
                "TLSSkipVerify": False,
                "supportRelativePath": True,
                "repositoryAuthentication": True,
                "fromAppTemplate": False,
                "registries": [6,3],
                "FromAppTemplate": False,
                "Namespace": "",
                "CreatedByUserId": "",
                "Webhook": "",
                "filesystemPath": "/share/docker_data/portainer/portainer-data/stacks",
                "RegistryID": 6
            }
            print(f"Creating stack: {ns}")
            create_stack(base, token, install_endpoint_id, data=req)
    else:
        print("Not creating stack, --create-stack not provided.")

    if args.delete_stack:
        delete_stack(base, token, install_endpoint_id, args.stack_id)
    else:
        print("Not deleting stack, --create-stack not provided.")