Coverage for baibu/submissions/admin.py: 100%

1from django.contrib import admin 

2from django.contrib import messages 

3from django.utils.html import format_html 

4from django.utils.translation import gettext_lazy as _ 

5from django.utils.translation import ngettext 

6 

7from baibu.core import storage 

8 

9from .models import Submission 

10from .services import requeue 

11from .services import review 

12 

13 

14def _stored_text(key: str) -> str: 

15 payload = storage.read_json(key) if key else None 

16 if payload is None: 

17 return "—" 

18 return format_html('<div style="white-space: pre-wrap; max-width: 60rem">{}</div>', payload.get("text", "")) 

19 

20 

21def _move(modeladmin, request, queryset, status): 

22 moved = sum(review(submission, status, reviewer=request.user) for submission in queryset) 

23 skipped = queryset.count() - moved 

24 modeladmin.message_user( 

25 request, ngettext("%(n)d submission updated.", "%(n)d submissions updated.", moved) % {"n": moved} 

26 ) 

27 if skipped: 

28 modeladmin.message_user( 

29 request, 

30 ngettext( 

31 "%(n)d submission skipped: that change is not allowed from its status.", 

32 "%(n)d submissions skipped: that change is not allowed from their status.", 

33 skipped, 

34 ) 

35 % {"n": skipped}, 

36 level=messages.WARNING, 

37 ) 

38 

39 

40@admin.register(Submission) 

41class SubmissionAdmin(admin.ModelAdmin): 

42 list_display = ("id", "user", "language_code", "status", "word_count", "consent_tier", "created_at") 

43 list_filter = ("status", "language_code", "consent_tier", "toxicity_detected") 

44 search_fields = ("id", "user__email", "excerpt") 

45 date_hierarchy = "created_at" 

46 actions = ("mark_verified", "mark_rejected", "clean_again") 

47 fieldsets = ( 

48 (None, {"fields": ("id", "user", "language_code", "status", "decision_reason", "staff_notes")}), 

49 (_("Review"), {"fields": ("reviewed_by", "reviewed_at")}), 

50 (_("Text"), {"fields": ("cleaned_text", "raw_text", "word_count", "character_count")}), 

51 ( 

52 _("Cleaning"), 

53 { 

54 "fields": ( 

55 "cleaner", 

56 "cleaned_at", 

57 "redaction_count", 

58 "toxicity_detected", 

59 "quality_score", 

60 "detected_language", 

61 "flags", 

62 ) 

63 }, 

64 ), 

65 (_("Consent and storage"), {"fields": ("consent_tier", "consent_record", "raw_key", "clean_key")}), 

66 (_("Dates"), {"fields": ("created_at", "updated_at")}), 

67 ) 

68 readonly_fields = tuple(f for _name, options in fieldsets for f in options["fields"] if f not in {"staff_notes"}) 

69 

70 def has_add_permission(self, request): 

71 return False 

72 

73 @admin.display(description=_("Cleaned text")) 

74 def cleaned_text(self, obj): 

75 return _stored_text(obj.clean_key) 

76 

77 @admin.display(description=_("Text as submitted")) 

78 def raw_text(self, obj): 

79 return _stored_text(obj.raw_key) 

80 

81 @admin.action(description=_("Mark as verified")) 

82 def mark_verified(self, request, queryset): 

83 _move(self, request, queryset, Submission.Status.VERIFIED) 

84 

85 @admin.action(description=_("Mark as rejected")) 

86 def mark_rejected(self, request, queryset): 

87 _move(self, request, queryset, Submission.Status.REJECTED) 

88 

89 @admin.action(description=_("Clean again")) 

90 def clean_again(self, request, queryset): 

91 queued = sum(requeue(submission) for submission in queryset) 

92 self.message_user( 

93 request, ngettext("%(n)d submission queued.", "%(n)d submissions queued.", queued) % {"n": queued} 

94 )