Switched Nextcloud to Nextcloud-AIO

_nextcloud/authentik_user_import.py

@@ -0,0 +1,101 @@
+import json
+import requests
+
+# Authentik API credentials and endpoints
+authentik_url = 'https://auth.delmar.bzh'
+authentik_api_key = 'wsH0RN4wHs6cgsTKRrOPUTpZkXXSr8NwDyhBSMDjNFCjSMiyoOXjSfjjRE2t'
+authentik_username = 'akadmin'
+
+# Read the contents of nextcloud_user_info.json
+print("Reading nextcloud_user_info.json...")
+with open('nextcloud_user_info.json', 'r') as file:
+    user_info = json.load(file)
+
+# Extract all groups from user records
+all_groups = set()
+for user_record in user_info:
+    all_groups.update(user_record["groups"])
+
+# Create a session
+session = requests.Session()
+
+# Define headers
+headers = {
+    'Accept': 'application/json',
+    'Authorization': 'Bearer ' + authentik_api_key,
+}
+
+# Fetch existing groups from Authentik
+print("Fetching existing groups from Authentik...")
+authentik_groups_url = f'{authentik_url}/api/v3/core/groups/'  # Updated API endpoint
+response = session.get(authentik_groups_url, headers=headers)
+print(f"Groups URL ({authentik_groups_url}")
+
+# Check for successful response and decode JSON if present
+if response.status_code == 200:
+    try:
+        authentik_groups = response.json()
+        print("Existing groups fetched successfully.")
+    except json.JSONDecodeError as e:
+        print(f"Failed to decode JSON response: {e}")
+        authentik_groups = []
+elif response.status_code == 404:
+    print("The requested resource could not be found. Please check the URL or the resource.")
+else:
+    print(f"Failed to fetch existing groups. Status Code: {response.status_code}")
+    authentik_groups = []
+
+# Check if groups from Nextcloud exist in Authentik, create missing groups
+missing_groups = all_groups - set(authentik_groups)
+for group in missing_groups:
+    group_data = {"name": group}
+    create_group_response = session.post(authentik_groups_url, json=group_data, headers=headers)
+    if create_group_response.status_code == 201:
+        print(f"Group '{group}' created successfully.")
+    else:
+        print(f"Failed to create group '{group}'. Status Code: {create_group_response.status_code}")
+
+# Fetch existing groups from Authentik
+print("Fetching existing groups from Authentik...")
+response = session.get(authentik_groups_url, headers=headers)
+authentik_groups = response.json()
+#print(authentik_groups)
+
+# Create a dictionary mapping group names to group UUIDs
+group_name_to_uuid = {group['name']: group['pk'] for group in authentik_groups['results']}
+print(group_name_to_uuid)
+
+# Import users to Authentik
+print(f"Importing users to Authentik ({authentik_url})...")
+for idx, user_record in enumerate(user_info, start=1):
+    print(f"Processing user {idx}/{len(user_info)} - ID: {user_record['user_id']}")
+
+    # Map Nextcloud groups to Authentik group UUIDs
+    authentik_group_uuids = [group_name_to_uuid[group] for group in user_record['groups'] if group in group_name_to_uuid]
+
+
+    # Format the user record according to Authentik's requirements
+    formatted_user = {
+        "username": user_record["user_id"],
+        "name": user_record["display_name"],
+        "email": user_record["email"],
+        "groups": authentik_group_uuids,
+        "is_active": True,  # Assuming all users are active
+        "type": "internal",  # Assuming all users are internal
+        # Add other fields as required by Authentik's API
+        # Map or transform data fields according to Authentik's expected format
+    }
+
+    # Send the formatted user data to Authentik's API endpoint for user import
+    user_import_url = f"{authentik_url}/api/v3/core/users/"  # Updated API endpoint
+    user_import_response = session.post(user_import_url, json=formatted_user, headers=headers  )
+
+    # Check response status or handle errors based on the API response
+    if user_import_response.status_code == 201:
+        print(f"User {formatted_user['username']} imported successfully.")
+    elif user_import_response.status_code == 400 and user_import_response.json().get('username') == ["This field must be unique."]:
+        print(f"User {formatted_user['username']} already exists. Skipping.")
+    else:
+        print(f"Failed to import user {formatted_user['username']}. Status Code: {user_import_response.status_code}")
+        # Print response content for debugging (if needed)
+        print(user_import_response.content)