# -*- coding: utf-8 -*-
'''
Creates a full Confluence user export as CSV
1. IMPORTANT: should be executed with Python 3 interpreter
2. Change the username, password and baseurl variables to match custom Confluence
3. Execution from terminal: python3 create_full_export_csv_python3.py
4. A new file named confluence_users.csv is now available in current directory
'''
from urllib import request, parse, error
import json
import base64
# CHANGE THESE VARIABLES
CONFLUENCE_USERNAME = "admin"
CONFLUENCE_PASSWORD = "admin"
CONFLUENCE_BASEURL = "http://localhost:1990/confluence"
CONFLUENCE_CSV_FILE_NAME = "confluence-users.csv"
# Search variables
# eg. "admin" or "admin@admin.com"
CONFLUENCE_USER_EXPORT_SEARCH_TERM = ""
# True if enabled users is to be searched
CONFLUENCE_USER_EXPORT_ACTIVE_USERS = True
# True if disabled users is to be searched
CONFLUENCE_USER_EXPORT_INACTIVE_USERS = True
# Starting offset is 0
CONFLUENCE_USER_EXPORT_OFFSET = 0
# Pagesize. 100 is max page size
CONFLUENCE_USER_EXPORT_PAGESIZE = 20
CONFLUENCE_USER_EXPORT_SEARCH = "/rest/confluenceuserexport/1.0/search"
CONFLUENCE_USER_EXPORT_CSV_SERVLET = "/rest/confluenceuserexport/1.0/file/csv"
CONFLUENCE_USER_EXPORT_DOWNLOAD = "/plugins/servlet/confluenceuserexport/admin/download"
JSON_MIME_TYPE = "application/json"
BASIC_AUTH = "Basic " + base64.b64encode(bytes('%s:%s' % (CONFLUENCE_USERNAME, CONFLUENCE_PASSWORD), 'utf-8')).decode('utf-8')
all_users = []
def search_users(search_string, active_users, inactive_users, page_size, offset):
'''
Search for Confluence user with the given parameters
:param search_string: the given search string
:param active_users: true if active users
:param inactive_users: true if inactive users
:param page_size: the given page size eg. 20
:param offset: the given offset eg. 0
:return: None
'''''
request_headers = {"Content-type": JSON_MIME_TYPE, "Accept": JSON_MIME_TYPE, "Authorization": BASIC_AUTH}
request_url = CONFLUENCE_BASEURL + CONFLUENCE_USER_EXPORT_SEARCH
request_body = {"searchString":search_string,
"activeUsers":active_users,
"inActiveUsers":inactive_users,
"pageSize":page_size,
"offset":offset
}
post_request_json_bytes = json.dumps(request_body).encode("utf-8")
post_request = request.Request(request_url, headers=request_headers)
post_request.get_method = lambda: 'POST'
try:
post_request_open = request.urlopen(post_request, data=post_request_json_bytes)
post_response = post_request_open.read()
if post_response:
json_response = json.loads(s=post_response)
if json_response:
found_users = json_response.get("users")
# Check for next page of users in order to search for more users
has_next_page = json_response.get("hasNextPage")
for found_user in found_users:
all_users.append(found_user)
if has_next_page:
# Recursive search if next page
search_users(search_string, active_users, inactive_users, page_size, (offset + page_size))
except error.HTTPError as e:
print(e.getcode())
def get_csv_file(file_name):
'''
Get CSV file by file name from Confluence temp directory
:param file_name: the name of the file to get from Confluence temp directory
:return: None
'''
request_headers = {"Content-type": JSON_MIME_TYPE, "Accept": JSON_MIME_TYPE, "Authorization": BASIC_AUTH}
request_url = CONFLUENCE_BASEURL + CONFLUENCE_USER_EXPORT_DOWNLOAD
request_parameters = {'fileName':file_name}
get_request = request.Request(url=request_url + "?" + parse.urlencode(request_parameters),
headers=request_headers)
get_request.get_method = lambda: 'GET'
try:
get_request_open = request.urlopen(get_request)
get_response = get_request_open.read()
if (get_response):
with open(CONFLUENCE_CSV_FILE_NAME, 'wb') as csv_file:
csv_file.write(get_response)
except error.HTTPError as e:
print(e.getcode())
def create_csv_file(users_as_json):
'''
Send JSON content to CSV servlet for file generation
:param users_as_json: the Confluence users as JSON
:return: None
'''
users_as_json = users_as_json.encode("utf-8")
request_headers = {"Content-type": JSON_MIME_TYPE, "Accept": JSON_MIME_TYPE, "Authorization": BASIC_AUTH}
request_url = CONFLUENCE_BASEURL + CONFLUENCE_USER_EXPORT_CSV_SERVLET
post_request = request.Request(url=request_url, headers=request_headers)
post_request.get_method = lambda: 'POST'
try:
post_request_open = request.urlopen(post_request, data=users_as_json)
post_response = post_request_open.read()
if post_response:
json_data = json.loads(post_response)
if json_data and json_data.get("fileName"):
get_csv_file(file_name=json_data.get("fileName"))
except error.HTTPError as e:
print(e.getcode())
# Start searching
search_users(search_string=CONFLUENCE_USER_EXPORT_SEARCH_TERM,
active_users=CONFLUENCE_USER_EXPORT_ACTIVE_USERS,
inactive_users=CONFLUENCE_USER_EXPORT_INACTIVE_USERS,
page_size=CONFLUENCE_USER_EXPORT_PAGESIZE,
offset=CONFLUENCE_USER_EXPORT_OFFSET)
# Call CSV servlet for conversion
all_users_json = json.dumps(all_users)
create_csv_file(users_as_json=all_users_json) |