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

1"""Creating submissions and applying cleaning results.""" 

2 

3import logging 

4import uuid 

5 

6from django.conf import settings 

7from django.db import IntegrityError 

8from django.db import transaction 

9from django.utils import timezone 

10 

11from baibu.core import storage 

12from baibu.users.consent import consent_state 

13 

14from . import text as text_rules 

15from .cleaners import CleanerError 

16from .cleaners import CleaningResult 

17from .cleaners import get_cleaner 

18from .models import Submission 

19 

20logger = logging.getLogger(__name__) 

21 

22RAW_AREA = "submissions/raw" 

23CLEAN_AREA = "submissions/clean" 

24 

25 

26class DuplicateSubmissionError(Exception): 

27 pass 

28 

29 

30def create_submission(*, user, raw_text: str, language_code: str, source: str = "web") -> Submission: 

31 """Store the full text privately, record the submission and queue cleaning. 

32 

33 Raises ``DuplicateSubmissionError`` if the same text was submitted before, 

34 and lets storage errors through (nothing is recorded then). 

35 """ 

36 from .tasks import clean_submission 

37 

38 raw_text = text_rules.normalise(raw_text) 

39 digest = text_rules.content_hash(raw_text) 

40 if Submission.objects.filter(content_hash=digest).exists(): 

41 raise DuplicateSubmissionError 

42 consent = consent_state(user=user, scope=settings.SUBMISSION_CONSENT_SCOPE) 

43 submission_id = uuid.uuid7() 

44 raw_key = storage.save_json( 

45 storage.build_key(RAW_AREA, f"{submission_id}.json"), 

46 { 

47 "schema": "submission.raw.v1", 

48 "submission_id": str(submission_id), 

49 "language_code": language_code, 

50 "source": source, 

51 "received_at": timezone.now().isoformat(), 

52 "text": raw_text, 

53 }, 

54 ) 

55 try: 

56 with transaction.atomic(): 

57 submission = Submission.objects.create( 

58 id=submission_id, 

59 user=user, 

60 language_code=language_code, 

61 consent_tier=consent.tier, 

62 consent_record=consent.latest_record, 

63 raw_key=raw_key, 

64 content_hash=digest, 

65 word_count=text_rules.word_count(raw_text), 

66 character_count=len(raw_text), 

67 ) 

68 except IntegrityError as exc: 

69 # The same text arrived at the same moment. 

70 storage.delete(raw_key) 

71 raise DuplicateSubmissionError from exc 

72 except Exception: 

73 storage.delete(raw_key) 

74 raise 

75 transaction.on_commit(lambda: clean_submission.delay(str(submission.pk))) 

76 return submission 

77 

78 

79def decide(submission: Submission, result: CleaningResult) -> tuple[str, str]: 

80 """Return the status and reason for a cleaning result.""" 

81 if not result.text.strip(): 

82 return Submission.Status.NEEDS_REVIEW, "Empty after cleaning." 

83 if result.toxicity_detected: 

84 return Submission.Status.NEEDS_REVIEW, "Possibly toxic." 

85 minimum = settings.SUBMISSION_MIN_QUALITY 

86 if result.quality_score is not None and result.quality_score < minimum: 

87 return Submission.Status.NEEDS_REVIEW, f"Quality {result.quality_score} below {minimum}." 

88 if result.detected_language and result.detected_language.lower() != submission.language_code.lower(): 

89 return Submission.Status.NEEDS_REVIEW, f"Detected language {result.detected_language}." 

90 return Submission.Status.VERIFIED, "" 

91 

92 

93def _mark_issue(submission: Submission, reason: str) -> Submission: 

94 logger.warning("Submission %s could not be cleaned: %s", submission.pk, reason) 

95 submission.status = Submission.Status.ISSUE 

96 submission.decision_reason = reason[:512] 

97 submission.save(update_fields=["status", "decision_reason", "updated_at"]) 

98 return submission 

99 

100 

101def run_cleaning(submission: Submission) -> Submission: 

102 """Clean a pending submission and record the outcome.""" 

103 if submission.status != Submission.Status.PENDING: 

104 return submission 

105 raw = storage.read_json(submission.raw_key) 

106 if not raw or not isinstance(raw.get("text"), str): 

107 return _mark_issue(submission, "Raw text is missing.") 

108 try: 

109 result = get_cleaner().clean(raw["text"], language_code=submission.language_code) 

110 except CleanerError as exc: 

111 return _mark_issue(submission, str(exc)) 

112 status, reason = decide(submission, result) 

113 try: 

114 clean_key = storage.save_json( 

115 storage.build_key(CLEAN_AREA, f"{submission.pk}.json"), 

116 { 

117 "schema": "submission.clean.v1", 

118 "submission_id": str(submission.pk), 

119 "language_code": submission.language_code, 

120 "cleaner": result.cleaner, 

121 "cleaned_at": timezone.now().isoformat(), 

122 "text": result.text, 

123 }, 

124 ) 

125 except Exception as exc: 

126 logger.exception("Could not store cleaned text for submission %s", submission.pk) 

127 return _mark_issue(submission, f"Could not store cleaned text: {type(exc).__name__}") 

128 

129 old_clean_key = submission.clean_key 

130 submission.status = status 

131 submission.decision_reason = reason 

132 submission.clean_key = clean_key 

133 limit = settings.SUBMISSION_EXCERPT_LENGTH 

134 submission.excerpt = result.text if len(result.text) <= limit else result.text[:limit].rstrip() + "…" 

135 submission.cleaner = result.cleaner 

136 submission.redaction_count = result.redaction_count 

137 submission.toxicity_detected = result.toxicity_detected 

138 submission.quality_score = result.quality_score 

139 submission.detected_language = result.detected_language 

140 submission.flags = result.flags 

141 submission.cleaned_at = timezone.now() 

142 try: 

143 submission.save() 

144 except Exception: 

145 storage.delete(clean_key) 

146 raise 

147 if old_clean_key and old_clean_key != clean_key: 

148 transaction.on_commit(lambda: storage.delete(old_clean_key)) 

149 return submission 

150 

151 

152def requeue(submission: Submission) -> bool: 

153 """Send a submission back for cleaning ("clean again").""" 

154 from .tasks import clean_submission 

155 

156 with transaction.atomic(): 

157 locked = Submission.objects.select_for_update().get(pk=submission.pk) 

158 if not locked.can_transition_to(Submission.Status.PENDING): 

159 return False 

160 submission.status = Submission.Status.PENDING 

161 submission.save(update_fields=["status", "updated_at"]) 

162 transaction.on_commit(lambda: clean_submission.delay(str(submission.pk))) 

163 return True 

164 

165 

166def review(submission: Submission, status: str, *, reviewer=None) -> bool: 

167 """Record a staff decision on a submission and tell the contributor. 

168 

169 Returns ``False`` if the change is not allowed from the current status. 

170 """ 

171 from django.urls import reverse 

172 

173 from baibu.notifications.services import notify 

174 

175 kinds = {Submission.Status.VERIFIED: "submission_accepted", Submission.Status.REJECTED: "submission_rejected"} 

176 if status not in kinds: 

177 msg = f"Not a review decision: {status}" 

178 raise ValueError(msg) 

179 with transaction.atomic(): 

180 # Lock the row so two reviewers cannot both decide. 

181 locked = Submission.objects.select_for_update().get(pk=submission.pk) 

182 if not locked.can_transition_to(status): 

183 submission.status = locked.status 

184 return False 

185 submission.status = status 

186 submission.reviewed_by = reviewer 

187 submission.reviewed_at = timezone.now() 

188 submission.save(update_fields=["status", "reviewed_by", "reviewed_at", "updated_at"]) 

189 notify(submission.user, kind=kinds[status], link=reverse("submissions:list")) 

190 return True 

191 

192 

193def delete_files(submission: Submission) -> None: 

194 for key in (submission.raw_key, submission.clean_key): 

195 if key: 

196 storage.delete(key)