ChatGPT-Confluence/confluence_utils.py at master · mapaction/ChatGPT-Confluence · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
import os
import atlassian
from atlassian import Confluence
import logging

# Configure logging (you can customize the level and format as needed)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def get_url():
    """
    Retrieves the Confluence URL from environment variables.

    Returns:
        str or None: The Confluence URL or None if not found.
    """

    url = os.getenv("confluence-url")
    if not url:
        logging.error("Error: 'confluence-url' environment variable not found.")
        return None
    return url

def get_spaces(confluence):
    """
    Retrieves a list of spaces from Confluence.

    Args:
        confluence: A Confluence API client instance.

    Returns:
        list: A list of dictionaries representing the spaces,
              each containing 'key' and 'name' attributes.
    """

    try:
        all_spaces = []
        start = 0
        limit = 10  # Adjust as needed
        while True:
            results = confluence.get_all_spaces(start=start, limit=limit)
            all_spaces.extend(results['results'])

            if '_links' in results and 'next' in results['_links']:
                start += limit
            else:
                break

        spaces = [{'key': space['key'], 'name': space['name']} for space in all_spaces]
        return spaces

    except atlassian.errors.ApiError as e:
        logging.error(f"Error fetching spaces from Confluence: {e}")
        return []

def connect_to_Confluence():
    """
    Connect to Confluence using Basic Authentication (email and API token).

    Returns:
        Confluence or None: A connector to Confluence or None if the connection fails.
    """

    try:
        url = get_url()
        if url is None:
            return None

        username = os.getenv("confluence-username")
        api_token = os.getenv("confluence-api-token")

        confluence = Confluence(
            url=url,
            username=username,
            password=api_token,
            cloud=True
        )

        return confluence

    except atlassian.errors.ApiPermissionError as e:
        logging.error(f"Error: Insufficient permissions to access Confluence: {e}")
        return None

    except atlassian.errors.ApiError as e:
        logging.error(f"Error connecting to Confluence: {e}")
        return None

def get_all_pages(confluence, space):
    '''
    Get all the pages within the specified Confluence space.

    Args:
        confluence: A Confluence API client instance.
        space: The key of the Confluence space.

    Returns:
        list: A list of page objects. Each page object contains information about a Confluence page.
    '''

    keep_going = True
    start = 0
    limit = 100
    pages = []
    while keep_going:
        results = confluence.get_all_pages_from_space(space, start=start, limit=limit, status=None, expand='body.storage,history', content_type='page')

        for page in results:
            try:
                if 'createdBy' in page['history']:
                    page['creator'] = page['history']['createdBy']['displayName']
                else:
                    page['creator'] = "Unknown"

                if 'createdDate' in page['history']:
                    page['created_date'] = page['history']['createdDate']
                else:
                    page['created_date'] = "Unknown"

                if 'lastUpdated' in page['history']:
                    page['last_updated'] = page['history']['lastUpdated']['when']
                else:
                    page['last_updated'] = "Unknown"

                if 'numberContentViews' in page['history']:
                    page['views'] = page['history']['numberContentViews']
                else:
                    page['views'] = 0

            except atlassian.errors.ApiError as e:
                logging.error(f"Error fetching details for page '{page['title']}': {e}")
                page['creator'] = "Unknown"
                page['created_date'] = "Unknown"
                page['last_updated'] = "Unknown"
                page['views'] = -1

        pages.extend(results)
        if len(results) < limit:
            keep_going = False
        else:
            start = start + limit
    return pages