feat: Add media cleanup functionality with UI and backend support for unused media management

2026-05-11 12:58:41 +01:00
parent c3131564e6
commit 8da8f27433
7 changed files with 392 additions and 3 deletions
@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import pytest
+from django.contrib.auth.models import Group
+from django.urls import reverse
+
+
+@pytest.fixture
+def cid_manager_user(db, django_user_model):
+    user = django_user_model.objects.create_user(
+        username="cid_manager",
+        email="cid_manager@example.com",
+        password="password123",
+    )
+    group, _ = Group.objects.get_or_create(name="cid_user_manager")
+    user.groups.add(group)
+    return user
+
+
+@pytest.mark.django_db
+def test_media_cleanup_requires_login(client):
+    response = client.get(reverse("media_cleanup"))
+    assert response.status_code == 302
+
+
+@pytest.mark.django_db
+def test_media_cleanup_requires_manager_group(client, django_user_model):
+    user = django_user_model.objects.create_user(
+        username="regular_user",
+        email="regular@example.com",
+        password="password123",
+    )
+    client.force_login(user)
+
+    response = client.get(reverse("media_cleanup"))
+    assert response.status_code == 403
+
+
+@pytest.mark.django_db
+def test_media_cleanup_page_for_manager(client, cid_manager_user):
+    client.force_login(cid_manager_user)
+
+    response = client.get(reverse("media_cleanup"))
+    assert response.status_code == 200
+    assert b"Unused Media Cleanup" in response.content
+
+
+@pytest.mark.django_db
+def test_media_cleanup_preview_uses_dry_run(client, cid_manager_user, monkeypatch):
+    client.force_login(cid_manager_user)
+
+    captured = {}
+
+    def fake_call_command(name, **kwargs):
+        captured["name"] = name
+        captured["kwargs"] = kwargs
+        kwargs["stdout"].write("preview ok")
+
+    monkeypatch.setattr("rad.views.call_command", fake_call_command)
+
+    response = client.post(
+        reverse("media_cleanup"),
+        {
+            "action": "preview",
+            "minimum_file_age": "30",
+            "remove_empty_dirs": "on",
+            "exclude": "thumbnails/*\ncache/*",
+        },
+        HTTP_HX_REQUEST="true",
+    )
+
+    assert response.status_code == 200
+    assert captured["name"] == "cleanup_unused_media"
+    assert captured["kwargs"]["dry_run"] is True
+    assert captured["kwargs"]["no_input"] is True
+    assert captured["kwargs"]["interactive"] is False
+    assert captured["kwargs"]["minimum_file_age"] == 30
+    assert captured["kwargs"]["remove_empty_dirs"] is True
+    assert captured["kwargs"]["exclude"] == ["thumbnails/*", "cache/*"]
+    assert b"preview ok" in response.content
+
+
+@pytest.mark.django_db
+def test_media_cleanup_live_run_disables_dry_run(client, cid_manager_user, monkeypatch):
+    client.force_login(cid_manager_user)
+
+    captured = {}
+
+    def fake_call_command(name, **kwargs):
+        captured["name"] = name
+        captured["kwargs"] = kwargs
+        kwargs["stdout"].write("cleanup ok")
+
+    monkeypatch.setattr("rad.views.call_command", fake_call_command)
+
+    response = client.post(
+        reverse("media_cleanup"),
+        {
+            "action": "cleanup",
+            "minimum_file_age": "0",
+        },
+        HTTP_HX_REQUEST="true",
+    )
+
+    assert response.status_code == 200
+    assert captured["name"] == "cleanup_unused_media"
+    assert captured["kwargs"]["dry_run"] is False
+    assert captured["kwargs"]["interactive"] is False
+    assert b"cleanup ok" in response.content
+    assert b"Recovered space" in response.content
@@ -94,6 +94,7 @@ urlpatterns = [
    ),
    path("server", views.server, name="server"),
    path("logs/", views.logs_view, name="logs_view"),
+    path("management/media-cleanup/", views.media_cleanup, name="media_cleanup"),
    path("people", views.people, name="people"),
    path("accounts/", views.UserListTableView.as_view(), name="accounts_list"),
    #path("accounts/", views.UserListView.as_view(), name="accounts_list"),
@@ -2,6 +2,7 @@ import secrets
 from typing import Any, Optional
 import os
 import re
+from io import StringIO
 from django.conf import settings

 from django_tables2 import SingleTableMixin
@@ -48,6 +49,7 @@ from django.http import Http404, JsonResponse

 from django.http import HttpResponseRedirect, HttpResponse
 from django.views.decorators.http import require_POST
+from django.core.management import call_command, CommandError

 from physics.models import UserAnswer as PhysicsUserAnswer
 from physics.models import Exam as PhysicsExam
@@ -107,6 +109,23 @@ import psutil



+def _directory_size_bytes(path: str) -> int:
+    """Return recursive directory size in bytes, skipping unreadable files."""
+    if not path or not os.path.exists(path):
+        return 0
+
+    total = 0
+    for root, _, files in os.walk(path):
+        for filename in files:
+            file_path = os.path.join(root, filename)
+            try:
+                total += os.path.getsize(file_path)
+            except OSError:
+                continue
+
+    return total
+
+
 def feedback_checker(user):
    return user.groups.filter(name="feedback_checker").exists()

@@ -409,6 +428,89 @@ def logs_view(request):
    )


+@login_required
+@user_is_cid_user_manager
+def media_cleanup(request):
+    """Run django-unused-media cleanup from a manager-facing maintenance page.
+
+    Scope: cid_user_manager group members and superusers.
+    Functionality: supports safe preview (dry-run) and confirmed deletion runs,
+    returning HTMX fragments for inline results.
+    """
+    context: dict[str, Any] = {
+        "ran": False,
+        "action": "preview",
+        "minimum_file_age": 60,
+        "remove_empty_dirs": True,
+        "exclude": "",
+    }
+
+    if request.method == "POST":
+        action = request.POST.get("action", "preview")
+        dry_run = action != "cleanup"
+        exclude_raw = (request.POST.get("exclude") or "").strip()
+        exclude = [line.strip() for line in exclude_raw.splitlines() if line.strip()]
+        remove_empty_dirs = request.POST.get("remove_empty_dirs") == "on"
+
+        min_age_raw = (request.POST.get("minimum_file_age") or "60").strip()
+        try:
+            minimum_file_age = max(0, int(min_age_raw))
+        except (TypeError, ValueError):
+            minimum_file_age = 0
+
+        out = StringIO()
+        success = True
+
+        command_kwargs: dict[str, Any] = {
+            "dry_run": dry_run,
+            "no_input": True,
+            "interactive": False,
+            "minimum_file_age": minimum_file_age,
+            "remove_empty_dirs": remove_empty_dirs,
+            "verbosity": 2,
+            "stdout": out,
+        }
+        if exclude:
+            command_kwargs["exclude"] = exclude
+
+        error_message = ""
+        media_size_before = _directory_size_bytes(settings.MEDIA_ROOT)
+        media_size_after = media_size_before
+        try:
+            call_command("cleanup_unused_media", **command_kwargs)
+            media_size_after = _directory_size_bytes(settings.MEDIA_ROOT)
+        except CommandError as exc:
+            success = False
+            error_message = str(exc)
+        except Exception as exc:  # pragma: no cover
+            success = False
+            error_message = str(exc)
+
+        output = out.getvalue().strip()
+        recovered_size = max(0, media_size_before - media_size_after)
+        context.update(
+            {
+                "ran": True,
+                "success": success,
+                "action": action,
+                "dry_run": dry_run,
+                "minimum_file_age": minimum_file_age,
+                "remove_empty_dirs": remove_empty_dirs,
+                "exclude": exclude_raw,
+                "output": output,
+                "error_message": error_message,
+                "media_size_before": media_size_before,
+                "media_size_after": media_size_after,
+                "recovered_size": recovered_size,
+            }
+        )
+
+        if request.htmx:
+            return render(request, "rad/partials/_media_cleanup_result.html", context)
+
+    return render(request, "rad/media_cleanup.html", context)
+
+
 def cid_selector(request):
    return render(
        request,