feat: Update social_user uid using csv from admin panel (#35048)

common/djangoapps/third_party_auth/admin.py

@@ -1,15 +1,17 @@
 """
 Admin site configuration for third party authentication
 """
-
+import csv
 
 from config_models.admin import KeyedConfigurationModelAdmin
 from django import forms
-from django.contrib import admin
+from django.contrib import admin, messages
 from django.db import transaction
-from django.urls import reverse
+from django.http import Http404, HttpResponseRedirect
+from django.urls import path, reverse
 from django.utils.html import format_html
 from django.utils.translation import gettext_lazy as _
+from django.views.decorators.csrf import csrf_exempt
 
 from .models import (
     _PSA_OAUTH2_BACKENDS,
@@ -21,7 +23,7 @@ from .models import (
     SAMLProviderConfig,
     SAMLProviderData
 )
-from .tasks import fetch_saml_metadata
+from .tasks import fetch_saml_metadata, update_saml_users_social_auth_uid
 
 
 class OAuth2ProviderConfigForm(forms.ModelForm):
@@ -72,7 +74,7 @@ class SAMLProviderConfigAdmin(KeyedConfigurationModelAdmin):
         """ Don't show every single field in the admin change list """
         return (
             'name_with_update_link', 'enabled', 'site', 'entity_id', 'metadata_source',
-            'has_data', 'mode', 'saml_configuration', 'change_date', 'changed_by',
+            'has_data', 'mode', 'saml_configuration', 'change_date', 'changed_by', 'csv_uuid_update_button',
         )
 
     list_display_links = None
@@ -135,6 +137,65 @@ class SAMLProviderConfigAdmin(KeyedConfigurationModelAdmin):
         super().save_model(request, obj, form, change)
         fetch_saml_metadata.apply_async((), countdown=2)
 
+    def get_urls(self):
+        """ Extend the admin URLs to include the custom CSV upload URL. """
+        urls = super().get_urls()
+        custom_urls = [
+            path('<slug:slug>/upload-csv/', self.admin_site.admin_view(self.upload_csv), name='upload_csv'),
+
+        ]
+        return custom_urls + urls
+
+    @csrf_exempt
+    def upload_csv(self, request, slug):
+        """ Handle CSV upload and update UserSocialAuth model. """
+        if not request.user.is_staff:
+            raise Http404
+        if request.method == 'POST':
+            csv_file = request.FILES.get('csv_file')
+            if not csv_file or not csv_file.name.endswith('.csv'):
+                self.message_user(request, "Please upload a valid CSV file.", level=messages.ERROR)
+            else:
+                try:
+                    decoded_file = csv_file.read().decode('utf-8').splitlines()
+                    reader = csv.DictReader(decoded_file)
+                    update_saml_users_social_auth_uid(reader, slug)
+                    self.message_user(request, "CSV file has been processed successfully.")
+                except Exception as e:  # pylint: disable=broad-except
+                    self.message_user(request, f"Failed to process CSV file: {e}", level=messages.ERROR)
+
+        # Always redirect back to the SAMLProviderConfig listing page
+        return HttpResponseRedirect(reverse('admin:third_party_auth_samlproviderconfig_changelist'))
+
+    def change_view(self, request, object_slug, form_url='', extra_context=None):
+        """ Extend the change view to include CSV upload. """
+        extra_context = extra_context or {}
+        extra_context['show_csv_upload'] = True
+        return super().change_view(request, object_slug, form_url, extra_context)
+
+    def csv_uuid_update_button(self, obj):
+        """ Add CSV upload button to the form. """
+        if obj:
+            form_url = reverse('admin:upload_csv', args=[obj.slug])
+            return format_html(
+                '<form method="post" enctype="multipart/form-data" action="{}">'
+                '<input type="file" name="csv_file" accept=".csv" style="margin-bottom: 10px;">'
+                '<button type="submit" class="button">Upload CSV</button>'
+                '</form>',
+                form_url
+            )
+        return ""
+
+    csv_uuid_update_button.short_description = 'UUID UPDATE CSV'
+    csv_uuid_update_button.allow_tags = True
+
+    def get_readonly_fields(self, request, obj=None):
+        """ Conditionally add csv_uuid_update_button to readonly fields. """
+        readonly_fields = list(super().get_readonly_fields(request, obj))
+        if obj:
+            readonly_fields.append('csv_uuid_update_button')
+        return readonly_fields
+
 admin.site.register(SAMLProviderConfig, SAMLProviderConfigAdmin)
 
 

common/djangoapps/third_party_auth/tasks.py

@@ -7,9 +7,11 @@ import logging
 
 import requests
 from celery import shared_task
+from django.core.exceptions import ObjectDoesNotExist
 from edx_django_utils.monitoring import set_code_owner_attribute
 from lxml import etree
 from requests import exceptions
+from social_django.models import UserSocialAuth
 
 from common.djangoapps.third_party_auth.models import SAMLConfiguration, SAMLProviderConfig
 from common.djangoapps.third_party_auth.utils import (
@@ -127,3 +129,63 @@ def fetch_saml_metadata():
 
     # Return counts for total, skipped, attempted, updated, and failed, along with any failure messages
     return num_total, num_skipped, num_attempted, num_updated, len(failure_messages), failure_messages
+
+
+@shared_task
+@set_code_owner_attribute
+def update_saml_users_social_auth_uid(reader, slug):
+    """
+    Update the UserSocialAuth UID for users based on a CSV reader input.
+
+    This function reads old and new UIDs from a CSV reader, fetches the corresponding
+    SAMLProviderConfig object using the provided slug, and updates the UserSocialAuth
+    records accordingly.
+
+    Args:
+        reader (csv.DictReader): A CSV reader object that iterates over rows containing 'old-uid' and 'new-uid'.
+        slug (str): The slug of the SAMLProviderConfig object to be fetched.
+
+    Returns:
+        None
+    """
+    log_prefix = "UpdateSamlUsersAuthUID"
+    log.info(f"{log_prefix}: Updated user UID request received with slug: {slug}")
+
+    try:
+        # Fetching the SAMLProviderConfig object with slug
+        saml_provider_config = SAMLProviderConfig.objects.current_set().get(slug=slug)
+    except SAMLProviderConfig.DoesNotExist:
+        log.error(f"{log_prefix}: SAMLProviderConfig with slug {slug} does not exist")
+        return
+    except Exception as e:  # pylint: disable=broad-except
+        log.error(f"{log_prefix}: An error occurred while fetching SAMLProviderConfig: {str(e)}")
+        return
+
+    success_count = 0
+    error_count = 0
+
+    for row in reader:
+        old_uid = row.get('old-uid')
+        new_uid = row.get('new-uid')
+
+        # Construct the UID using the SAML provider slug and old UID
+        uid = f'{saml_provider_config.slug}:{old_uid}'
+
+        try:
+            user_social_auth = UserSocialAuth.objects.get(uid=uid)
+            user_social_auth.uid = f'{saml_provider_config.slug}:{new_uid}'
+            user_social_auth.save()
+            log.info(f"{log_prefix}: Updated UID from {old_uid} to {new_uid} for user:{user_social_auth.user.id}.")
+            success_count += 1
+
+        except ObjectDoesNotExist:
+            log.error(f"{log_prefix}: UserSocialAuth with UID {uid} does not exist for old UID {old_uid}")
+            error_count += 1
+
+        except Exception as e:  # pylint: disable=broad-except
+            log.error(f"{log_prefix}: An error occurred while updating UID for old UID {old_uid}"
+                      f" to new UID {new_uid}: {str(e)}")
+            error_count += 1
+
+    log.info(f"{log_prefix}: Process completed for SAML configuration with slug: {slug}, {success_count} records"
+             f" successfully processed, {error_count} records encountered errors")