With this script you can access Confluence via REST-API to create, modify or delete pages. You can create the needed "auth" file for authentication with the script create_authorization.py. - Mit diesem Skript kann man per REST-API auf Confluence zugreifen, um Seiten anzulegen, zu ändern oder zu löschen. Die benötigte "auth"-Datei zur Authentifiz…

confluence_util.py

"""With this script you can access Confluence via REST-API to create, modify or delete pages.
An auth file for the Confluence authorization is needed.
You can create this file with the script create_authorization.py."""
import datetime
from os.path import exists

import json
import requests

URL_BASE = 'https://<your-confluence-url>/rest/api/content'


def _get_authentication() -> str:
    """Read authentication string from file auth.
    if the auth file not exists, the script will exit with error code 1.

    :return: content from auth file as a string
    """

    if exists('./auth'):
        with open('./auth') as auth_file:
            auth_string = auth_file.read()
    else:
        print('Datei auth existiert nicht!')
        exit(1)

    return auth_string


def _get_request_header() -> dict:
    """Returns the header for usage in a HTTP-request

    :return: header in json-format
    """

    return {
        'authorization': f'{_get_authentication()}',
        'content-type': 'application/json;charset=iso-8859-1',
    }


def get_page_data(page_title: str) -> str:
    """Gets the page data of given page title.

    :param page_title: page title of the requested page
    :return: page data in json format
    """
    page_id = exists_page(page_title)

    request_headers = _get_request_header()
    page_url = URL_BASE + "/" + page_id + "?expand=body.storage,version"
    response = requests.get(url=page_url, headers=request_headers, verify=True)
    response.raise_for_status()
    result = response.json()

    return result


def exists_page(page_title: str) -> str:
    """Checks if page of given page title exists.

    :param page_title: page title of the requested page
    :return: page id, if page exists
            empty string, if page doesn't exist
    """
    request_headers = _get_request_header()
    title_url = URL_BASE + "?title=" + page_title
    response = requests.get(url=title_url, headers=request_headers, verify=True)
    response.raise_for_status()
    result = response.json()

    if result["results"] and result["results"][0]["title"] == page_title:
        return result["results"][0]["id"]
    else:
        return ""


def create_page(parent_page_id: str, page_title: str, page_html: str, space_key: str) -> None:
    """Creates a confluence page with the given parameters.

     :param parent_page_id: id of the existing parent page in Confluence
     :param page_title: title of the new page in Confluence
     :param page_html: content of the page in html format
     :param space_key: name of the confluence space where page should be created
     """

    request_headers = _get_request_header()
    data = {
        'type': 'page',
        'title': page_title,
        'ancestors': [{"type": "page", 'id': parent_page_id}],
        'space': {'key': space_key},
        'body': {
            'storage': {
                'value': page_html,
                'representation': 'storage',
            }
        }
    }

    response = requests.post(url=URL_BASE, data=json.dumps(data), headers=request_headers, verify=True)
    response.raise_for_status()


def add_content_to_page(page_id: str, additional_content: str, confluence_username: str,
                        confluence_display_name: str) -> None:
    """Append content to existing page given by page id.

    :param page_id: id of the requested page
    :param additional_content: content to be appended to the page in html format
    :param confluence_username: username of the confluence user who has the update permission
    :param confluence_display_name: display name for the username which is used in confluence
    """

    request_headers = _get_request_header()
    page_url_for_update = URL_BASE + "/" + page_id + "?expand=body.storage,version"

    response = requests.get(url=page_url_for_update, headers=request_headers, verify=True)
    response.raise_for_status()

    result = response.json()
    result["body"]["storage"]["value"] += additional_content
    result["version"]["by"]["username"] = confluence_username
    result["version"]["by"]["displayName"] = confluence_display_name
    result["version"]["when"] = datetime.datetime.now(datetime.timezone.utc).\
        astimezone().strftime('%Y-%m-%dT%H:%M:%S.%f%z')
    result["version"]["number"] += 1

    page_update_url = URL_BASE + "/" + page_id

    response = requests.put(url=page_update_url, data=json.dumps(result), headers=request_headers, verify=True)
    response.raise_for_status()


def _move_page_to_recycle_bin(page_id: str, request_headers: dict) -> None:
    """Calls the delete request and moves the confluence page with the given page id to the recycle bin

    :param page_id: id of the requested page
    :param request_headers: content of the header for the https request
    """
    
    page_id_url = URL_BASE + "/" + page_id
    response = requests.delete(url=page_id_url, headers=request_headers, verify=True)
    response.raise_for_status()


def delete_page(page_title: str, delete_sub_pages: bool) -> None:
    """Deletes given page and all sub pages if param delete_sub_page is true.

    :param page_title: page title of the requested page to be deleted
    :param delete_sub_pages: if true, all sub pages will be deleted
                             if false, only requested page will be deleted
    """

    request_headers = _get_request_header()
    title_url = URL_BASE + "?title=" + page_title

    response = requests.get(url=title_url, headers=request_headers, verify=True)
    response.raise_for_status()

    result = response.json()
    if result["results"]:
        if result["results"][0]["title"] == page_title:
            page_id = result["results"][0]["id"]
            if delete_sub_pages:
                search_url = URL_BASE + "/search?cql=ancestor=" + page_id
                response = requests.get(url=search_url, headers=request_headers, verify=True)
                response.raise_for_status()
                result = response.json()

                if result["results"]:
                    for entry in result["results"]:
                        _move_page_to_recycle_bin(entry['id'], request_headers)

            _move_page_to_recycle_bin(page_id, request_headers)

create_autorization.py

import getpass
from base64 import b64encode


def create_auth():
    encoded_credentials = b64encode(bytes(f'{username}:{password}', encoding='ascii')).decode('ascii')

    auth_header = f'Basic {encoded_credentials}'
    with open('./auth', 'w') as auth_file:
        auth_file.write(auth_header)


if __name__ == '__main__':
    username = input('Confluence-Benutzername: ')
    password = getpass.getpass(prompt=f'Passwort für {username}: ')
    create_auth()