edx-platform/cms/djangoapps/modulestore_migrator/admin.py

"""
A nice little admin interface for migrating courses and libraries from modulstore to Learning Core.
"""
import logging

from django import forms
from django.contrib import admin, messages
from django.contrib.admin.helpers import ActionForm
from django.db import models


from opaque_keys import InvalidKeyError
from opaque_keys.edx.locator import LibraryCollectionLocator, LibraryLocatorV2
from user_tasks.models import UserTaskStatus

from openedx.core.types.http import AuthenticatedHttpRequest

from . import api
from .data import CompositionLevel, RepeatHandlingStrategy
from .models import ModulestoreSource, ModulestoreMigration, ModulestoreBlockSource, ModulestoreBlockMigration


log = logging.getLogger(__name__)


class StartMigrationTaskForm(ActionForm):
    """
    Params for start_migration_task admin adtion, displayed next the "Go" button.
    """
    target_key = forms.CharField(label="Target library or collection key →", required=False)
    repeat_handling_strategy = forms.ChoiceField(
        label="How to handle existing content? →",
        choices=RepeatHandlingStrategy.supported_choices,
        required=False,
    )
    preserve_url_slugs = forms.BooleanField(label="Preserve current slugs? →", required=False, initial=True)
    forward_to_target = forms.BooleanField(label="Forward references? →", required=False)
    composition_level = forms.ChoiceField(
        label="Aggregate up to →", choices=CompositionLevel.supported_choices, required=False
    )


def task_status_details(obj: ModulestoreMigration) -> str:
    """
    Return the state and, if available, details of the status of the migration.
    """
    details: str | None = None
    if obj.task_status.state == UserTaskStatus.FAILED:
        # Calling fail(msg) from a task should automatically generates an "Error" artifact with that msg.
        # https://django-user-tasks.readthedocs.io/en/latest/user_tasks.html#user_tasks.models.UserTaskStatus.fail
        if error_artifacts := obj.task_status.artifacts.filter(name="Error"):
            if error_text := error_artifacts.order_by("-created").first().text:
                details = error_text
    elif obj.task_status.state == UserTaskStatus.SUCCEEDED:
        details = f"Migrated {obj.block_migrations.count()} blocks"
    return f"{obj.task_status.state}: {details}" if details else obj.task_status.state


migration_admin_fields = (
    "target",
    "target_collection",
    "task_status",
    # The next line works, but django-stubs incorrectly thinks that these should all be strings,
    # so we will need to use type:ignore below.
    task_status_details,
    "composition_level",
    "repeat_handling_strategy",
    "preserve_url_slugs",
    "change_log",
    "staged_content",
)


class ModulestoreMigrationInline(admin.TabularInline):
    """
    Readonly table within the ModulestoreSource page; each row is a Migration from this Source.
    """
    model = ModulestoreMigration
    fk_name = "source"
    show_change_link = True
    readonly_fields = migration_admin_fields  # type: ignore[assignment]
    ordering = ("-task_status__created",)

    def has_add_permission(self, _request, _obj):
        return False


class ModulestoreBlockSourceInline(admin.TabularInline):
    """
    Readonly table within the ModulestoreSource page; each row is a BlockSource.
    """
    model = ModulestoreBlockSource
    fk_name = "overall_source"
    readonly_fields = (
        "key",
        "forwarded"
    )

    def has_add_permission(self, _request, _obj):
        return False


@admin.register(ModulestoreSource)
class ModulestoreSourceAdmin(admin.ModelAdmin):
    """
    Admin interface for source legacy libraries and courses.
    """
    readonly_fields = ("forwarded",)
    list_display = ("id", "key", "forwarded")
    actions = ["start_migration_task"]
    action_form = StartMigrationTaskForm
    inlines = [ModulestoreMigrationInline, ModulestoreBlockSourceInline]

    @admin.action(description="Start migration for selected sources")
    def start_migration_task(
        self,
        request: AuthenticatedHttpRequest,
        queryset: models.QuerySet[ModulestoreSource],
    ) -> None:
        """
        Start a migration for each selected source
        """
        form = StartMigrationTaskForm(request.POST)
        form.is_valid()
        target_key_string = form.cleaned_data['target_key']
        if not target_key_string:
            messages.add_message(request, messages.ERROR, "Target key is required")
            return
        try:
            target_library_key = LibraryLocatorV2.from_string(target_key_string)
            target_collection_slug = None
        except InvalidKeyError:
            try:
                target_collection_key = LibraryCollectionLocator.from_string(target_key_string)
                target_library_key = target_collection_key.lib_key
                target_collection_slug = target_collection_key.collection_id
            except InvalidKeyError:
                messages.add_message(request, messages.ERROR, f"Invalid target key: {target_key_string}")
                return
        started = 0
        total = 0
        for source in queryset:
            total += 1
            try:
                api.start_migration_to_library(
                    user=request.user,
                    source_key=source.key,
                    target_library_key=target_library_key,
                    target_collection_slug=target_collection_slug,
                    composition_level=CompositionLevel(form.cleaned_data['composition_level']),
                    repeat_handling_strategy=RepeatHandlingStrategy(form.cleaned_data['repeat_handling_strategy']),
                    preserve_url_slugs=form.cleaned_data['preserve_url_slugs'],
                    forward_source_to_target=form.cleaned_data['forward_to_target'],
                )
            except Exception as exc:  # pylint: disable=broad-except
                message = f"Failed to start migration {source.key} -> {target_key_string}"
                messages.add_message(request, messages.ERROR, f"{message}: {exc}")
                log.exception(message)
                continue
            started += 1
        click_in = "Click into the source objects to see migration details."

        if not started:
            messages.add_message(request, messages.WARNING, f"Failed to start {total} migration(s).")
        if started < total:
            messages.add_message(request, messages.WARNING, f"Started {started} of {total} migration(s). {click_in}")
        else:
            messages.add_message(request, messages.INFO, f"Started {started} migration(s). {click_in}")


class ModulestoreBlockMigrationInline(admin.TabularInline):
    """
    Readonly table witin the Migration admin; each row is a block
    """
    model = ModulestoreBlockMigration
    fk_name = "overall_migration"
    readonly_fields = (
        "source",
        "target",
        "change_log_record",
        "unsupported_reason",
    )
    list_display = ("id", *readonly_fields)


@admin.register(ModulestoreMigration)
class ModulestoreMigrationAdmin(admin.ModelAdmin):
    """
    Readonly admin page for viewing Migrations
    """
    readonly_fields = ("source", *migration_admin_fields)  # type: ignore[assignment]
    list_display = ("id", "source", *migration_admin_fields)  # type: ignore[assignment]
    inlines = [ModulestoreBlockMigrationInline]