Coverage for castellum/recruitment/views.py: 82%

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

2# MPIB <https://www.mpib-berlin.mpg.de/>,

3# MPI-CBS <https://www.cbs.mpg.de/>,

4# MPIP <http://www.psych.mpg.de/>

6# This file is part of Castellum.

8# Castellum is free software; you can redistribute it and/or modify it

9# under the terms of the GNU Affero General Public License as published

10# by the Free Software Foundation; either version 3 of the License, or

11# (at your option) any later version.

12#

13# Castellum is distributed in the hope that it will be useful, but

14# WITHOUT ANY WARRANTY; without even the implied warranty of

15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU

16# Affero General Public License for more details.

17#

18# You should have received a copy of the GNU Affero General Public

19# License along with Castellum. If not, see

20# <http://www.gnu.org/licenses/>.

22import datetime

23import logging

24import random

25from collections import defaultdict

27from django.conf import settings

28from django.contrib import messages

29from django.core.exceptions import PermissionDenied

30from django.core.mail import EmailMessage

31from django.core.mail import get_connection

32from django.db import models

33from django.shortcuts import redirect

34from django.urls import reverse

35from django.utils import timezone

36from django.utils.translation import gettext_lazy as _

37from django.views.generic import FormView

38from django.views.generic import ListView

39from django.views.generic import UpdateView

41from simplecharts import StackedColumnRenderer

43from castellum.castellum_auth.mixins import PermissionRequiredMixin

44from castellum.contacts.mixins import BaseContactUpdateView

45from castellum.recruitment import filter_queries

46from castellum.recruitment.mixins import BaseAttributeSetUpdateView

47from castellum.recruitment.models.attributesets import get_description_by_statistics_rank

48from castellum.studies.mixins import StudyMixin

49from castellum.studies.models import Study

50from castellum.subjects.mixins import BaseAdditionalInfoUpdateView

51from castellum.subjects.mixins import BaseDataProtectionUpdateView

52from castellum.subjects.models import Subject

54from .forms import ContactForm

55from .forms import SendMailForm

56from .mixins import BaseCalendarView

57from .mixins import ParticipationRequestMixin

58from .models import AttributeDescription

59from .models import AttributeSet

60from .models import MailBatch

61from .models import ParticipationRequest

63logger = logging.getLogger(__name__)

66def get_recruitable(study):

67 """Get all subjects who are recruitable for this study, i.e. apply all filters."""

68 qs = Subject.objects.filter(

69 filter_queries.study(study),

70 filter_queries.has_consent(),

71 filter_queries.already_in_study(study),

72 )

74 last_contacted = timezone.now() - settings.CASTELLUM_PERIOD_BETWEEN_CONTACT_ATTEMPTS

75 qs = qs.annotate(last_contacted=models.Max(

76 'participationrequest__revisions__created_at',

77 filter=~models.Q(

78 participationrequest__revisions__status=ParticipationRequest.NOT_CONTACTED

79 ),

80 )).filter(

81 models.Q(last_contacted__lte=last_contacted) |

82 models.Q(last_contacted__isnull=True)

83 )

85 return list(qs)

88class RecruitmentView(StudyMixin, PermissionRequiredMixin, ListView):

89 model = ParticipationRequest

90 template_name = 'recruitment/recruitment.html'

91 permission_required = 'recruitment.view_participationrequest'

92 study_status = [Study.EXECUTION]

93 shown_status = []

94 tab = 'recruitment'

96 def dispatch(self, request, *args, **kwargs):

97 self.sort = self.request.GET.get('sort', self.request.session.get('recruitment_sort'))

98 self.request.session['recruitment_sort'] = self.sort

99 return super().dispatch(request, *args, **kwargs)

100

101 def get_queryset(self):

102 if self.sort == 'followup': 102 ↛ 103line 102 didn't jump to line 103, because the condition on line 102 was never true

103 order_by = ['followup_date', 'followup_time', '-updated_at']

104 elif self.sort == 'last_contact_attempt': 104 ↛ 105line 104 didn't jump to line 105, because the condition on line 104 was never true

105 order_by = ['-status_not_reached', '-updated_at']

106 elif self.sort == 'statistics': 106 ↛ 107line 106 didn't jump to line 107, because the condition on line 106 was never true

107 desc1 = get_description_by_statistics_rank('primary')

108 desc2 = get_description_by_statistics_rank('secondary')

109 order_by = [

110 '-subject__attributeset__data__' + desc1.json_key,

111 '-subject__attributeset__data__' + desc2.json_key,

112 '-updated_at',

113 ]

114 else:

115 order_by = [

116 '-followup_urgent', 'status', 'followup_date', 'followup_time', '-updated_at'

117 ]

118

119 return ParticipationRequest.objects\

120 .prefetch_related('subject__attributeset')\

121 .filter(status__in=self.shown_status, study=self.study)\

122 .order_by(*order_by)

123

124 def get_statistics(self):

125 participation_requests = ParticipationRequest.objects.filter(

126 study=self.study, status=ParticipationRequest.INVITED

127 ).prefetch_related('subject__attributeset')

128

129 buckets1 = AttributeDescription.get_statistics_buckets('primary')

130 buckets2 = AttributeDescription.get_statistics_buckets('secondary')

131

132 if len(buckets1) == 1: 132 ↛ 133line 132 didn't jump to line 133, because the condition on line 132 was never true

133 return None

134

135 statistics = defaultdict(lambda: defaultdict(lambda: 0))

136 for participation_request in participation_requests: 136 ↛ 137line 136 didn't jump to line 137, because the loop on line 136 never started

137 try:

138 attributeset = participation_request.subject.attributeset

139 key1 = attributeset.get_statistics_bucket('primary')

140 key2 = attributeset.get_statistics_bucket('secondary')

141 statistics[key1][key2] += 1

142 except AttributeSet.DoesNotExist:

143 pass

144

145 data = {

146 'rows': [{

147 'label': str(label1),

148 'values': [statistics[key1][key2] for key2, label2 in buckets2]

149 } for key1, label1 in buckets1],

150 }

151

152 if len(buckets2) > 1: 152 ↛ 155line 152 didn't jump to line 155, because the condition on line 152 was never false

153 data['legend'] = [str(label2) for key2, label2 in buckets2]

154

155 renderer = StackedColumnRenderer(width=760, height=320)

156 return renderer.render(data)

157

158 def get_context_data(self, **kwargs):

159 context = super().get_context_data(**kwargs)

160

161 buckets1 = dict(AttributeDescription.get_statistics_buckets('primary'))

162 buckets2 = dict(AttributeDescription.get_statistics_buckets('secondary'))

163

164 participants = []

165 for participation_request in self.get_queryset():

166 try:

167 attributeset = participation_request.subject.attributeset

168 bucket1 = buckets1[attributeset.get_statistics_bucket('primary')]

169 bucket2 = buckets2[attributeset.get_statistics_bucket('secondary')]

170 buckets = ', '.join(str(bucket) for bucket in [bucket1, bucket2] if bucket)

171 except AttributeSet.DoesNotExist:

172 buckets = ''

173

174 can_access = self.request.user.has_privacy_level(

175 participation_request.subject.privacy_level

176 )

177

178 participants.append((participation_request, buckets, can_access))

179 context['participants'] = participants

180

181 context['sort_options'] = [

182 ('relevance', _('Relevance')),

183 ('followup', _('Follow-up date')),

184 ('last_contact_attempt', _('Last contact attempt')),

185 ('statistics', _('Statistics')),

186 ]

187 sort_options = dict(context['sort_options'])

188 context['sort_label'] = sort_options.get(self.sort, _('Relevance'))

189

190 context['statistics'] = self.get_statistics()

191

192 queryset = ParticipationRequest.objects.filter(study=self.study)

193 context['all_count'] = queryset.count()

194 context['invited_count'] = queryset.filter(

195 status=ParticipationRequest.INVITED

196 ).count()

197 context['unsuitable_count'] = queryset.filter(

198 status=ParticipationRequest.UNSUITABLE

199 ).count()

200 context['open_count'] = queryset.filter(status__in=[

201 ParticipationRequest.NOT_CONTACTED,

202 ParticipationRequest.NOT_REACHED,

203 ParticipationRequest.FOLLOWUP_APPOINTED,

204 ParticipationRequest.AWAITING_RESPONSE,

205 ]).count()

206

207 context['last_agreeable_contact_time'] = (

208 datetime.date.today() - settings.CASTELLUM_PERIOD_BETWEEN_CONTACT_ATTEMPTS

209 )

210

211 return context

212

213

214class RecruitmentViewInvited(RecruitmentView):

215 shown_status = [ParticipationRequest.INVITED]

216 subtab = 'invited'

217

218

219class RecruitmentViewUnsuitable(RecruitmentView):

220 shown_status = [ParticipationRequest.UNSUITABLE]

221 subtab = 'unsuitable'

222

223

224class RecruitmentViewOpen(RecruitmentView):

225 shown_status = [

226 ParticipationRequest.NOT_CONTACTED,

227 ParticipationRequest.NOT_REACHED,

228 ParticipationRequest.FOLLOWUP_APPOINTED,

229 ParticipationRequest.AWAITING_RESPONSE,

230 ]

231 subtab = 'open'

232

233 def create_participation_requests(self, batch_size):

234 subjects = get_recruitable(self.study)

235

236 added = min(batch_size, len(subjects))

237 for subject in random.sample(subjects, k=added):

238 ParticipationRequest.objects.create(study=self.study, subject=subject)

239

240 return added

241

242 def post(self, request, *args, **kwargs):

243 if not request.user.has_perm('recruitment.add_participationrequest', obj=self.study): 243 ↛ 244line 243 didn't jump to line 244, because the condition on line 243 was never true

244 raise PermissionDenied

245

246 if self.study.min_subject_count == 0:

247 messages.error(request, _(

248 'This study does not require participants. '

249 'Please, contact the responsible person who can set it up.'

250 ))

251 return self.get(request, *args, **kwargs)

252

253 not_contacted_count = self.get_queryset()\

254 .filter(status=ParticipationRequest.NOT_CONTACTED).count()

255 hard_limit = self.study.min_subject_count * settings.CASTELLUM_RECRUITMENT_HARD_LIMIT_FACTOR

256

257 if not_contacted_count >= hard_limit:

258 messages.error(request, _(

259 'Application privacy does not allow you to add more subjects. '

260 'Please contact provided subjects before adding new ones.'

261 ))

262 return self.get(request, *args, **kwargs)

263

264 # Silently trim down to respect hard limit.

265 # In many cases, the soft limit warning will already be shown in that case.

266 batch_size = min(

267 settings.CASTELLUM_RECRUITMENT_BATCH_SIZE,

268 hard_limit - not_contacted_count,

269 )

270 try:

271 added = self.create_participation_requests(batch_size)

272 except KeyError:

273 added = 0

274 messages.error(request, _(

275 'The custom filter that is set for this study does not exist.'

276 ))

277

278 if not_contacted_count + added > settings.CASTELLUM_RECRUITMENT_SOFT_LIMIT:

279 messages.error(request, _(

280 'Such workflow is not intended due to privacy reasons. '

281 'Please contact provided subjects before adding new ones.'

282 ))

283

284 if added == 0:

285 messages.error(

286 self.request,

287 _('No potential participants could be found for this study.'),

288 )

289

290 elif added < batch_size:

291 messages.warning(

292 self.request,

293 _(

294 'Only {added} out of {batchsize} potential participants '

295 'could be found for this study. These have been added.'

296 ).format(added=added, batchsize=batch_size),

297 )

298

299 return self.get(request, *args, **kwargs)

300

301

302class MailRecruitmentView(StudyMixin, PermissionRequiredMixin, FormView):

303 permission_required = 'recruitment.view_participationrequest'

304 study_status = [Study.EXECUTION]

305 template_name = 'recruitment/mail.html'

306 tab = 'mail'

307 form_class = SendMailForm

308

309 def personalize_mail_body(self, mail_body, contact):

310 replace_first = mail_body.replace("{firstname}", contact.first_name)

311 replace_second = replace_first.replace("{lastname}", contact.last_name)

312 return replace_second

313

314 def get_mail_settings(self):

315 from_email = settings.CASTELLUM_RECRUITMENT_EMAIL or settings.DEFAULT_FROM_EMAIL

316 connection = get_connection(

317 host=settings.CASTELLUM_RECRUITMENT_EMAIL_HOST,

318 port=settings.CASTELLUM_RECRUITMENT_EMAIL_PORT,

319 username=settings.CASTELLUM_RECRUITMENT_EMAIL_USER,

320 password=settings.CASTELLUM_RECRUITMENT_EMAIL_PASSWORD,

321 use_tls=settings.CASTELLUM_RECRUITMENT_EMAIL_USE_TLS,

322 )

323 return from_email, connection

324

325 def get_contact_email(self, contact):

326 if contact.email:

327 return contact.email

328 else:

329 for data in contact.guardians.exclude(email='').values('email'): 329 ↛ exitline 329 didn't return from function 'get_contact_email', because the loop on line 329 didn't complete