Coverage for crateweb/consent/models.py: 41%
1200 statements
« prev ^ index » next coverage.py v7.8.0, created at 2025-08-27 10:34 -0500
1"""
2crate_anon/crateweb/consent/models.py
4===============================================================================
6 Copyright (C) 2015, University of Cambridge, Department of Psychiatry.
7 Created by Rudolf Cardinal (rnc1001@cam.ac.uk).
9 This file is part of CRATE.
11 CRATE is free software: you can redistribute it and/or modify
12 it under the terms of the GNU General Public License as published by
13 the Free Software Foundation, either version 3 of the License, or
14 (at your option) any later version.
16 CRATE is distributed in the hope that it will be useful,
17 but WITHOUT ANY WARRANTY; without even the implied warranty of
18 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19 GNU General Public License for more details.
21 You should have received a copy of the GNU General Public License
22 along with CRATE. If not, see <https://www.gnu.org/licenses/>.
24===============================================================================
26**Django ORM models for the consent-to-contact system.**
28"""
30import datetime
31from dateutil.relativedelta import relativedelta
32import logging
33import os
34from typing import Any, List, Optional, Tuple, Type, Union
36# from audit_log.models import AuthStampedModel # django-audit-log
37from cardinal_pythonlib.django.admin import admin_view_url
38from cardinal_pythonlib.django.fields.helpers import choice_explanation
39from cardinal_pythonlib.django.fields.restrictedcontentfile import (
40 ContentTypeRestrictedFileField,
41)
42from cardinal_pythonlib.django.files import (
43 auto_delete_files_on_instance_change,
44 auto_delete_files_on_instance_delete,
45)
46from cardinal_pythonlib.django.reprfunc import modelrepr
47from cardinal_pythonlib.httpconst import ContentType
48from cardinal_pythonlib.logs import BraceStyleAdapter
49from cardinal_pythonlib.pdf import get_concatenated_pdf_in_memory
50from cardinal_pythonlib.reprfunc import simple_repr
51from django import forms
52from django.conf import settings
53from django.contrib.auth import get_user_model
54from django.core.exceptions import ObjectDoesNotExist, ValidationError
55from django.core.mail import EmailMessage, EmailMultiAlternatives
56from django.core.validators import validate_email
57from django.db import models, transaction
58from django.db.models import Q, QuerySet
59from django.dispatch import receiver
60from django.http import QueryDict, Http404
61from django.http.request import HttpRequest
62from django.urls import reverse
63from django.utils import timezone
64from django.utils.crypto import get_random_string
65from django.utils.html import escape
67from crate_anon.crateweb.config.constants import (
68 ClinicalDatabaseType,
69 SOURCE_DB_NAME_MAX_LENGTH,
70 UrlNames,
71)
72from crate_anon.crateweb.consent.constants import EthicsInfo
73from crate_anon.crateweb.core.constants import (
74 LEN_ADDRESS,
75 LEN_FIELD_DESCRIPTION,
76 LEN_NAME,
77 LEN_PHONE,
78 LEN_TITLE,
79 MAX_HASH_LENGTH,
80)
81from crate_anon.crateweb.core.utils import (
82 site_absolute_url,
83 string_time_now,
84 url_with_querystring,
85)
86from crate_anon.crateweb.extra.pdf import (
87 make_pdf_on_disk_from_html_with_django_settings,
88 CratePdfPlan,
89)
90from crate_anon.crateweb.extra.salutation import (
91 forename_surname,
92 get_initial_surname_tuple_from_string,
93 salutation,
94 title_forename_surname,
95)
96from crate_anon.crateweb.consent.storage import privatestorage
97from crate_anon.crateweb.consent.tasks import (
98 email_rdbm_task,
99 process_consent_change,
100 process_contact_request,
101 finalize_clinician_response,
102)
103from crate_anon.crateweb.consent.teamlookup import get_teams
104from crate_anon.crateweb.consent.utils import (
105 days_to_years,
106 make_cpft_email_address,
107 render_email_html_to_string,
108 render_pdf_html_to_string,
109 to_date,
110 validate_researcher_email_domain,
111)
112from crate_anon.crateweb.research.models import get_mpid
113from crate_anon.crateweb.research.research_db_info import (
114 get_research_db_info,
115)
116from crate_anon.crateweb.userprofile.models import UserProfile
118log = BraceStyleAdapter(logging.getLogger(__name__))
120CLINICIAN_RESPONSE_FWD_REF = "ClinicianResponse"
121CONSENT_MODE_FWD_REF = "ConsentMode"
122CONTACT_REQUEST_FWD_REF = "ContactRequest"
123EMAIL_FWD_REF = "Email"
124EMAIL_TRANSMISSION_FWD_REF = "EmailTransmission"
125LEAFLET_FWD_REF = "Leaflet"
126LETTER_FWD_REF = "Letter"
127STUDY_FWD_REF = "Study"
129TEST_ID = -1
130TEST_ID_STR = str(TEST_ID)
131TEST_ID_TWO = -2
132TEST_ID_TWO_STR = str(TEST_ID_TWO)
133TEST_ID_STRINGS = (TEST_ID_STR, TEST_ID_TWO_STR)
136# =============================================================================
137# Study
138# =============================================================================
141def study_details_upload_to(instance: STUDY_FWD_REF, filename: str) -> str:
142 """
143 Determines the filename used for study information PDF uploads.
145 Args:
146 instance: instance of :class:`Study` (potentially unsaved;
147 and you can't call :func:`save`; it goes into infinite recursion)
148 filename: uploaded filename
150 Returns:
151 filename with extension but without path, to be used on the server
152 filesystem
153 """
154 extension = os.path.splitext(filename)[1] # includes the '.' if present
155 return os.path.join(
156 "study",
157 f"{instance.institutional_id}_details_{string_time_now()}{extension}",
158 )
159 # ... as id may not exist yet
162def study_form_upload_to(instance: STUDY_FWD_REF, filename: str) -> str:
163 """
164 Determines the filename used for study clinician-form PDF uploads.
166 Args:
167 instance: instance of :class:`Study` (potentially unsaved
168 and you can't call :func:`save`; it goes into infinite recursion)
169 filename: uploaded filename
171 Returns:
172 filename with extension but without path, to be used on the server
173 filesystem
174 """
175 extension = os.path.splitext(filename)[1]
176 return os.path.join(
177 "study",
178 f"{instance.institutional_id}_form_{string_time_now()}{extension}",
179 )
182class Study(models.Model):
183 """
184 Represents a research study.
185 """
187 # implicit 'id' field
188 institutional_id = models.PositiveIntegerField(
189 verbose_name="Institutional (e.g. NHS Trust) study number", unique=True
190 )
191 title = models.CharField(max_length=255, verbose_name="Study title")
192 lead_researcher = models.ForeignKey(
193 settings.AUTH_USER_MODEL,
194 on_delete=models.PROTECT,
195 related_name="studies_as_lead",
196 )
197 researchers = models.ManyToManyField(
198 settings.AUTH_USER_MODEL,
199 related_name="studies_as_researcher",
200 blank=True,
201 )
202 registered_at = models.DateTimeField(
203 null=True, blank=True, verbose_name="When was the study registered?"
204 )
205 summary = models.TextField(verbose_name="Summary of study")
206 summary_is_html = models.BooleanField(
207 default=False,
208 verbose_name="Is the summary in HTML (not plain text) format?",
209 )
210 search_methods_planned = models.TextField(
211 blank=True, verbose_name="Search methods planned"
212 )
213 patient_contact = models.BooleanField(
214 verbose_name="Involves patient contact?"
215 )
216 include_under_16s = models.BooleanField(
217 verbose_name="Include patients under 16?"
218 )
219 include_lack_capacity = models.BooleanField(
220 verbose_name="Include patients lacking capacity?"
221 )
222 clinical_trial = models.BooleanField(
223 verbose_name="Clinical trial (CTIMP)?"
224 )
225 include_discharged = models.BooleanField(
226 verbose_name="Include discharged patients?"
227 )
228 request_direct_approach = models.BooleanField(
229 verbose_name="Researchers request direct approach to patients?"
230 )
231 approved_by_rec = models.BooleanField(verbose_name="Approved by REC?")
232 rec_reference = models.CharField(
233 max_length=50,
234 blank=True,
235 verbose_name="Research Ethics Committee reference",
236 )
237 approved_locally = models.BooleanField(
238 verbose_name="Approved by local institution?"
239 )
240 local_approval_at = models.DateTimeField(
241 null=True,
242 blank=True,
243 verbose_name="When approved by local institution?",
244 )
245 study_details_pdf = ContentTypeRestrictedFileField(
246 blank=True,
247 storage=privatestorage,
248 content_types=[ContentType.PDF],
249 max_upload_size=settings.MAX_UPLOAD_SIZE_BYTES,
250 upload_to=study_details_upload_to,
251 )
252 subject_form_template_pdf = ContentTypeRestrictedFileField(
253 blank=True,
254 storage=privatestorage,
255 content_types=[ContentType.PDF],
256 max_upload_size=settings.MAX_UPLOAD_SIZE_BYTES,
257 upload_to=study_form_upload_to,
258 )
259 # http://nemesisdesign.net/blog/coding/django-private-file-upload-and-serving/ # noqa: E501
260 # https://stackoverflow.com/questions/8609192/differentiate-null-true-blank-true-in-django # noqa: E501
261 AUTODELETE_OLD_FILE_FIELDS = [
262 "study_details_pdf",
263 "subject_form_template_pdf",
264 ]
266 class Meta:
267 verbose_name_plural = "studies"
269 def __str__(self) -> str:
270 # noinspection PyUnresolvedReferences
271 return (
272 f"[Study {self.id}] {self.institutional_id}: "
273 f"{self.lead_researcher.get_full_name()} / {self.title}"
274 )
276 def __repr__(self) -> str:
277 return modelrepr(self)
279 def get_lead_researcher_name_address(self) -> List[str]:
280 """
281 Returns name/address components (as lines you might use on a letter or
282 envelope) for the study's lead researcher.
283 """
284 # noinspection PyUnresolvedReferences
285 return [
286 self.lead_researcher.profile.get_title_forename_surname()
287 ] + self.lead_researcher.profile.get_address_components()
289 def get_lead_researcher_salutation(self) -> str:
290 """
291 Returns the salutation for the study's lead researcher (e.g.
292 "Prof. Jones").
293 """
294 # noinspection PyUnresolvedReferences
295 return self.lead_researcher.profile.get_salutation()
297 def get_involves_lack_of_capacity(self) -> str:
298 """
299 Returns a human-readable string indicating whether or not the study
300 involves patients lacking capacity (and if so, whether it's a clinical
301 trial [CTIMP]).
302 """
303 if not self.include_lack_capacity:
304 return "No"
305 if self.clinical_trial:
306 return "Yes (and it is a clinical trial)"
307 return "Yes (and it is not a clinical trial)"
309 @staticmethod
310 def get_queryset_possible_contact_studies() -> QuerySet:
311 """
312 Returns all approved studies involving direct patient contact that
313 have a properly identifiable lead researcher.
314 """
315 return (
316 Study.objects.filter(patient_contact=True)
317 .filter(approved_by_rec=True)
318 .filter(approved_locally=True)
319 .exclude(study_details_pdf="")
320 .exclude(lead_researcher__profile__title="")
321 .exclude(lead_researcher__first_name="")
322 .exclude(lead_researcher__last_name="")
323 )
325 @staticmethod
326 def filter_studies_for_researcher(
327 queryset: QuerySet, user: settings.AUTH_USER_MODEL
328 ) -> QuerySet:
329 """
330 Filters the supplied query set down to those studies for which the
331 given user is a researcher on.
332 """
333 return queryset.filter(
334 Q(lead_researcher=user) | Q(researchers__in=[user])
335 ).distinct()
337 @property
338 def html_summary(self) -> str:
339 """
340 Returns a version of the study's summary with HTML tags marking up
341 paragraphs. If the summary is already in HTML format, just return
342 that.
343 """
344 # Check if summary exists and if not return the empty string
345 summary = self.summary
346 if not summary:
347 return ""
349 # If the summary is already HTML, return it as it is.
350 if self.summary_is_html:
351 return summary
353 # Split lines and ensure each line is HTML-escaped (e.g. if there's a
354 # "<" or similar in the raw text).
355 paragraphs = [escape(x) for x in summary.splitlines() if x]
357 # NB an equivalent to
358 # [x for x in something if x]
359 # is
360 # list(filter(None, something))
362 if len(paragraphs) <= 1:
363 # 0 or 1 paragraphs; no point using <p>
364 return "".join(paragraphs)
365 # Otherwise:
367 # Method 1: with <p>
368 # Visually better once CSS fixed.
369 return "".join(f"<p>{x}</p>" for x in paragraphs)
371 # Method 2: with <br>
372 # Wider gaps.
373 # return "<br><br>".join(paragraphs)
376# noinspection PyUnusedLocal
377@receiver(models.signals.post_delete, sender=Study)
378def auto_delete_study_files_on_delete(
379 sender: Type[Study], instance: Study, **kwargs: Any
380) -> None:
381 """
382 Django signal receiver.
384 Deletes files from filesystem when :class:`Study` object is deleted.
385 """
386 auto_delete_files_on_instance_delete(
387 instance, Study.AUTODELETE_OLD_FILE_FIELDS
388 )
391# noinspection PyUnusedLocal
392@receiver(models.signals.pre_save, sender=Study)
393def auto_delete_study_files_on_change(
394 sender: Type[Study], instance: Study, **kwargs: Any
395) -> None:
396 """
397 Django signal receiver.
399 Deletes files from filesystem when :class:`Study` object is changed.
400 """
401 auto_delete_files_on_instance_change(
402 instance, Study.AUTODELETE_OLD_FILE_FIELDS, Study
403 )
406# =============================================================================
407# Generic leaflets
408# =============================================================================
411def leaflet_upload_to(instance: LEAFLET_FWD_REF, filename: str) -> str:
412 """
413 Determines the filename used for leaflet uploads.
415 Args:
416 instance: instance of :class:`Leaflet` (potentially unsaved;
417 and you can't call :func:`save`; it goes into infinite recursion)
418 filename: uploaded filename
420 Returns:
421 filename with extension but without path, to be used on the server
422 filesystem
423 """
424 extension = os.path.splitext(filename)[1] # includes the '.' if present
425 return os.path.join(
426 "leaflet", f"{instance.name}_{string_time_now()}{extension}"
427 )
428 # ... as id may not exist yet
431class Leaflet(models.Model):
432 """
433 Represents a system-wide patient information leaflet.
434 """
436 CPFT_TPIR = "cpft_tpir" # mandatory
437 NIHR_YHRSL = "nihr_yhrsl" # not used automatically
438 CPFT_TRAFFICLIGHT_CHOICE = "cpft_trafficlight_choice"
439 CPFT_CLINRES = "cpft_clinres"
441 LEAFLET_CHOICES = (
442 (CPFT_TPIR, "CPFT: Taking part in research [MANDATORY]"),
443 (
444 NIHR_YHRSL,
445 "NIHR: Your health records save lives [not currently used]",
446 ),
447 (
448 CPFT_TRAFFICLIGHT_CHOICE,
449 "CPFT: traffic-light choice decision form [not currently used: "
450 "personalized version created instead]",
451 ),
452 (CPFT_CLINRES, "CPFT: clinical research [not currently used]"),
453 )
454 # https://docs.djangoproject.com/en/dev/ref/models/fields/#django.db.models.Field.choices # noqa: E501
456 name = models.CharField(
457 max_length=50,
458 unique=True,
459 choices=LEAFLET_CHOICES,
460 verbose_name="leaflet name",
461 )
462 pdf = ContentTypeRestrictedFileField(
463 blank=True,
464 storage=privatestorage,
465 content_types=[ContentType.PDF],
466 max_upload_size=settings.MAX_UPLOAD_SIZE_BYTES,
467 upload_to=leaflet_upload_to,
468 )
470 def __str__(self) -> str:
471 for x in Leaflet.LEAFLET_CHOICES:
472 if x[0] == self.name:
473 name = x[1]
474 if not self.pdf:
475 name += " (MISSING)"
476 return name
477 return f"? (bad name: {self.name})"
479 @staticmethod
480 def populate() -> None:
481 """
482 Pre-create records for all the system-wide leaflets we use.
483 """
484 keys = [x[0] for x in Leaflet.LEAFLET_CHOICES]
485 for x in keys:
486 if not Leaflet.objects.filter(name=x).exists():
487 obj = Leaflet(name=x)
488 obj.save()
491# noinspection PyUnusedLocal
492@receiver(models.signals.post_delete, sender=Leaflet)
493def auto_delete_leaflet_files_on_delete(
494 sender: Type[Leaflet], instance: Leaflet, **kwargs: Any
495) -> None:
496 """
497 Django signal receiver.
499 Deletes files from filesystem when :class:`Leaflet` object is deleted.
500 """
501 auto_delete_files_on_instance_delete(instance, ["pdf"])
504# noinspection PyUnusedLocal
505@receiver(models.signals.pre_save, sender=Leaflet)
506def auto_delete_leaflet_files_on_change(
507 sender: Type[Leaflet], instance: Leaflet, **kwargs: Any
508) -> None:
509 """
510 Django signal receiver.
511 Deletes files from filesystem when Leaflet object is changed.
512 """
513 auto_delete_files_on_instance_change(instance, ["pdf"], Leaflet)
516# =============================================================================
517# Generic fields for decisions
518# =============================================================================
521class Decision(models.Model):
522 """
523 Abstract class to represent how a decision has been made (e.g. by a patient
524 or their surrogate decision-maker or clinician).
525 """
527 # Note that Decision._meta.get_fields() doesn't care about the
528 # ordering of its fields (and, I think, they can change). So:
529 FIELDS = [
530 "decision_signed_by_patient",
531 "decision_otherwise_directly_authorized_by_patient",
532 "decision_under16_signed_by_parent",
533 "decision_under16_signed_by_clinician",
534 "decision_lack_capacity_signed_by_representative",
535 "decision_lack_capacity_signed_by_clinician",
536 ]
537 decision_signed_by_patient = models.BooleanField(
538 default=False, verbose_name="Request signed by patient?"
539 )
540 decision_otherwise_directly_authorized_by_patient = models.BooleanField(
541 default=False,
542 verbose_name="Request otherwise directly authorized by patient?",
543 )
544 decision_under16_signed_by_parent = models.BooleanField(
545 default=False,
546 verbose_name="Patient under 16 and request countersigned by parent?",
547 )
548 decision_under16_signed_by_clinician = models.BooleanField(
549 default=False,
550 verbose_name="Patient under 16 and request countersigned by "
551 "clinician?",
552 )
553 decision_lack_capacity_signed_by_representative = models.BooleanField(
554 default=False,
555 verbose_name="Patient lacked capacity and request signed by "
556 "authorized representative?",
557 )
558 decision_lack_capacity_signed_by_clinician = models.BooleanField(
559 default=False,
560 verbose_name="Patient lacked capacity and request countersigned by "
561 "clinician?",
562 )
564 class Meta:
565 abstract = True
567 def decision_valid(self) -> bool:
568 """
569 Does the decision meet our rules about who can make decisions?
570 """
571 # We can never electronically validate being under 16 (time may have
572 # passed since the lookup) or, especially, lacking capacity, so let's
573 # just trust the user
574 return (
575 (
576 self.decision_signed_by_patient
577 or self.decision_otherwise_directly_authorized_by_patient
578 )
579 or (
580 # Lacks capacity
581 self.decision_lack_capacity_signed_by_representative
582 and self.decision_lack_capacity_signed_by_clinician
583 )
584 or (
585 # Under 16: 2/3 rule
586 int(
587 self.decision_signed_by_patient
588 or self.decision_otherwise_directly_authorized_by_patient
589 )
590 + int(self.decision_under16_signed_by_parent)
591 + int(self.decision_under16_signed_by_clinician)
592 >= 2
593 # I know the logic overlaps. But there you go.
594 )
595 )
597 def validate_decision(self) -> None:
598 """
599 Ensure that the decision is valid according to our rules, or raise
600 :exc:`django.forms.ValidationError`.
601 """
602 if not self.decision_valid():
603 raise forms.ValidationError(
604 "Invalid decision. Options are: "
605 "(*) Signed/authorized by patient. "
606 "(*) Lacks capacity - signed by rep + clinician. "
607 "(*) Under 16 - signed by 2/3 of (patient, clinician, "
608 "parent); see special rules"
609 )
612# =============================================================================
613# Information about patient captured from clinical database
614# =============================================================================
617class ClinicianInfoHolder:
618 """
619 Represents information about a clinician (relating to a patient, whose
620 details are not held here). Also embodies information about which sort of
621 clinician we prefer to ask about patient contact requests (via
622 :attr:`clinician_preference_order`).
624 Python object only; not stored in the database.
625 """
627 CARE_COORDINATOR = "care_coordinator"
628 CONSULTANT = "consultant"
629 HCP = "HCP"
630 TEAM = "team"
632 def __init__(
633 self,
634 clinician_type: str,
635 title: str,
636 first_name: str,
637 surname: str,
638 email: str,
639 signatory_title: str,
640 is_consultant: bool,
641 start_date: Union[datetime.date, datetime.datetime] = None,
642 end_date: Union[datetime.date, datetime.datetime] = None,
643 address_components: List[str] = None,
644 ) -> None:
645 """
646 Args:
647 clinician_type: one of our special strings indicating what type
648 of clinician (e.g. care coordinator, consultant, other
649 healthcare professional, authorized clinical team
650 representative).
651 title: clinician's title
652 first_name: clinician's first name
653 surname: clinician's surname
654 email: clinician's e-mail address
655 signatory_title: when the clinician signs a letter, what should
656 go under their name to say what job they do? (For example:
657 "Consultant psychiatrist"; "Care coordinator").
658 is_consultant: is the clinician an NHS consultant? (Relates to
659 special legal rules regarding CTIMPs.)
660 start_date:
661 when did this clinician's involvement start?
662 end_date:
663 when did this clinician's involvement end?
664 address_components:
665 address lines for the clinician
666 """
667 self.clinician_type = clinician_type
668 self.title = title
669 self.first_name = first_name
670 self.surname = surname
671 self.email = email or make_cpft_email_address(first_name, surname)
672 self.signatory_title = signatory_title
673 self.is_consultant = is_consultant
674 self.start_date = to_date(start_date)
675 self.end_date = to_date(end_date)
676 self.address_components = address_components or [] # type: List[str]
678 if clinician_type == self.CARE_COORDINATOR:
679 self.clinician_preference_order = 1 # best
680 elif clinician_type == self.CONSULTANT:
681 self.clinician_preference_order = 2
682 elif clinician_type == self.HCP:
683 self.clinician_preference_order = 3
684 elif clinician_type == self.TEAM:
685 self.clinician_preference_order = 4
686 else:
687 self.clinician_preference_order = 99999 # worst
689 def __repr__(self) -> str:
690 return simple_repr(
691 self,
692 [
693 "clinician_type",
694 "title",
695 "first_name",
696 "surname",
697 "email",
698 "signatory_title",
699 "is_consultant",
700 "start_date",
701 "end_date",
702 "address_components",
703 ],
704 )
706 def current(self) -> bool:
707 """
708 Is the clinician currently involved (with this patient's care)?
709 """
710 return self.end_date is None or self.end_date >= datetime.date.today()
712 def contactable(self) -> bool:
713 """
714 Do we have enough information to contact the clinician
715 (electronically)?
716 """
717 return bool(self.surname and self.email)
720class PatientLookupBase(models.Model):
721 """
722 Base class for :class:`PatientLookup` and :class:`DummyPatientSourceInfo`.
724 Must be able to be instantiate with defaults, for the "not found"
725 situation.
727 Note that derived classes must implement ``nhs_number`` as a column.
728 """
730 MALE = "M"
731 FEMALE = "F"
732 INTERSEX = "X"
733 UNKNOWNSEX = "?"
734 SEX_CHOICES = (
735 (MALE, "Male"),
736 (FEMALE, "Female"),
737 (INTERSEX, "Inderminate/intersex"),
738 (UNKNOWNSEX, "Unknown"),
739 )
741 # Details of lookup
742 pt_local_id_description = models.CharField(
743 blank=True,
744 max_length=LEN_FIELD_DESCRIPTION,
745 verbose_name="Description of database-specific ID",
746 )
747 pt_local_id_number = models.BigIntegerField(
748 null=True, blank=True, verbose_name="Database-specific ID"
749 )
750 # Information coming out: patient
751 pt_dob = models.DateField(
752 null=True, blank=True, verbose_name="Patient date of birth"
753 )
754 pt_dod = models.DateField(
755 null=True,
756 blank=True,
757 verbose_name="Patient date of death (NULL if alive)",
758 )
759 pt_dead = models.BooleanField(
760 default=False, verbose_name="Patient is dead"
761 )
762 pt_discharged = models.BooleanField(
763 null=True, verbose_name="Patient discharged"
764 )
765 pt_discharge_date = models.DateField(
766 null=True, blank=True, verbose_name="Patient date of discharge"
767 )
768 pt_sex = models.CharField(
769 max_length=1,
770 blank=True,
771 choices=SEX_CHOICES,
772 verbose_name="Patient sex",
773 )
774 pt_title = models.CharField(
775 max_length=LEN_TITLE, blank=True, verbose_name="Patient title"
776 )
777 pt_first_name = models.CharField(
778 max_length=LEN_NAME, blank=True, verbose_name="Patient first name"
779 )
780 pt_last_name = models.CharField(
781 max_length=LEN_NAME, blank=True, verbose_name="Patient last name"
782 )
783 pt_address_1 = models.CharField(
784 max_length=LEN_ADDRESS,
785 blank=True,
786 verbose_name="Patient address line 1",
787 )
788 pt_address_2 = models.CharField(
789 max_length=LEN_ADDRESS,
790 blank=True,
791 verbose_name="Patient address line 2",
792 )
793 pt_address_3 = models.CharField(
794 max_length=LEN_ADDRESS,
795 blank=True,
796 verbose_name="Patient address line 3",
797 )
798 pt_address_4 = models.CharField(
799 max_length=LEN_ADDRESS,
800 blank=True,
801 verbose_name="Patient address line 4",
802 )
803 pt_address_5 = models.CharField(
804 max_length=LEN_ADDRESS,
805 blank=True,
806 verbose_name="Patient address line 5 (county)",
807 )
808 pt_address_6 = models.CharField(
809 max_length=LEN_ADDRESS,
810 blank=True,
811 verbose_name="Patient address line 6 (postcode)",
812 )
813 pt_address_7 = models.CharField(
814 max_length=LEN_ADDRESS,
815 blank=True,
816 verbose_name="Patient address line 7 (country)",
817 )
818 pt_telephone = models.CharField(
819 max_length=LEN_PHONE, blank=True, verbose_name="Patient telephone"
820 )
821 pt_email = models.EmailField(blank=True, verbose_name="Patient email")
823 # Information coming out: GP
824 gp_title = models.CharField(
825 max_length=LEN_TITLE, blank=True, verbose_name="GP title"
826 )
827 gp_first_name = models.CharField(
828 max_length=LEN_NAME, blank=True, verbose_name="GP first name"
829 )
830 gp_last_name = models.CharField(
831 max_length=LEN_NAME, blank=True, verbose_name="GP last name"
832 )
833 gp_address_1 = models.CharField(
834 max_length=LEN_ADDRESS, blank=True, verbose_name="GP address line 1"
835 )
836 gp_address_2 = models.CharField(
837 max_length=LEN_ADDRESS, blank=True, verbose_name="GP address line 2"
838 )
839 gp_address_3 = models.CharField(
840 max_length=LEN_ADDRESS, blank=True, verbose_name="GP address line 3"
841 )
842 gp_address_4 = models.CharField(
843 max_length=LEN_ADDRESS, blank=True, verbose_name="GP address line 4"
844 )
845 gp_address_5 = models.CharField(
846 max_length=LEN_ADDRESS,
847 blank=True,
848 verbose_name="GP address line 5 (county)",
849 )
850 gp_address_6 = models.CharField(
851 max_length=LEN_ADDRESS,
852 blank=True,
853 verbose_name="GP address line 6 (postcode)",
854 )
855 gp_address_7 = models.CharField(
856 max_length=LEN_ADDRESS,
857 blank=True,
858 verbose_name="GP address line 7 (country)",
859 )
860 gp_telephone = models.CharField(
861 max_length=LEN_PHONE, blank=True, verbose_name="GP telephone"
862 )
863 gp_email = models.EmailField(blank=True, verbose_name="GP email")
865 # Information coming out: clinician
866 clinician_title = models.CharField(
867 max_length=LEN_TITLE, blank=True, verbose_name="Clinician title"
868 )
869 clinician_first_name = models.CharField(
870 max_length=LEN_NAME, blank=True, verbose_name="Clinician first name"
871 )
872 clinician_last_name = models.CharField(
873 max_length=LEN_NAME, blank=True, verbose_name="Clinician last name"
874 )
875 clinician_address_1 = models.CharField(
876 max_length=LEN_ADDRESS,
877 blank=True,
878 verbose_name="Clinician address line 1",
879 )
880 clinician_address_2 = models.CharField(
881 max_length=LEN_ADDRESS,
882 blank=True,
883 verbose_name="Clinician address line 2",
884 )
885 clinician_address_3 = models.CharField(
886 max_length=LEN_ADDRESS,
887 blank=True,
888 verbose_name="Clinician address line 3",
889 )
890 clinician_address_4 = models.CharField(
891 max_length=LEN_ADDRESS,
892 blank=True,
893 verbose_name="Clinician address line 4",
894 )
895 clinician_address_5 = models.CharField(
896 max_length=LEN_ADDRESS,
897 blank=True,
898 verbose_name="Clinician address line 5 (county)",
899 )
900 clinician_address_6 = models.CharField(
901 max_length=LEN_ADDRESS,
902 blank=True,
903 verbose_name="Clinician address line 6 (postcode)",
904 )
905 clinician_address_7 = models.CharField(
906 max_length=LEN_ADDRESS,
907 blank=True,
908 verbose_name="Clinician address line 7 (country)",
909 )
910 clinician_telephone = models.CharField(
911 max_length=LEN_PHONE, blank=True, verbose_name="Clinician telephone"
912 )
913 clinician_email = models.EmailField(
914 blank=True, verbose_name="Clinician email"
915 )
916 clinician_is_consultant = models.BooleanField(
917 default=False, verbose_name="Clinician is a consultant"
918 )
919 clinician_signatory_title = models.CharField(
920 max_length=LEN_NAME,
921 blank=True,
922 verbose_name="Clinician's title for signature "
923 "(e.g. 'Consultant psychiatrist')",
924 )
926 class Meta:
927 abstract = True
929 # Generic title stuff:
931 # -------------------------------------------------------------------------
932 # Patient
933 # -------------------------------------------------------------------------
935 def pt_salutation(self) -> str:
936 """
937 Returns a salutation for the patient, like "Mrs Smith".
938 """
939 # noinspection PyTypeChecker
940 return salutation(
941 self.pt_title,
942 self.pt_first_name,
943 self.pt_last_name,
944 sex=self.pt_sex,
945 )
947 def pt_title_forename_surname(self) -> str:
948 """
949 Returns the patient's title/forename/surname, like "Mrs Ann Smith".
950 """
951 # noinspection PyTypeChecker
952 return title_forename_surname(
953 self.pt_title, self.pt_first_name, self.pt_last_name
954 )
956 def pt_forename_surname(self) -> str:
957 """
958 Returns the patient's forename/surname, like "Ann Smith".
959 """
960 # noinspection PyTypeChecker
961 return forename_surname(self.pt_first_name, self.pt_last_name)
963 def pt_address_components(self) -> List[str]:
964 """
965 Returns lines of the patient's address (e.g. for letter headings or
966 envelopes).
967 """
968 return list(
969 filter(
970 None,
971 [
972 self.pt_address_1,
973 self.pt_address_2,
974 self.pt_address_3,
975 self.pt_address_4,
976 self.pt_address_5,
977 self.pt_address_6,
978 self.pt_address_7,
979 ],
980 )
981 )
983 def pt_address_components_str(self) -> str:
984 """
985 Returns the patient's address as a single (one-line) string.
986 """
987 return ", ".join(filter(None, self.pt_address_components()))
989 def pt_name_address_components(self) -> List[str]:
990 """
991 Returns the patient's name and address, as lines (e.g. for an
992 envelope).
993 """
994 return [
995 self.pt_title_forename_surname()
996 ] + self.pt_address_components()
998 def get_id_numbers_as_str(self) -> str:
999 """
1000 Returns ID numbers, in a format like "NHS#: 123. RiO# 456."
1001 """
1002 # Note that self.nhs_number must be implemented by derived classes:
1003 # noinspection PyUnresolvedReferences
1004 idnums = [f"NHS#: {self.nhs_number}"]
1005 if self.pt_local_id_description:
1006 idnums.append(
1007 f"{self.pt_local_id_description}: {self.pt_local_id_number}"
1008 )
1009 return ". ".join(idnums)
1011 def get_pt_age_years(self) -> Optional[int]:
1012 """
1013 Returns the patient's current age in years, or ``None`` if unknown.
1014 """
1015 if self.pt_dob is None:
1016 return None
1017 now = datetime.datetime.now() # timezone-naive
1018 # now = timezone.now() # timezone-aware
1019 return relativedelta(now, self.pt_dob).years
1021 def is_under_16(self) -> bool:
1022 """
1023 Is the patient under 16?
1024 """
1025 age = self.get_pt_age_years()
1026 return age is not None and age < 16
1028 def is_under_15(self) -> bool:
1029 """
1030 Is the patient under 15?
1031 """
1032 age = self.get_pt_age_years()
1033 return age is not None and age < 15
1035 def days_since_discharge(self) -> Optional[int]:
1036 """
1037 Returns the number of days since discharge, or ``None`` if the patient
1038 is not discharged (or if we don't know).
1039 """
1040 if not self.pt_discharged or not self.pt_discharge_date:
1041 return None
1042 try:
1043 today = datetime.date.today()
1044 discharged = self.pt_discharge_date # type: datetime.date
1045 diff = today - discharged
1046 return diff.days
1047 except (AttributeError, TypeError, ValueError):
1048 return None
1050 # @property
1051 # def nhs_number(self) -> int:
1052 # raise NotImplementedError()
1053 #
1054 # ... NO; do not do this; it makes nhs_number a read-only attribute, so
1055 # derived class creation fails with
1056 #
1057 # AttributeError: can't set attribute
1058 #
1059 # when trying to write nhs_number
1061 # -------------------------------------------------------------------------
1062 # GP
1063 # -------------------------------------------------------------------------
1065 def gp_title_forename_surname(self) -> str:
1066 """
1067 Returns the title/forename/surname for the patient's GP, like
1068 "Dr Joe Bloggs".
1069 """
1070 return title_forename_surname(
1071 self.gp_title,
1072 self.gp_first_name,
1073 self.gp_last_name,
1074 always_title=True,
1075 assume_dr=True,
1076 )
1078 def gp_address_components(self) -> List[str]:
1079 """
1080 Returns address lines for the GP (e.g. for an envelope).
1081 """
1082 return list(
1083 filter(
1084 None,
1085 [
1086 self.gp_address_1,
1087 self.gp_address_2,
1088 self.gp_address_3,
1089 self.gp_address_4,
1090 self.gp_address_5,
1091 self.gp_address_6,
1092 self.gp_address_7,
1093 ],
1094 )
1095 )
1097 def gp_address_components_str(self) -> str:
1098 """
1099 Returns the GP's address as a single line.
1100 """
1101 return ", ".join(self.gp_address_components())
1103 def gp_name_address_str(self) -> str:
1104 """
1105 Returns the GP's name and address as a single line.
1106 """
1107 return ", ".join(
1108 filter(
1109 None,
1110 [
1111 self.gp_title_forename_surname(),
1112 self.gp_address_components_str(),
1113 ],
1114 )
1115 )
1117 # noinspection PyUnusedLocal
1118 def set_gp_name_components(
1119 self, name: str, decisions: List[str], secret_decisions: List[str]
1120 ) -> None:
1121 """
1122 Takes a name, splits it into components as best it can, and stores it
1123 in the ``gp_title``, ``gp_first_name``, and ``gp_last_name`` fields.
1125 Args:
1126 name: GP name, e.g. "Dr Joe Bloggs"
1127 decisions: list of human-readable decisions; will be modified
1128 secret_decisions: list of human-readable decisions containing
1129 secret (identifiable) information; will be modified
1130 """
1131 secret_decisions.append(f"Setting GP name components from: {name}.")
1132 self.gp_title = ""
1133 self.gp_first_name = ""
1134 self.gp_last_name = ""
1135 if name == "No Registered GP" or not name:
1136 self.gp_last_name = "[No registered GP]"
1137 return
1138 if "(" in name:
1139 # A very odd thing like "LINTON H C (PL)"
1140 self.gp_last_name = name
1141 return
1142 initial, surname = get_initial_surname_tuple_from_string(name)
1143 initial = initial.title() # title case
1144 surname = surname.title() # title case
1145 self.gp_title = "Dr"
1146 self.gp_first_name = initial + ("." if initial else "")
1147 self.gp_last_name = surname
1149 # -------------------------------------------------------------------------
1150 # Clinician
1151 # -------------------------------------------------------------------------
1153 def clinician_salutation(self) -> str:
1154 """
1155 Returns the salutation for the patient's clinician (e.g. "Dr
1156 Paroxetine").
1157 """
1158 # noinspection PyTypeChecker
1159 return salutation(
1160 self.clinician_title,
1161 self.clinician_first_name,
1162 self.clinician_last_name,
1163 assume_dr=True,
1164 )
1166 def clinician_title_forename_surname(self) -> str:
1167 """
1168 Returns the title/forename/surname for the patient's clinician (e.g.
1169 "Dr Petra Paroxetine").
1170 """
1171 # noinspection PyTypeChecker
1172 return title_forename_surname(
1173 self.clinician_title,
1174 self.clinician_first_name,
1175 self.clinician_last_name,
1176 )
1178 def clinician_address_components(self) -> List[str]:
1179 """
1180 Returns the clinician's address -- or the Research Database Manager's
1181 (with "c/o") if we don't know the clinician's.
1183 (We're going to put the clinician's postal address into letters to
1184 patients. Therefore, we need a sensible fallback, i.e. the RDBM's.)
1185 """
1186 address_components = [
1187 self.clinician_address_1,
1188 self.clinician_address_2,
1189 self.clinician_address_3,
1190 self.clinician_address_4,
1191 self.clinician_address_5,
1192 self.clinician_address_6,
1193 self.clinician_address_7,
1194 ]
1195 if not any(x for x in address_components):
1196 address_components = settings.RDBM_ADDRESS.copy()
1197 if address_components:
1198 address_components[0] = "c/o " + address_components[0]
1199 return list(filter(None, address_components))
1201 def clinician_address_components_str(self) -> str:
1202 """
1203 Returns the clinician's address in single-line format.
1204 """
1205 return ", ".join(self.clinician_address_components())
1207 def clinician_name_address_str(self) -> str:
1208 """
1209 Returns the clinician's name and address in single-line format.
1210 """
1211 return ", ".join(
1212 filter(
1213 None,
1214 [
1215 self.clinician_title_forename_surname(),
1216 self.clinician_address_components_str(),
1217 ],
1218 )
1219 )
1221 # -------------------------------------------------------------------------
1222 # Paperwork
1223 # -------------------------------------------------------------------------
1225 def get_traffic_light_decision_form(self, generic: bool = False) -> str:
1226 """
1227 Returns HTML for a traffic-light decision form, customized to this
1228 patient.
1229 """
1230 context = {
1231 "patient_lookup": self,
1232 "generic": generic,
1233 "settings": settings,
1234 }
1235 return render_pdf_html_to_string(
1236 "traffic_light_decision_form.html", context, patient=True
1237 )
1240class DummyPatientSourceInfo(PatientLookupBase):
1241 """
1242 A patient lookup class that is a dummy one, for testing.
1243 """
1245 # Key
1246 nhs_number = models.BigIntegerField(verbose_name="NHS number", unique=True)
1248 class Meta:
1249 verbose_name_plural = "Dummy patient source information"
1251 def __str__(self) -> str:
1252 return (
1253 f"[DummyPatientSourceInfo {self.id}] "
1254 f"Dummy patient lookup for NHS# {self.nhs_number}"
1255 )
1258class PatientLookup(PatientLookupBase):
1259 """
1260 Represents a moment of lookup up identifiable data about patient, GP,
1261 and clinician from the relevant clinical database.
1263 Inherits from :class:`PatientLookupBase` so it has the same fields, and
1264 more.
1265 """
1267 nhs_number = models.BigIntegerField(
1268 verbose_name="NHS number used for lookup"
1269 )
1270 lookup_at = models.DateTimeField(
1271 verbose_name="When fetched from clinical database", auto_now_add=True
1272 )
1274 # Information going in
1275 source_db = models.CharField(
1276 max_length=SOURCE_DB_NAME_MAX_LENGTH,
1277 choices=ClinicalDatabaseType.DATABASE_CHOICES,
1278 verbose_name="Source database used for lookup",
1279 )
1281 # Information coming out: general
1282 decisions = models.TextField(
1283 blank=True, verbose_name="Decisions made during lookup"
1284 )
1285 secret_decisions = models.TextField(
1286 blank=True,
1287 verbose_name="Secret (identifying) decisions made during lookup",
1288 )
1290 # Information coming out: patient
1291 pt_found = models.BooleanField(default=False, verbose_name="Patient found")
1293 # Information coming out: GP
1294 gp_found = models.BooleanField(default=False, verbose_name="GP found")
1296 # Information coming out: clinician
1297 clinician_found = models.BooleanField(
1298 default=False, verbose_name="Clinician found"
1299 )
1301 def __repr__(self) -> str:
1302 return modelrepr(self)
1304 def __str__(self) -> str:
1305 return f"[PatientLookup {self.id}] NHS# {self.nhs_number}"
1307 def get_first_traffic_light_letter_html(self) -> str:
1308 """
1309 Covering letter to patient for first enquiry about research preference.
1311 Returns HTML for this document, customized to the patient.
1312 """
1313 context = {
1314 # Letter bits
1315 "address_from": self.clinician_address_components(),
1316 "address_to": self.pt_name_address_components(),
1317 "salutation": self.pt_salutation(),
1318 "signatory_name": self.clinician_title_forename_surname(),
1319 "signatory_title": self.clinician_signatory_title,
1320 # Specific bits
1321 "settings": settings,
1322 "patient_lookup": self,
1323 }
1324 return render_pdf_html_to_string(
1325 "letter_patient_first_traffic_light.html", context, patient=True
1326 )
1328 def set_from_clinician_info_holder(
1329 self, info: ClinicianInfoHolder
1330 ) -> None:
1331 """
1332 Sets the clinician information fields from the supplied
1333 :class:`ClinicianInfoHolder`.
1334 """
1335 self.clinician_found = True
1336 self.clinician_title = info.title
1337 self.clinician_first_name = info.first_name
1338 self.clinician_last_name = info.surname
1339 self.clinician_email = info.email
1340 self.clinician_is_consultant = info.is_consultant
1341 self.clinician_signatory_title = info.signatory_title
1342 # Slice notation returns an empty list, rather than an exception,
1343 # if the index is out of range
1344 self.clinician_address_1 = info.address_components[0:1] or ""
1345 self.clinician_address_2 = info.address_components[1:2] or ""
1346 self.clinician_address_3 = info.address_components[2:3] or ""
1347 self.clinician_address_4 = info.address_components[3:4] or ""
1348 self.clinician_address_5 = info.address_components[4:5] or ""
1349 self.clinician_address_6 = info.address_components[5:6] or ""
1350 self.clinician_address_7 = info.address_components[6:7] or ""
1353# =============================================================================
1354# Clinical team representative
1355# =============================================================================
1358class TeamInfo:
1359 """
1360 Represents information about all clinical teams, fetched from a clinical
1361 source database.
1363 Provides some simple views on
1364 :func:`crate_anon.crateweb.consent.teamlookup.get_teams`.
1365 """
1367 @staticmethod
1368 def teams() -> List[str]:
1369 """
1370 Returns all clinical team names.
1371 """
1372 return get_teams() # cached function
1374 @classmethod
1375 def team_choices(cls) -> List[Tuple[str, str]]:
1376 """
1377 Returns a Django choice list, i.e. a list of tuples like ``value,
1378 description``.
1379 """
1380 teams = cls.teams()
1381 return [(team, team) for team in teams]
1384class TeamRep(models.Model):
1385 """
1386 Represents a clinical team representative, which is recorded in CRATE.
1387 """
1389 team = models.CharField(
1390 max_length=LEN_NAME, unique=True, verbose_name="Team description"
1391 )
1392 user = models.ForeignKey(
1393 settings.AUTH_USER_MODEL, on_delete=models.CASCADE
1394 )
1396 class Meta:
1397 verbose_name = "clinical team representative"
1398 verbose_name_plural = "clinical team representatives"
1401# =============================================================================
1402# Record of payments to charity
1403# =============================================================================
1404# In passing - singleton objects:
1405# http://goodcode.io/articles/django-singleton-models/
1408class CharityPaymentRecord(models.Model):
1409 """
1410 A record of a payment made to charity.
1411 """
1413 created_at = models.DateTimeField(
1414 verbose_name="When created", auto_now_add=True
1415 )
1416 payee = models.CharField(max_length=255)
1417 amount = models.DecimalField(max_digits=8, decimal_places=2)
1420# =============================================================================
1421# Record of consent mode for a patient
1422# =============================================================================
1425class ConsentMode(Decision):
1426 """
1427 Represents a consent-to-contact consent mode for a patient.
1428 """
1430 RED = "red"
1431 YELLOW = "yellow"
1432 GREEN = "green"
1434 VALID_CONSENT_MODES = [RED, YELLOW, GREEN]
1435 CONSENT_MODE_CHOICES = (
1436 (RED, "red"),
1437 (YELLOW, "yellow"),
1438 (GREEN, "green"),
1439 )
1440 # ... https://stackoverflow.com/questions/12822847/best-practice-for-python-django-constants # noqa: E501
1442 SOURCE_USER_ENTRY = "crate_user_entry"
1443 SOURCE_AUTOCREATED = "crate_auto_created"
1444 SOURCE_LEGACY = "legacy" # default, for old versions
1446 nhs_number = models.BigIntegerField(verbose_name="NHS number")
1447 current = models.BooleanField(default=False)
1448 # see save() and process_change() below
1449 created_at = models.DateTimeField(
1450 verbose_name="When was this record created?", auto_now_add=True
1451 )
1452 created_by = models.ForeignKey(
1453 settings.AUTH_USER_MODEL, on_delete=models.PROTECT
1454 )
1456 exclude_entirely = models.BooleanField(
1457 default=False,
1458 verbose_name="Exclude patient from Research Database entirely?",
1459 )
1460 consent_mode = models.CharField(
1461 max_length=10,
1462 default="",
1463 choices=CONSENT_MODE_CHOICES,
1464 verbose_name="Consent mode (red/yellow/green)",
1465 )
1466 consent_after_discharge = models.BooleanField(
1467 default=False,
1468 verbose_name="Consent given to contact patient after discharge?",
1469 )
1470 max_approaches_per_year = models.PositiveSmallIntegerField(
1471 verbose_name="Maximum number of approaches permissible per year "
1472 "(0 = no limit)",
1473 default=0,
1474 )
1475 other_requests = models.TextField(
1476 blank=True, verbose_name="Other special requests by patient"
1477 )
1478 prefers_email = models.BooleanField(
1479 default=False, verbose_name="Patient prefers e-mail contact?"
1480 )
1481 changed_by_clinician_override = models.BooleanField(
1482 default=False,
1483 verbose_name="Consent mode changed by clinician's override?",
1484 )
1486 source = models.CharField(
1487 max_length=SOURCE_DB_NAME_MAX_LENGTH,
1488 default=SOURCE_USER_ENTRY,
1489 verbose_name="Source of information",
1490 )
1492 skip_letter_to_patient = models.BooleanField(
1493 default=False
1494 ) # added 2018-06-29
1495 needs_processing = models.BooleanField(default=False) # added 2018-06-29
1496 processed = models.BooleanField(default=False) # added 2018-06-29
1497 processed_at = models.DateTimeField(null=True) # added 2018-06-29
1499 # class Meta:
1500 # get_latest_by = "created_at"
1502 def save(self, *args, **kwargs) -> None:
1503 """
1504 Custom save method. Ensures that only one :class:`ConsentMode` has
1505 ``current == True`` for a given patient.
1507 This is better than a ``get_latest_by`` clause, because with a flag
1508 like this, we can have a simple query that says "get the current
1509 records for all patients" -- which is harder if done by date (group by
1510 patient, order by patient/date, pick last one for each patient...).
1512 See
1513 https://stackoverflow.com/questions/1455126/unique-booleanfield-value-in-django
1514 """
1515 if self.current:
1516 ConsentMode.objects.filter(
1517 nhs_number=self.nhs_number, current=True
1518 ).update(current=False)
1519 super().save(*args, **kwargs)
1521 def __str__(self) -> str:
1522 return (
1523 f"[ConsentMode {self.id}] "
1524 f"NHS# {self.nhs_number}, {self.consent_mode}"
1525 )
1527 @classmethod
1528 def get_or_create(
1529 cls, nhs_number: int, created_by: settings.AUTH_USER_MODEL
1530 ) -> CONSENT_MODE_FWD_REF:
1531 """
1532 Fetches the current :class:`ConsentMode` for this patient.
1533 If there isn't one, creates a default one and returns that.
1534 """
1535 try:
1536 consent_mode = cls.objects.get(nhs_number=nhs_number, current=True)
1537 except cls.DoesNotExist:
1538 consent_mode = cls(
1539 nhs_number=nhs_number,
1540 created_by=created_by,
1541 source=cls.SOURCE_AUTOCREATED,
1542 needs_processing=False,
1543 current=True,
1544 )
1545 consent_mode.save()
1546 except cls.MultipleObjectsReturned:
1547 log.warning(
1548 "bug: ConsentMode.get_or_create() received "
1549 "exception ConsentMode.MultipleObjectsReturned"
1550 )
1551 consent_mode = cls(
1552 nhs_number=nhs_number,
1553 created_by=created_by,
1554 source=cls.SOURCE_AUTOCREATED,
1555 needs_processing=False,
1556 current=True,
1557 )
1558 consent_mode.save()
1559 return consent_mode
1561 @classmethod
1562 def get_or_none(cls, nhs_number: int) -> Optional[CONSENT_MODE_FWD_REF]:
1563 """
1564 Fetches the current :class:`ConsentMode` for this patient.
1565 If there isn't one, returns ``None``.
1566 """
1567 try:
1568 return cls.objects.get(nhs_number=nhs_number, current=True)
1569 except cls.DoesNotExist:
1570 return None
1572 @classmethod
1573 def refresh_from_primary_clinical_record(
1574 cls,
1575 nhs_number: int,
1576 created_by: settings.AUTH_USER_MODEL,
1577 source_db: str = None,
1578 ) -> List[str]:
1579 """
1580 Checks the primary clinical record and CRATE's own records for consent
1581 modes for this patient. If the most recent one is in the external
1582 database, copies it to CRATE's database and marks that one as current.
1584 This has the effect that external primary clinical records (e.g. RiO)
1585 take priority, but if there's no record in RiO, we can still proceed.
1587 Returns a list of human-readable decisions.
1589 Internally, we do this:
1591 - Fetch the most recent record.
1592 - If its date is later than the most recent CRATE record:
1594 - create a new ConsentMode with (..., source=source_db)
1595 - save it
1596 - trigger
1597 :func:`crate_anon.crateweb.consent.tasks.process_consent_change` on
1598 commit, which might take further action
1600 """
1601 from crate_anon.crateweb.consent.lookup import (
1602 lookup_consent,
1603 ) # delayed import
1605 decisions = [] # type: List[str]
1606 source_db = source_db or settings.CLINICAL_LOOKUP_CONSENT_DB
1607 decisions.append(f"source_db = {source_db}")
1609 latest = lookup_consent(
1610 nhs_number=nhs_number, source_db=source_db, decisions=decisions
1611 )
1612 if latest is None:
1613 decisions.append(
1614 "No consent decision found in primary clinical " "record"
1615 )
1616 return decisions
1618 crate_version = cls.get_or_none(nhs_number=nhs_number)
1620 if crate_version and crate_version.created_at >= latest.created_at:
1621 decisions.append(
1622 f"CRATE stored version is at least as recent "
1623 f"({crate_version.created_at}) as the version from the "
1624 f"clinical record ({latest.created_at}); ignoring"
1625 )
1626 return decisions
1628 # If we get here, we've found a newer version in the clinical record.
1629 latest.created_by = created_by
1630 latest.source = source_db
1631 latest.current = True
1632 latest.needs_processing = True
1633 latest.skip_letter_to_patient = True # the patient already knows;
1634 # they made the decision with the clinician who entered this into the
1635 # primary clinical record.
1636 latest.save() # This now becomes the current CRATE consent mode.
1637 transaction.on_commit(
1638 lambda: process_consent_change.delay(latest.id)
1639 ) # Asynchronous
1640 # Without transaction.on_commit, we get a RACE CONDITION:
1641 # object is received in the pre-save() state.
1642 return decisions
1644 def consider_withdrawal(self) -> None:
1645 """
1646 If required, withdraw consent for other studies.
1648 Note that as per Major Amendment 1 to 12/EE/0407, this happens
1649 automatically, rather than having a special flag to control it.
1650 """
1651 try:
1652 previous = ConsentMode.objects.filter(
1653 nhs_number=self.nhs_number,
1654 current=False,
1655 created_at__isnull=False,
1656 ).latest("created_at")
1657 # ... https://docs.djangoproject.com/en/dev/ref/models/querysets/#latest # noqa: E501
1658 if not previous:
1659 return # no previous ConsentMode; nothing to do
1660 if (
1661 previous.consent_mode == ConsentMode.GREEN
1662 and self.consent_mode != ConsentMode.GREEN
1663 ):
1664 contact_requests = (
1665 ContactRequest.objects.filter(nhs_number=self.nhs_number)
1666 .filter(consent_mode__consent_mode=ConsentMode.GREEN)
1667 .filter(decided_send_to_researcher=True)
1668 .filter(consent_withdrawn=False)
1669 )
1670 for contact_request in contact_requests:
1671 (
1672 letter,
1673 email_succeeded,
1674 ) = contact_request.withdraw_consent()
1675 if not email_succeeded:
1676 self.notify_rdbm_of_work(letter, to_researcher=True)
1677 except ConsentMode.DoesNotExist:
1678 pass # no previous ConsentMode; nothing to do.
1679 except ConsentMode.MultipleObjectsReturned:
1680 log.warning(
1681 "bug: ConsentMode.consider_withdrawal() received "
1682 "exception ConsentMode.MultipleObjectsReturned"
1683 )
1684 # do nothing else
1686 def get_latest_patient_lookup(self) -> PatientLookup:
1687 """
1688 Returns the latest :class:`PatientLookup` information (from the CRATE
1689 admin database) for this patient.
1690 """
1691 from crate_anon.crateweb.consent.lookup import (
1692 lookup_patient,
1693 ) # delayed import
1695 # noinspection PyTypeChecker
1696 return lookup_patient(self.nhs_number, existing_ok=True)
1698 def get_confirm_traffic_to_patient_letter_html(
1699 self, patient_lookup_override: PatientLookup = None
1700 ) -> str:
1701 """
1702 Letter to patient, confirming traffic-light choice.
1704 Returns HTML for this letter, customized to the patient.
1705 """
1706 patient_lookup = (
1707 patient_lookup_override or self.get_latest_patient_lookup()
1708 )
1709 context = {
1710 # Letter bits
1711 "address_from": settings.RDBM_ADDRESS + [settings.RDBM_EMAIL],
1712 "address_to": patient_lookup.pt_name_address_components(),
1713 "salutation": patient_lookup.pt_salutation(),
1714 "signatory_name": settings.RDBM_NAME,
1715 "signatory_title": settings.RDBM_TITLE,
1716 # Specific bits
1717 "consent_mode": self,
1718 "patient_lookup": patient_lookup,
1719 "settings": settings,
1720 # URLs
1721 # 'red_img_url': site_absolute_url(static('red.png')),
1722 # 'yellow_img_url': site_absolute_url(static('yellow.png')),
1723 # 'green_img_url': site_absolute_url(static('green.png')),
1724 }
1725 # 1. Building a static URL in code:
1726 # https://stackoverflow.com/questions/11721818/django-get-the-static-files-url-in-view # noqa: E501
1727 # 2. Making it an absolute URL means that wkhtmltopdf will also see it
1728 # (by fetching it from this web server).
1729 # 3. Works with Django testing server.
1730 # 4. Works with Apache, + proxying to backend, + SSL
1731 return render_pdf_html_to_string(
1732 "letter_patient_confirm_traffic.html", context, patient=True
1733 )
1735 def notify_rdbm_of_work(
1736 self, letter: LETTER_FWD_REF, to_researcher: bool = False
1737 ) -> None:
1738 """
1739 E-mail the RDBM saying that there's new work to do: a letter to be
1740 sent.
1742 Args:
1743 letter: :class:`Letter`
1744 to_researcher: is it a letter that needs to go to a researcher,
1745 rather than to a patient?
1746 """
1747 subject = (
1748 f"WORK FROM RESEARCH DATABASE COMPUTER - consent mode {self.id}"
1749 )
1750 if to_researcher:
1751 template = "email_rdbm_new_work_researcher.html"
1752 else:
1753 template = "email_rdbm_new_work_pt_from_rdbm.html"
1754 html = render_email_html_to_string(template, {"letter": letter})
1755 email = Email.create_rdbm_email(subject, html)
1756 email.send()
1758 @staticmethod
1759 def get_unprocessed() -> QuerySet:
1760 """
1761 Return all :class:`ConsentMode` objects that need processing.
1763 See :func:`crate_anon.crateweb.consent.tasks.process_consent_change`
1764 and :func:`process_change`, which does the work.
1765 """
1766 return ConsentMode.objects.filter(
1767 needs_processing=True,
1768 current=True,
1769 processed=False,
1770 )
1772 def process_change(self) -> None:
1773 """
1774 Called upon saving.
1776 - Will create a letter to patient.
1777 - May create a withdrawal-of-consent letter to researcher.
1778 - Marks the :class:`ConsentMode` as having been processed.
1780 **Major Amendment 1 (Oct 2014) to 12/EE/0407:** always withdraw consent
1781 and tell researchers, i.e. "active cancellation" of ongoing permission,
1782 where the researchers have not yet made contact.
1783 """
1784 if self.processed:
1785 log.warning(
1786 f"ConsentMode #{self.id}: already processed; "
1787 f"not processing again"
1788 )
1789 return
1790 if not self.needs_processing:
1791 return
1792 if not self.current:
1793 # No point processing non-current things.
1794 return
1796 if not self.skip_letter_to_patient:
1797 # noinspection PyTypeChecker
1798 letter = Letter.create_consent_confirmation_to_patient(self)
1799 # ... will save
1800 self.notify_rdbm_of_work(letter, to_researcher=False)
1802 self.consider_withdrawal()
1804 self.processed = True
1805 self.needs_processing = False
1806 self.processed_at = timezone.now()
1807 self.save()
1810# =============================================================================
1811# Request for patient contact
1812# =============================================================================
1815class ContactRequest(models.Model):
1816 """
1817 Represents a contact request to a patient (directly or indirectly) about a
1818 study.
1819 """
1821 CLINICIAN_INVOLVEMENT_NONE = 0
1822 CLINICIAN_INVOLVEMENT_REQUESTED = 1
1823 CLINICIAN_INVOLVEMENT_REQUIRED_YELLOW = 2
1824 CLINICIAN_INVOLVEMENT_REQUIRED_UNKNOWN = 3
1826 CLINICIAN_CONTACT_MODE_CHOICES = (
1827 (
1828 CLINICIAN_INVOLVEMENT_NONE,
1829 "No clinician involvement required or requested",
1830 ),
1831 (
1832 CLINICIAN_INVOLVEMENT_REQUESTED,
1833 "Clinician involvement requested by researchers",
1834 ),
1835 (
1836 CLINICIAN_INVOLVEMENT_REQUIRED_YELLOW,
1837 "Clinician involvement required by YELLOW consent mode",
1838 ),
1839 (
1840 CLINICIAN_INVOLVEMENT_REQUIRED_UNKNOWN,
1841 "Clinician involvement required by UNKNOWN consent mode",
1842 ),
1843 )
1845 # Created initially:
1846 created_at = models.DateTimeField(
1847 verbose_name="When created", auto_now_add=True
1848 )
1849 request_by = models.ForeignKey(
1850 settings.AUTH_USER_MODEL, on_delete=models.PROTECT
1851 )
1852 study = models.ForeignKey(Study, on_delete=models.PROTECT) # type: Study
1853 request_direct_approach = models.BooleanField(
1854 verbose_name="Request direct contact with patient if available"
1855 " (not contact with clinician first)"
1856 )
1857 # One of these will be non-NULL
1858 lookup_nhs_number = models.BigIntegerField(
1859 null=True, verbose_name="NHS number used for lookup"
1860 )
1861 lookup_rid = models.CharField(
1862 max_length=MAX_HASH_LENGTH,
1863 null=True,
1864 verbose_name="Research ID used for lookup",
1865 )
1866 lookup_mrid = models.CharField(
1867 max_length=MAX_HASH_LENGTH,
1868 null=True,
1869 verbose_name="Master research ID used for lookup",
1870 )
1872 processed = models.BooleanField(default=False)
1873 processed_at = models.DateTimeField(null=True) # added 2018-06-29
1874 # Below: created during processing.
1876 # Those numbers translate to this:
1877 nhs_number = models.BigIntegerField(null=True, verbose_name="NHS number")
1878 # ... from which:
1879 patient_lookup = models.ForeignKey(
1880 PatientLookup, on_delete=models.SET_NULL, null=True
1881 )
1882 consent_mode = models.ForeignKey(
1883 ConsentMode, on_delete=models.SET_NULL, null=True
1884 )
1885 # Now decisions:
1886 approaches_in_past_year = models.PositiveIntegerField(null=True)
1887 decisions = models.TextField(blank=True, verbose_name="Decisions made")
1888 decided_no_action = models.BooleanField(default=False)
1889 decided_send_to_researcher = models.BooleanField(default=False)
1890 decided_send_to_clinician = models.BooleanField(default=False)
1891 clinician_involvement = models.PositiveSmallIntegerField(
1892 choices=CLINICIAN_CONTACT_MODE_CHOICES, null=True
1893 )
1894 consent_withdrawn = models.BooleanField(default=False)
1895 consent_withdrawn_at = models.DateTimeField(
1896 verbose_name="When consent withdrawn", null=True
1897 )
1898 clinician_initiated = models.BooleanField(default=False)
1899 clinician_email = models.TextField(null=True, default=None)
1900 # Specifically for clinician-initiated case:
1901 rdbm_to_contact_pt = models.BooleanField(default=False)
1902 # Should be in form 'title firstname lastname'
1903 clinician_signatory_name = models.TextField(null=True, default=None)
1904 clinician_signatory_title = models.TextField(null=True, default=None)
1906 def __init__(self, *args: Any, **kwargs: Any) -> None:
1907 super().__init__(*args, **kwargs)
1908 self.decisionlist = [] # type: List[str]
1910 def __str__(self) -> str:
1911 return f"[ContactRequest {self.id}] Study {self.study_id}"
1913 def __repr__(self) -> str:
1914 return modelrepr(self)
1916 @classmethod
1917 def create(
1918 cls,
1919 request: HttpRequest,
1920 study: Study,
1921 request_direct_approach: bool,
1922 lookup_nhs_number: int = None,
1923 lookup_rid: str = None,
1924 lookup_mrid: str = None,
1925 clinician_initiated: bool = False,
1926 clinician_email: str = None,
1927 rdbm_to_contact_pt: bool = False,
1928 clinician_signatory_name: Optional[str] = None,
1929 clinician_signatory_title: Optional[str] = None,
1930 ) -> CONTACT_REQUEST_FWD_REF:
1931 """
1932 Create a contact request and act on it.
1934 Args:
1935 request: the :class:`django.http.request.HttpRequest`
1936 study: the :class:`Study`
1937 request_direct_approach: would the researchers prefer to approach
1938 the patient directly, if permitted?
1939 lookup_nhs_number: NHS number to look up patient from
1940 lookup_rid: research ID (RID) to look up patient from
1941 lookup_mrid: master research ID (MRID) to look up patient from
1942 clinician_initiated: contact request initiated by the clinician?
1943 clinician_email: override the clinician email in patient_lookup
1944 rdbm_to_contact_pt: should the rbdm contact the patient - for cases
1945 where the request was initiated by clinician
1946 clinician_signatory_name: name of clinician for letter - if None
1947 will use PatientLookup
1948 clinician_signatory_title: signatory title of clinician - if None
1949 will use PatientLookup
1951 Returns:
1952 a :class:`ContactRequest`
1953 """
1954 # https://docs.djangoproject.com/en/1.9/ref/request-response/
1955 # noinspection PyTypeChecker
1956 cr = cls(
1957 request_by=request.user,
1958 study=study,
1959 request_direct_approach=request_direct_approach,
1960 lookup_nhs_number=lookup_nhs_number,
1961 lookup_rid=lookup_rid,
1962 lookup_mrid=lookup_mrid,
1963 clinician_initiated=clinician_initiated,
1964 clinician_email=clinician_email,
1965 rdbm_to_contact_pt=rdbm_to_contact_pt,
1966 clinician_signatory_name=clinician_signatory_name,
1967 clinician_signatory_title=clinician_signatory_title,
1968 )
1969 cr.save()
1970 transaction.on_commit(
1971 lambda: process_contact_request.delay(cr.id)
1972 ) # Asynchronous
1973 return cr
1975 @staticmethod
1976 def get_unprocessed() -> QuerySet:
1977 """
1978 Return all :class:`ContactRequest` objects that need processing.
1980 See :func:`crate_anon.crateweb.consent.tasks.process_contact_request`
1981 and :func:`process_request`, which does the work.
1982 """
1983 return ContactRequest.objects.filter(processed=False)
1985 def process_request(self) -> None:
1986 """
1987 Processes the :class:`ContactRequest` and marks it as processed. The
1988 main work is done by :func:`process_request_main`.
1989 """
1990 if self.processed:
1991 log.warning(
1992 f"ContactRequest #{self.id}: already processed; "
1993 f"not processing again"
1994 )
1995 return
1996 self.decisionlist = [] # type: List[str]
1997 self.process_request_main()
1998 self.decisions = " ".join(self.decisionlist)
1999 self.processed = True
2000 self.processed_at = timezone.now()
2001 self.save()
2003 def mockup(self):
2004 """
2005 Used to ensure test objects are OK.
2006 """
2007 self.store_clinician_details()
2009 def store_clinician_details(self) -> None:
2010 """
2011 Ensure that if we have not got "override" details for the clinician,
2012 that we copy them from the patient lookup.
2013 """
2014 # We may need to input clinician email manually, otherwise use default
2015 if not self.patient_lookup:
2016 return
2017 if not self.clinician_email:
2018 self.clinician_email = self.patient_lookup.clinician_email
2019 if not self.clinician_signatory_name:
2020 self.clinician_signatory_name = (
2021 self.patient_lookup.clinician_title_forename_surname()
2022 )
2023 if not self.clinician_signatory_title:
2024 self.clinician_signatory_title = (
2025 self.patient_lookup.clinician_signatory_title
2026 )
2028 def process_request_main(self) -> None:
2029 """
2030 Act on a contact request and store the decisions made.
2032 **CORE DECISION-MAKING FUNCTION FOR THE CONSENT-TO-CONTACT PROCESS.**
2034 Writes to :attr:`decisionlist`.
2035 """
2036 from crate_anon.crateweb.consent.lookup import (
2037 lookup_patient,
2038 ) # delayed import
2040 # Translate to an NHS number
2041 research_database_info = get_research_db_info()
2042 dbinfo = research_database_info.dbinfo_for_contact_lookup
2043 if self.lookup_nhs_number is not None:
2044 self.nhs_number = self.lookup_nhs_number
2045 elif self.lookup_rid is not None:
2046 self.nhs_number = get_mpid(dbinfo=dbinfo, rid=self.lookup_rid)
2047 elif self.lookup_mrid is not None:
2048 self.nhs_number = get_mpid(dbinfo=dbinfo, mrid=self.lookup_mrid)
2049 else:
2050 raise ValueError("No NHS number, RID, or MRID supplied.")
2051 # Look up patient details (afresh)
2052 self.patient_lookup = lookup_patient(self.nhs_number, save=True)
2053 # Ensure clinician details are OK
2054 self.store_clinician_details()
2055 # Establish consent mode (always do this to avoid NULL problem)
2056 ConsentMode.refresh_from_primary_clinical_record(
2057 nhs_number=self.nhs_number, created_by=self.request_by
2058 )
2059 self.consent_mode = ConsentMode.get_or_create(
2060 nhs_number=self.nhs_number, created_by=self.request_by
2061 )
2062 # Rest of processing
2063 self.calc_approaches_in_past_year()
2065 # ---------------------------------------------------------------------
2066 # Main decision process
2067 # ---------------------------------------------------------------------
2069 # Simple failures
2070 if not self.patient_lookup.pt_found:
2071 self.stop("no patient found")
2072 return
2073 if self.consent_mode.exclude_entirely:
2074 self.stop(
2075 "patient has exclude_entirely flag set; "
2076 " POTENTIALLY SERIOUS ERROR in that this patient shouldn't"
2077 " have been in the anonymised database."
2078 )
2079 return
2080 if self.patient_lookup.pt_dead:
2081 self.stop("patient is dead")
2082 return
2083 if self.consent_mode.consent_mode == ConsentMode.RED:
2084 self.stop("patient's consent mode is RED")
2085 return
2087 # Age?
2088 if self.patient_lookup.pt_dob is None:
2089 self.stop("patient DOB unknown")
2090 return
2091 if (
2092 not self.study.include_under_16s
2093 and self.patient_lookup.is_under_16()
2094 ):
2095 self.stop("patient is under 16 and study not approved for that")
2096 return
2098 # Discharged/outside discharge criteria?
2099 if self.patient_lookup.pt_discharged:
2100 if not self.study.include_discharged:
2101 self.stop(
2102 "patient is discharged and study not approved for that"
2103 )
2104 return
2105 # if self.consent_mode.consent_mode not in (ConsentMode.GREEN,
2106 # ConsentMode.YELLOW):
2107 # self.stop("patient is discharged and consent mode is not "
2108 # "GREEN or YELLOW")
2109 # return
2110 days_since_discharge = self.patient_lookup.days_since_discharge()
2111 permitted_n_days = (
2112 settings.PERMITTED_TO_CONTACT_DISCHARGED_PATIENTS_FOR_N_DAYS
2113 )
2114 if not self.consent_mode.consent_after_discharge:
2115 if days_since_discharge is None:
2116 self.stop(
2117 "patient is discharged; patient did not consent "
2118 "to contact after discharge; unable to "
2119 "determine days since discharge"
2120 )
2121 return
2122 if days_since_discharge > permitted_n_days:
2123 self.stop(
2124 f"patient was discharged {days_since_discharge} days "
2125 f"ago; permission exists only for up to "
2126 f"{permitted_n_days} days; patient did not consent to "
2127 f"contact after discharge"
2128 )
2129 return
2131 # Maximum number of approaches exceeded?
2132 if self.consent_mode.max_approaches_per_year > 0:
2133 if (
2134 self.approaches_in_past_year
2135 >= self.consent_mode.max_approaches_per_year
2136 ):
2137 self.stop(
2138 f"patient has had {self.approaches_in_past_year} "
2139 f"approaches in the past year and has set a cap of "
2140 f"{self.consent_mode.max_approaches_per_year} per year"
2141 )
2142 return
2144 # ---------------------------------------------------------------------
2145 # OK. If we get here, we're going to try to contact someone!
2146 # ---------------------------------------------------------------------
2148 # Direct?
2149 self.save() # makes self.id, needed for FKs
2150 if (
2151 self.consent_mode.consent_mode == ConsentMode.GREEN
2152 and self.request_direct_approach
2153 ):
2154 # noinspection PyTypeChecker
2155 letter = Letter.create_researcher_approval(self) # will save
2156 self.decided_send_to_researcher = True
2157 self.clinician_involvement = (
2158 ContactRequest.CLINICIAN_INVOLVEMENT_NONE
2159 )
2160 self.decide(
2161 "GREEN: Researchers prefer direct approach and patient"
2162 " has chosen green mode: send approval to researcher."
2163 )
2165 # CLARIFICATION, CPFT Research Database Oversight Committee
2166 # 2018-11-12: even for CTIMPs, if patient is GREEN, researchers can
2167 # contact directly -- but will need consultant involvement at the
2168 # later (consent) stage.
2170 # noinspection PyUnresolvedReferences
2171 researcher_emailaddr = self.study.lead_researcher.email
2172 try:
2173 validate_email(researcher_emailaddr)
2174 # noinspection PyTypeChecker
2175 email = Email.create_researcher_approval_email(self, letter)
2176 emailtransmission = email.send()
2177 if emailtransmission.sent:
2178 self.decide(
2179 f"Sent approval to researcher at "
2180 f"{researcher_emailaddr}"
2181 )
2182 return
2183 self.decide(
2184 f"Failed to e-mail approval to researcher at "
2185 f"{researcher_emailaddr}."
2186 )
2187 # noinspection PyTypeChecker
2188 self.decide(emailtransmission.failure_reason)
2189 except ValidationError:
2190 pass
2191 self.decide(
2192 "Approval letter to researcher created and needs " "printing"
2193 )
2194 self.notify_rdbm_of_work(letter, to_researcher=True)
2195 return
2197 # All other routes are via clinician.
2199 # noinspection PyTypeChecker
2200 self.clinician_involvement = self.get_clinician_involvement(
2201 consent_mode_str=self.consent_mode.consent_mode,
2202 request_direct_approach=self.request_direct_approach,
2203 )
2205 # Do we have a clinician?
2206 if not self.patient_lookup.clinician_found:
2207 self.stop("don't know clinician; can't proceed")
2208 return
2209 clinician_emailaddr = self.clinician_email
2210 try:
2211 validate_email(clinician_emailaddr)
2212 except ValidationError:
2213 self.stop(f"clinician e-mail ({clinician_emailaddr}) is invalid")
2214 return
2215 try:
2216 # noinspection PyTypeChecker
2217 validate_researcher_email_domain(clinician_emailaddr)
2218 except ValidationError:
2219 self.stop(
2220 f"clinician e-mail ({clinician_emailaddr}) "
2221 f"is not in a permitted domain"
2222 )
2223 return
2225 # Warnings
2226 if (
2227 ContactRequest.objects.filter(nhs_number=self.nhs_number)
2228 .filter(study=self.study)
2229 .filter(decided_send_to_clinician=True)
2230 .filter(clinician_response__responded=False)
2231 .exists()
2232 ):
2233 self.decide(
2234 "WARNING: outstanding request to clinician for same "
2235 "patient/study."
2236 )
2237 if (
2238 ContactRequest.objects.filter(nhs_number=self.nhs_number)
2239 .filter(study=self.study)
2240 .filter(decided_send_to_clinician=True)
2241 .filter(clinician_response__responded=True)
2242 .filter(
2243 clinician_response__response__in=[
2244 ClinicianResponse.RESPONSE_B,
2245 ClinicianResponse.RESPONSE_C,
2246 ClinicianResponse.RESPONSE_D,
2247 ]
2248 )
2249 .exists()
2250 ):
2251 self.decide(
2252 "WARNING: clinician has already rejected a request "
2253 "about this patient/study."
2254 )
2256 # If the request is clinician initiated, we need to send a different
2257 # email. This will also create a clinician response and set the
2258 # clinician's response to either 'yes I will contact the patient' or
2259 # 'yes but let the RDBM contact them for me'
2260 if self.clinician_initiated:
2261 email = Email.create_clinician_initiated_cr_email(self)
2262 emailtransmission = email.send()
2263 if not emailtransmission.sent:
2264 # noinspection PyTypeChecker
2265 self.decide(emailtransmission.failure_reason)
2266 self.stop(
2267 f"Failed to send e-mail to clinician at "
2268 f"{clinician_emailaddr}"
2269 )
2270 self.decided_send_to_clinician = True
2271 self.decide(f"Sent request to clinician at {clinician_emailaddr}")
2272 return
2274 # Send e-mail to clinician
2275 # noinspection PyTypeChecker
2276 email = Email.create_clinician_email(self)
2277 # ... will also create a ClinicianResponse
2278 emailtransmission = email.send()
2279 if not emailtransmission.sent:
2280 # noinspection PyTypeChecker
2281 self.decide(emailtransmission.failure_reason)
2282 self.stop(
2283 f"Failed to send e-mail to clinician at "
2284 f"{clinician_emailaddr}"
2285 )
2286 # We don't set decided_send_to_clinician because this attempt has
2287 # failed, and we don't want to put anyone off trying again
2288 # immediately.
2289 self.decided_send_to_clinician = True
2290 self.decide(f"Sent request to clinician at {clinician_emailaddr}")
2292 @staticmethod
2293 def get_clinician_involvement(
2294 consent_mode_str: str, request_direct_approach: bool
2295 ) -> int:
2296 """
2297 Returns a number indicating why a clinician is involved.
2299 Args:
2300 consent_mode_str: consent mode in use (see :class:`ConsentMode`)
2301 request_direct_approach: do the researchers request direct
2302 approach to the patient, if permitted?
2304 Returns:
2305 an integer constant; see :class:`ContactRequest`
2307 """
2308 # Let's be precise about why the clinician is involved.
2309 if not request_direct_approach:
2310 return ContactRequest.CLINICIAN_INVOLVEMENT_REQUESTED
2311 elif consent_mode_str == ConsentMode.YELLOW:
2312 return ContactRequest.CLINICIAN_INVOLVEMENT_REQUIRED_YELLOW
2313 else:
2314 # Only other possibility
2315 return ContactRequest.CLINICIAN_INVOLVEMENT_REQUIRED_UNKNOWN
2317 def decide(self, msg: str) -> None:
2318 """
2319 Make a note of a decision.
2320 """
2321 self.decisionlist.append(msg)
2323 def stop(self, msg: str) -> None:
2324 """
2325 Make a note of a decision and that we have finished processing this
2326 contact request, taking no further action.
2327 """
2328 self.decide("Stopping: " + msg)
2329 self.decided_no_action = True
2331 def calc_approaches_in_past_year(self) -> None:
2332 """
2333 Sets :attr:`approaches_in_past_year` to indicate the number of
2334 approaches in the past year to this patient via CRATE.
2336 How best to count this? Not by e.g. calendar year, with a flag that
2337 gets reset to zero annually, because you might have a limit of 5, and
2338 get 4 requests in Dec 2020 and then another 4 in Jan 2021 just after
2339 the flag resets. Instead, we count the number of requests to that
2340 patient in the past year.
2342 """
2343 one_year_ago = timezone.now() - datetime.timedelta(days=365)
2345 self.approaches_in_past_year = ContactRequest.objects.filter(
2346 Q(decided_send_to_researcher=True)
2347 | (
2348 Q(decided_send_to_clinician=True)
2349 & (
2350 Q(
2351 clinician_response__response=ClinicianResponse.RESPONSE_A # noqa: E501
2352 )
2353 | Q(
2354 clinician_response__response=ClinicianResponse.RESPONSE_R # noqa: E501
2355 )
2356 )
2357 ),
2358 nhs_number=self.nhs_number,
2359 created_at__gte=one_year_ago,
2360 ).count()
2362 def withdraw_consent(self) -> Tuple[LETTER_FWD_REF, bool]:
2363 """
2364 Withdraws consent that had previously been given. Will e-mail the
2365 researcher to let them know, if it can.
2367 Returns:
2368 tuple: ``letter, email_succeeded`` where ``letter`` is a
2369 :class:`Letter` to the researcher and ``email_succeeded`` indicates
2370 whether we managed to e-mail the researcher.
2372 """
2373 self.consent_withdrawn = True
2374 self.consent_withdrawn_at = timezone.now()
2375 self.save()
2376 # noinspection PyTypeChecker
2377 letter = Letter.create_researcher_withdrawal(self) # will save
2378 # noinspection PyUnresolvedReferences
2379 researcher_emailaddr = self.study.lead_researcher.email
2380 email_succeeded = False
2381 try:
2382 validate_email(researcher_emailaddr)
2383 # noinspection PyTypeChecker
2384 email = Email.create_researcher_withdrawal_email(self, letter)
2385 emailtransmission = email.send()
2386 email_succeeded = emailtransmission.sent
2387 except ValidationError:
2388 pass
2389 return letter, email_succeeded
2391 def get_permission_date(self) -> Optional[datetime.datetime]:
2392 """
2393 When was the researcher given permission? Used for the letter
2394 withdrawing permission.
2395 """
2396 if self.decided_no_action:
2397 return None
2398 if self.decided_send_to_researcher:
2399 # Green route
2400 # noinspection PyTypeChecker
2401 return self.created_at
2402 if self.decided_send_to_clinician:
2403 # Yellow route -> patient -> said yes
2404 if hasattr(self, "patient_response"):
2405 if self.patient_response.response == PatientResponse.YES:
2406 return self.patient_response.created_at
2407 return None
2409 def notify_rdbm_of_work(
2410 self, letter: LETTER_FWD_REF, to_researcher: bool = False
2411 ) -> None:
2412 """
2413 E-mail the RDBM to say that there's work to do.
2415 Args:
2416 letter: a :class:`Letter`
2417 to_researcher: is it a letter that needs to go to a researcher
2418 manually, rather than a letter that a clinician wants the
2419 RDBM to send on their behalf?
2420 """
2421 subject = (
2422 f"CHEERFUL WORK FROM RESEARCH DATABASE COMPUTER - "
2423 f"contact request {self.id}"
2424 )
2425 if to_researcher:
2426 template = "email_rdbm_new_work_researcher.html"
2427 else:
2428 template = "email_rdbm_new_work_pt_from_clinician.html"
2429 html = render_email_html_to_string(template, {"letter": letter})
2430 email = Email.create_rdbm_email(subject, html)
2431 email.send()
2433 def notify_rdbm_of_bad_progress(self) -> None:
2434 """
2435 Lets the RDBM know that a clinician refused (vetoed) a request.
2436 """
2437 subject = (
2438 f"INFO ONLY - clinician refused Research Database request "
2439 f"- contact request {self.id}"
2440 )
2441 html = render_email_html_to_string(
2442 "email_rdbm_bad_progress.html",
2443 {
2444 "id": self.id,
2445 "response": self.clinician_response.response,
2446 "explanation": self.clinician_response.get_response_explanation(), # noqa: E501
2447 },
2448 )
2449 email = Email.create_rdbm_email(subject, html)
2450 email.send()
2452 def notify_rdbm_of_good_progress(self) -> None:
2453 """
2454 Lets the RDBM know that a clinician said yes to a request and wishes to
2455 do the work themselves.
2456 """
2457 subject = (
2458 f"INFO ONLY - clinician agreed to Research Database request"
2459 f" - contact request {self.id}"
2460 )
2461 html = render_email_html_to_string(
2462 "email_rdbm_good_progress.html",
2463 {
2464 "id": self.id,
2465 "response": self.clinician_response.response,
2466 "explanation": self.clinician_response.get_response_explanation(), # noqa: E501
2467 },
2468 )
2469 email = Email.create_rdbm_email(subject, html)
2470 email.send()
2472 def get_clinician_email_html(self, save: bool = True) -> str:
2473 """
2474 **REC DOCUMENTS 09, 11, 13 (A): E-mail to clinician asking them to pass
2475 on contact request.**
2477 Args:
2478 save: save the e-mail to the database? (Only false for testing.)
2480 Returns:
2481 HTML for this e-mail
2483 - When we create a URL, should we put parameters in the path,
2484 querystring, or both?
2486 - see notes in ``core/utils.py``
2487 - In this case, we decide as follows: since we are creating a
2488 :class:`ClinicianResponse`, we should use its ModelForm.
2489 - URL path for PK
2490 - querystring for other parameters, with form-based validation
2492 """
2493 clinician_response = ClinicianResponse.create(self, save=save)
2494 if not save:
2495 clinician_response.id = -1 # dummy PK, guaranteed to fail
2496 context = {
2497 "contact_request": self,
2498 "study": self.study,
2499 "patient_lookup": self.patient_lookup,
2500 "consent_mode": self.consent_mode,
2501 "settings": settings,
2502 "url_yes": clinician_response.get_abs_url_yes(),
2503 "url_no": clinician_response.get_abs_url_no(),
2504 "url_maybe": clinician_response.get_abs_url_maybe(),
2505 "permitted_to_contact_discharged_patients_for_n_days": settings.PERMITTED_TO_CONTACT_DISCHARGED_PATIENTS_FOR_N_DAYS, # noqa: E501
2506 "permitted_to_contact_discharged_patients_for_n_years": days_to_years( # noqa: E501
2507 settings.PERMITTED_TO_CONTACT_DISCHARGED_PATIENTS_FOR_N_DAYS
2508 ),
2509 }
2510 return render_email_html_to_string("email_clinician.html", context)
2512 def get_clinician_initiated_email_html(self, save: bool = True) -> str:
2513 """
2514 Email to clinician confirming a clinician-initiated contact request.
2515 Will inlcude a link to the clinician pack if they do not want the RDBM
2516 to contact the patient for them. Also sets the clinician's response.
2518 Args:
2519 save: save the e-mail to the database? (Only false for testing.)
2521 Returns:
2522 HTML for this e-mail
2524 """
2525 clinician_response = ClinicianResponse.create(self, save=save)
2526 if not save:
2527 clinician_response.id = -1 # dummy PK, guaranteed to fail
2528 if self.rdbm_to_contact_pt:
2529 clinician_response.response = ClinicianResponse.RESPONSE_R
2530 else:
2531 clinician_response.response = ClinicianResponse.RESPONSE_A
2532 clinician_response.finalize_a() # first part of processing
2533 transaction.on_commit(
2534 lambda: finalize_clinician_response.delay(clinician_response.id)
2535 )
2536 rev = reverse(
2537 "clinician_pack",
2538 args=[clinician_response.id, clinician_response.token],
2539 )
2540 url_pack = site_absolute_url(rev)
2541 context = {
2542 "contact_request": self,
2543 "study": self.study,
2544 "patient_lookup": self.patient_lookup,
2545 "consent_mode": self.consent_mode,
2546 "clinician_response": clinician_response,
2547 "settings": settings,
2548 "url_pack": url_pack,
2549 }
2550 return render_email_html_to_string(
2551 "email_clinician_initiated_cr.html", context
2552 )
2554 def get_approval_letter_html(self) -> str:
2555 """
2556 **REC DOCUMENT 15. Letter to researcher approving contact.**
2558 Returns the HTML for this letter.
2559 """
2560 context = {
2561 # Letter bits
2562 "address_from": (
2563 settings.RDBM_ADDRESS
2564 + [settings.RDBM_TELEPHONE, settings.RDBM_EMAIL]
2565 ),
2566 "address_to": self.study.get_lead_researcher_name_address(),
2567 "salutation": self.study.get_lead_researcher_salutation(),
2568 "signatory_name": settings.RDBM_NAME,
2569 "signatory_title": settings.RDBM_TITLE,
2570 # Specific bits
2571 "contact_request": self,
2572 "study": self.study,
2573 "patient_lookup": self.patient_lookup,
2574 "consent_mode": self.consent_mode,
2575 "permitted_to_contact_discharged_patients_for_n_days": settings.PERMITTED_TO_CONTACT_DISCHARGED_PATIENTS_FOR_N_DAYS, # noqa: E501
2576 "permitted_to_contact_discharged_patients_for_n_years": days_to_years( # noqa: E501
2577 settings.PERMITTED_TO_CONTACT_DISCHARGED_PATIENTS_FOR_N_DAYS
2578 ),
2579 "RDBM_ADDRESS": settings.RDBM_ADDRESS,
2580 }
2581 return render_pdf_html_to_string(
2582 "letter_researcher_approve.html", context, patient=False
2583 )
2585 def get_withdrawal_letter_html(self) -> str:
2586 """
2587 **REC DOCUMENT 16. Letter to researcher notifying them of withdrawal of
2588 consent.**
2590 Returns the HTML for this letter.
2591 """
2592 context = {
2593 # Letter bits
2594 "address_from": (
2595 settings.RDBM_ADDRESS
2596 + [settings.RDBM_TELEPHONE, settings.RDBM_EMAIL]
2597 ),
2598 "address_to": self.study.get_lead_researcher_name_address(),
2599 "salutation": self.study.get_lead_researcher_salutation(),
2600 "signatory_name": settings.RDBM_NAME,
2601 "signatory_title": settings.RDBM_TITLE,
2602 # Specific bits
2603 "contact_request": self,
2604 "study": self.study,
2605 "patient_lookup": self.patient_lookup,
2606 "consent_mode": self.consent_mode,
2607 }
2608 return render_pdf_html_to_string(
2609 "letter_researcher_withdraw.html", context, patient=False
2610 )
2612 def get_approval_email_html(self) -> str:
2613 """
2614 Returns HTML for a simple e-mail to the researcher attaching an
2615 approval letter.
2616 """
2617 context = {
2618 "contact_request": self,
2619 "study": self.study,
2620 "patient_lookup": self.patient_lookup,
2621 "consent_mode": self.consent_mode,
2622 }
2623 return render_email_html_to_string(
2624 "email_researcher_approval.html", context
2625 )
2627 def get_withdrawal_email_html(self) -> str:
2628 """
2629 Returns HTML for a simple e-mail to the researcher attaching an
2630 withdrawal-of-previous-consent letter.
2631 """
2632 context = {
2633 "contact_request": self,
2634 "study": self.study,
2635 "patient_lookup": self.patient_lookup,
2636 "consent_mode": self.consent_mode,
2637 }
2638 return render_email_html_to_string(
2639 "email_researcher_withdrawal.html", context
2640 )
2642 def get_letter_clinician_to_pt_re_study(self) -> str:
2643 """
2644 Letters from clinician to patient, with decision form.
2646 Returns the HTML for this letter.
2647 """
2648 patient_lookup = self.patient_lookup
2649 if not patient_lookup:
2650 raise Http404(
2651 "No patient_lookup: is the back-end message queue "
2652 "(e.g. Celery + RabbitMQ) running?"
2653 )
2654 yellow = (
2655 self.clinician_involvement
2656 == ContactRequest.CLINICIAN_INVOLVEMENT_REQUIRED_YELLOW
2657 )
2658 if self.clinician_initiated:
2659 clinician_address_components = self.request_by_address_components()
2660 else:
2661 clinician_address_components = (
2662 patient_lookup.clinician_address_components()
2663 )
2664 context = {
2665 # Letter bits
2666 "address_from": clinician_address_components,
2667 "address_to": patient_lookup.pt_name_address_components(),
2668 "salutation": patient_lookup.pt_salutation(),
2669 "signatory_name": self.clinician_signatory_name,
2670 "signatory_title": self.clinician_signatory_title,
2671 # Specific bits
2672 "contact_request": self,
2673 "study": self.study,
2674 "patient_lookup": patient_lookup,
2675 "settings": settings,
2676 "extra_form": self.is_extra_form(),
2677 "yellow": yellow,
2678 "unknown_consent_mode": self.is_consent_mode_unknown(),
2679 }
2680 return render_pdf_html_to_string(
2681 "letter_patient_from_clinician_re_study.html",
2682 context,
2683 patient=True,
2684 )
2686 def is_extra_form(self) -> bool:
2687 """
2688 Is there an extra form from the researchers that they wish passed on to
2689 the patient?
2690 """
2691 study = self.study
2692 clinician_requested = not self.request_direct_approach
2693 extra_form = (
2694 clinician_requested and study.subject_form_template_pdf.name
2695 )
2696 # log.debug(f"clinician_requested: {clinician_requested}")
2697 # log.debug(f"extra_form: {extra_form}")
2698 return extra_form
2700 def is_consent_mode_unknown(self) -> bool:
2701 """
2702 Is the consent mode "unknown" (NULL in the database)?
2703 """
2704 return not self.consent_mode.consent_mode
2706 def get_decision_form_to_pt_re_study(self) -> str:
2707 """
2708 Returns HTML for the form for the patient to decide about this
2709 study.
2710 """
2711 n_forms = 1
2712 extra_form = self.is_extra_form()
2713 if extra_form:
2714 n_forms += 1
2715 yellow = (
2716 self.clinician_involvement
2717 == ContactRequest.CLINICIAN_INVOLVEMENT_REQUIRED_YELLOW
2718 )
2719 unknown = (
2720 self.clinician_involvement
2721 == ContactRequest.CLINICIAN_INVOLVEMENT_REQUIRED_UNKNOWN
2722 )
2723 if unknown:
2724 n_forms += 1
2725 context = {
2726 "contact_request": self,
2727 "study": self.study,
2728 "patient_lookup": self.patient_lookup,
2729 "settings": settings,
2730 "extra_form": extra_form,
2731 "n_forms": n_forms,
2732 "yellow": yellow,
2733 }
2734 return render_pdf_html_to_string(
2735 "decision_form_to_patient_re_study.html", context, patient=True
2736 )
2738 def get_clinician_pack_pdf(self) -> bytes:
2739 """
2740 Returns a PDF of the "clinician pack": a cover letter, decision forms,
2741 and any other information required, customized for this request.
2742 """
2743 # Order should match letter...
2745 # Letter to patient from clinician
2746 pdf_plans = [
2747 CratePdfPlan(
2748 is_html=True,
2749 html=self.get_letter_clinician_to_pt_re_study(),
2750 ethics_doccode=EthicsInfo.LTR_C_P_STUDY,
2751 )
2752 ]
2753 # Study details
2754 if self.study.study_details_pdf:
2755 # noinspection PyUnresolvedReferences
2756 pdf_plans.append(
2757 CratePdfPlan(
2758 is_filename=True,
2759 filename=self.study.study_details_pdf.path,
2760 )
2761 )
2762 # Decision form about this study
2763 pdf_plans.append(
2764 CratePdfPlan(
2765 is_html=True,
2766 html=self.get_decision_form_to_pt_re_study(),
2767 ethics_doccode=EthicsInfo.FORM_STUDY,
2768 )
2769 )
2770 # Additional form for this study
2771 if self.is_extra_form():
2772 if self.study.subject_form_template_pdf:
2773 # noinspection PyUnresolvedReferences
2774 pdf_plans.append(
2775 CratePdfPlan(
2776 is_filename=True,
2777 filename=self.study.subject_form_template_pdf.path,
2778 )
2779 )
2780 # Traffic-light decision form, if consent mode unknown
2781 if self.is_consent_mode_unknown():
2782 # 2017-03-03: changed to a personalized version
2784 # try:
2785 # leaflet = Leaflet.objects.get(
2786 # name=Leaflet.CPFT_TRAFFICLIGHT_CHOICE)
2787 # pdf_plans.append(PdfPlan(is_filename=True,
2788 # filename=leaflet.pdf.path))
2789 # except ObjectDoesNotExist:
2790 # log.warning("Missing traffic-light leaflet!")
2791 # email_rdbm_task.delay(
2792 # subject="ERROR FROM RESEARCH DATABASE COMPUTER",
2793 # text=(
2794 # "Missing traffic-light leaflet! Incomplete clinician " # noqa: E501
2795 # "pack accessed for contact request {}.".format(
2796 # self.id)
2797 # )
2798 # )
2800 pdf_plans.append(
2801 CratePdfPlan(
2802 is_html=True,
2803 html=self.patient_lookup.get_traffic_light_decision_form(),
2804 ethics_doccode=EthicsInfo.FORM_TRAFFIC_PERSONALIZED,
2805 )
2806 )
2807 # General info leaflet
2808 try:
2809 leaflet = Leaflet.objects.get(name=Leaflet.CPFT_TPIR)
2810 pdf_plans.append(
2811 CratePdfPlan(is_filename=True, filename=leaflet.pdf.path)
2812 )
2813 except ObjectDoesNotExist:
2814 log.warning("Missing taking-part-in-research leaflet!")
2815 email_rdbm_task.delay(
2816 subject="ERROR FROM RESEARCH DATABASE COMPUTER",
2817 text=(
2818 f"Missing taking-part-in-research leaflet! Incomplete "
2819 f"clinician pack accessed for contact request {self.id}."
2820 ),
2821 )
2822 return get_concatenated_pdf_in_memory(pdf_plans, start_recto=True)
2824 def get_mgr_admin_url(self) -> str:
2825 """
2826 Returns the URL for the admin site to view this
2827 :class:`ContactRequest`.
2828 """
2829 from crate_anon.crateweb.core.admin import (
2830 mgr_admin_site,
2831 ) # delayed import
2833 return admin_view_url(mgr_admin_site, self)
2835 def request_by_address_components(self) -> List[str]:
2836 """
2837 Returns the address of the person who made the contact request -- or
2838 the Research Database Manager's (with "c/o") if we don't know the
2839 requester's.
2841 This will be used in cases of a clinician-iniated request, for use in
2842 letters to the patient.
2843 """
2844 try:
2845 userprofile = UserProfile.objects.get(user=self.request_by)
2846 except UserProfile.DoesNotExist:
2847 log.warning(
2848 "ContactRequest object needs 'request_by' to be "
2849 "a valid user for 'request_by_address_components'."
2850 )
2851 address_components = [] # type: List[str]
2852 else:
2853 address_components = [
2854 userprofile.address_1,
2855 userprofile.address_2,
2856 userprofile.address_3,
2857 userprofile.address_4,
2858 userprofile.address_5,
2859 userprofile.address_6,
2860 userprofile.address_7,
2861 ]
2862 if not any(x for x in address_components):
2863 address_components = settings.RDBM_ADDRESS.copy()
2864 if address_components:
2865 address_components[0] = "c/o " + address_components[0]
2866 return list(filter(None, address_components))
2869# =============================================================================
2870# Clinician response
2871# =============================================================================
2874class ClinicianResponse(models.Model):
2875 """
2876 Represents the response of a clinician to a :class:`ContactRequest` that
2877 was routed to them.
2878 """
2880 TOKEN_LENGTH_CHARS = 20
2881 # info_bits = math.log(math.pow(26 + 26 + 10, TOKEN_LENGTH_CHARS), 2)
2882 # p_guess = math.pow(0.5, info_bits)
2884 RESPONSE_A = "A"
2885 RESPONSE_B = "B"
2886 RESPONSE_C = "C"
2887 RESPONSE_D = "D"
2888 RESPONSE_R = "R"
2889 RESPONSES = (
2890 (RESPONSE_R, "R: Clinician asks RDBM to pass request to patient"),
2891 (RESPONSE_A, "A: Clinician will pass the request to the patient"),
2892 (RESPONSE_B, "B: Clinician vetoes on clinical grounds"),
2893 (RESPONSE_C, "C: Patient is definitely ineligible"),
2894 (
2895 RESPONSE_D,
2896 "D: Patient is deceased, discharged, or details are defunct",
2897 ),
2898 )
2900 ROUTE_EMAIL = "e"
2901 ROUTE_WEB = "w"
2902 RESPONSE_ROUTES = (
2903 (ROUTE_EMAIL, "E-mail"),
2904 (ROUTE_WEB, "Web"),
2905 )
2907 EMAIL_CHOICE_Y = "y"
2908 EMAIL_CHOICE_N = "n"
2909 EMAIL_CHOICE_TELL_ME_MORE = "more"
2910 EMAIL_CHOICES = (
2911 (EMAIL_CHOICE_Y, "Yes"),
2912 (EMAIL_CHOICE_N, "No"),
2913 (EMAIL_CHOICE_TELL_ME_MORE, "Tell me more"),
2914 )
2916 created_at = models.DateTimeField(
2917 verbose_name="When created", auto_now_add=True
2918 )
2919 contact_request = models.OneToOneField(
2920 ContactRequest,
2921 on_delete=models.PROTECT,
2922 related_name="clinician_response",
2923 )
2924 token = models.CharField(max_length=TOKEN_LENGTH_CHARS)
2925 responded = models.BooleanField(default=False, verbose_name="Responded?")
2926 responded_at = models.DateTimeField(
2927 verbose_name="When responded", null=True
2928 )
2929 response_route = models.CharField(max_length=1, choices=RESPONSE_ROUTES)
2930 email_choice = models.CharField(max_length=4, choices=EMAIL_CHOICES)
2931 response = models.CharField(max_length=1, choices=RESPONSES)
2932 veto_reason = models.TextField(
2933 blank=True, verbose_name="Reason for clinical veto"
2934 )
2935 ineligible_reason = models.TextField(
2936 blank=True, verbose_name="Reason patient is ineligible"
2937 )
2938 pt_uncontactable_reason = models.TextField(
2939 blank=True, verbose_name="Reason patient is not contactable"
2940 )
2941 clinician_confirm_name = models.CharField(
2942 max_length=255, verbose_name="Type your name to confirm"
2943 )
2944 charity_amount_due = models.DecimalField(
2945 max_digits=8, decimal_places=2, default=0
2946 )
2947 # ... set to settings.CHARITY_AMOUNT_CLINICIAN_RESPONSE upon response
2949 processed = models.BooleanField(default=False) # added 2018-06-29
2950 processed_at = models.DateTimeField(null=True) # added 2018-06-29
2952 def get_response_explanation(self) -> str:
2953 """
2954 Returns the human-readable description of the clinician's response.
2955 """
2956 # log.debug(f"get_response_explanation: {self.response}")
2957 # noinspection PyTypeChecker
2958 return choice_explanation(self.response, ClinicianResponse.RESPONSES)
2960 @classmethod
2961 def create(
2962 cls, contact_request: ContactRequest, save: bool = True
2963 ) -> CLINICIAN_RESPONSE_FWD_REF:
2964 """
2965 Creates a new clinician response object.
2967 Args:
2968 contact_request: a :class:`ContactRequest`
2969 save: save to the database? (Only false for debugging.)
2971 Returns:
2972 a :class:`ClinicianResponse`
2974 """
2975 newtoken = get_random_string(ClinicianResponse.TOKEN_LENGTH_CHARS)
2976 # https://github.com/django/django/blob/master/django/utils/crypto.py#L51 # noqa: E501
2977 clinician_response = cls(
2978 contact_request=contact_request,
2979 token=newtoken,
2980 )
2981 if save:
2982 clinician_response.save()
2983 return clinician_response
2985 def get_abs_url_path(self) -> str:
2986 """
2987 Returns an absolute URL path to the page that lets the clinician
2988 respond for this :class:`ClinicianResponse`.
2990 This is used in the e-mail to the clinician.
2991 """
2992 rev = reverse(UrlNames.CLINICIAN_RESPONSE, args=[self.id])
2993 url = site_absolute_url(rev)
2994 return url
2996 def get_common_querydict(self, email_choice: str) -> QueryDict:
2997 """
2998 Returns a query dictionary that will contribute to our final URLs. That
2999 is, information about the clinician's choice (and also a security
3000 token) that will be added to the base "response" URL path.
3002 Args:
3003 email_choice: code for the clinician's choice
3005 Returns:
3006 a :class:`django.http.request.QueryDict`
3008 """
3009 querydict = QueryDict(mutable=True)
3010 querydict["token"] = self.token
3011 querydict["email_choice"] = email_choice
3012 return querydict
3014 def get_abs_url(self, email_choice: str) -> str:
3015 """
3016 Returns an absolute URL representing a specific choice for the
3017 clinician.
3019 Args:
3020 email_choice: code for the clinician's choice
3022 Returns:
3023 a URL
3025 """
3026 path = self.get_abs_url_path()
3027 querydict = self.get_common_querydict(email_choice)
3028 return url_with_querystring(path, querydict)
3030 def get_abs_url_yes(self) -> str:
3031 """
3032 Returns an absolute URL for "clinician says yes".
3033 """
3034 return self.get_abs_url(ClinicianResponse.EMAIL_CHOICE_Y)
3036 def get_abs_url_no(self) -> str:
3037 """
3038 Returns an absolute URL for "clinician says no".
3039 """
3040 return self.get_abs_url(ClinicianResponse.EMAIL_CHOICE_N)
3042 def get_abs_url_maybe(self) -> str:
3043 """
3044 Returns an absolute URL for "clinician says tell me more".
3045 """
3046 return self.get_abs_url(ClinicianResponse.EMAIL_CHOICE_TELL_ME_MORE)
3048 def __str__(self) -> str:
3049 return (
3050 f"[ClinicianResponse {self.id}] "
3051 f"ContactRequest {self.contact_request_id}"
3052 )
3054 def finalize_a(self) -> None:
3055 """
3056 Call this when the clinician completes their response.
3058 Part A: immediate, called from the web front end, for acknowledgement.
3059 """
3060 self.responded = True
3061 self.responded_at = timezone.now()
3062 self.charity_amount_due = settings.CHARITY_AMOUNT_CLINICIAN_RESPONSE
3063 self.save()
3065 @staticmethod
3066 def get_unprocessed() -> QuerySet:
3067 return ClinicianResponse.objects.filter(processed=False)
3069 def finalize_b(self) -> None:
3070 """
3071 Call this when the clinician completes their response.
3073 Part B: called by the background task processor, for the slower
3074 aspects.
3075 """
3076 if self.processed:
3077 log.warning(
3078 f"ClinicianResponse #{self.id}: already processed; "
3079 f"not processing again"
3080 )
3081 return
3082 if self.response == ClinicianResponse.RESPONSE_R:
3083 # noinspection PyTypeChecker
3084 letter = Letter.create_request_to_patient(
3085 self.contact_request, rdbm_may_view=True
3086 )
3087 # ... will save
3088 # noinspection PyTypeChecker
3089 PatientResponse.create(self.contact_request)
3090 # ... will save
3091 self.contact_request.notify_rdbm_of_work(letter)
3092 elif self.response == ClinicianResponse.RESPONSE_A:
3093 # noinspection PyTypeChecker
3094 Letter.create_request_to_patient(
3095 self.contact_request, rdbm_may_view=False
3096 )
3097 # ... return value not used
3098 # noinspection PyTypeChecker
3099 PatientResponse.create(self.contact_request)
3100 self.contact_request.notify_rdbm_of_good_progress()
3101 elif self.response in (
3102 ClinicianResponse.RESPONSE_B,
3103 ClinicianResponse.RESPONSE_C,
3104 ClinicianResponse.RESPONSE_D,
3105 ):
3106 self.contact_request.notify_rdbm_of_bad_progress()
3107 self.processed = True
3108 self.processed_at = timezone.now()
3109 self.save()
3112# =============================================================================
3113# Patient response
3114# =============================================================================
3116PATIENT_RESPONSE_FWD_REF = "PatientResponse"
3119class PatientResponse(Decision):
3120 """
3121 Represents the patient's decision about a specific study. (We get one of
3122 these if the clinician passed details to the patient and the patient has
3123 responded.)
3124 """
3126 YES = 1
3127 NO = 2
3128 RESPONSES = (
3129 (YES, "1: Yes"),
3130 (NO, "2: No"),
3131 )
3132 created_at = models.DateTimeField(
3133 verbose_name="When created", auto_now_add=True
3134 )
3135 contact_request = models.OneToOneField(
3136 ContactRequest,
3137 on_delete=models.PROTECT,
3138 related_name="patient_response",
3139 )
3140 recorded_by = models.ForeignKey(
3141 settings.AUTH_USER_MODEL, on_delete=models.PROTECT, null=True
3142 )
3143 response = models.PositiveSmallIntegerField(
3144 null=True, choices=RESPONSES, verbose_name="Patient's response"
3145 )
3146 processed = models.BooleanField(default=False) # added 2018-06-29
3147 processed_at = models.DateTimeField(null=True) # added 2018-06-29
3149 def __str__(self) -> str:
3150 if self.response:
3151 # noinspection PyTypeChecker
3152 suffix = "response was {}".format(
3153 choice_explanation(self.response, PatientResponse.RESPONSES)
3154 )
3155 else:
3156 suffix = "AWAITING RESPONSE"
3157 return (
3158 f"Patient response {self.id} "
3159 f"(contact request {self.contact_request.id}, "
3160 f"study {self.contact_request.study.id}): {suffix}"
3161 )
3163 @classmethod
3164 def create(
3165 cls, contact_request: ContactRequest
3166 ) -> PATIENT_RESPONSE_FWD_REF:
3167 """
3168 Creates a patient response object for a given contact request.
3170 Args:
3171 contact_request: a :class:`ContactRequest`
3173 Returns:
3174 :class:`PatientResponse`
3176 """
3177 patient_response = cls(contact_request=contact_request)
3178 patient_response.save()
3179 return patient_response
3181 @staticmethod
3182 def get_unprocessed() -> QuerySet:
3183 """
3184 Return all :class:`PatientResponse` objects that need processing.
3186 See :func:`crate_anon.crateweb.consent.tasks.process_patient_response`
3187 and :func:`process_response`, which does the work.
3188 """
3189 return PatientResponse.objects.filter(processed=False)
3191 def process_response(self) -> None:
3192 """
3193 Processes the :class:`PatientResponse` and marks it as processed.
3195 If the patient said yes, this triggers a letter to the researcher.
3196 """
3197 # log.debug(f"process_response: PatientResponse: {modelrepr(self)}")
3198 if self.processed:
3199 log.warning(
3200 f"PatientResponse #{self.id}: already processed; "
3201 f"not processing again"
3202 )
3203 return
3204 if self.response == PatientResponse.YES:
3205 contact_request = self.contact_request
3206 # noinspection PyTypeChecker
3207 letter = Letter.create_researcher_approval(contact_request)
3208 # ... will save
3209 # noinspection PyTypeChecker
3210 email = Email.create_researcher_approval_email(
3211 contact_request, letter
3212 )
3213 emailtransmission = email.send()
3214 emailed = emailtransmission.sent
3215 if not emailed:
3216 contact_request.notify_rdbm_of_work(letter, to_researcher=True)
3217 self.processed = True
3218 self.processed_at = timezone.now()
3219 self.save()
3222# =============================================================================
3223# Letter, and record of letter being printed
3224# =============================================================================
3227class Letter(models.Model):
3228 """
3229 Represents a letter (e.g. to a patient, clinician, or researcher).
3230 """
3232 created_at = models.DateTimeField(
3233 verbose_name="When created", auto_now_add=True
3234 )
3235 pdf = models.FileField(storage=privatestorage)
3236 # Other flags:
3237 to_clinician = models.BooleanField(default=False)
3238 to_researcher = models.BooleanField(default=False)
3239 to_patient = models.BooleanField(default=False)
3240 rdbm_may_view = models.BooleanField(default=False)
3241 study = models.ForeignKey(Study, on_delete=models.PROTECT, null=True)
3242 contact_request = models.ForeignKey(
3243 ContactRequest, on_delete=models.PROTECT, null=True
3244 )
3245 sent_manually_at = models.DateTimeField(null=True)
3247 def __str__(self) -> str:
3248 return f"Letter {self.id}"
3250 @classmethod
3251 def create(
3252 cls,
3253 basefilename: str,
3254 html: str = None,
3255 pdf: bytes = None,
3256 to_clinician: bool = False,
3257 to_researcher: bool = False,
3258 to_patient: bool = False,
3259 rdbm_may_view: bool = False,
3260 study: Study = None,
3261 contact_request: ContactRequest = None,
3262 debug_store_html: bool = False,
3263 ) -> LETTER_FWD_REF:
3264 """
3265 Creates a letter.
3267 Args:
3268 basefilename: filename to be used to store a PDF copy of the letter
3269 on disk (without a path)
3270 html: for letters supplied as HTML, the HTML
3271 pdf: for letters supplied as PDF, the PDF
3272 to_clinician: is the letter to a clinician?
3273 to_researcher: is the letter to a researcher?
3274 to_patient: is the letter to a patient?
3275 rdbm_may_view: may the RDBM view this letter?
3276 study: which :class:`Study` does it relate to, if any?
3277 contact_request: which :class:`ContactRequest` does it relate to,
3278 if any?
3279 debug_store_html: should we store the HTML of the letter, as well
3280 as the PDF (for letters originating in HTML only)?
3282 Returns:
3283 a :class:`Letter`
3285 """
3286 # Writing to a FileField directly: you can use field.save(), but then
3287 # you having to write one file and copy to another, etc.
3288 # Here we use the method of assigning to field.name (you can't assign
3289 # to field.path). Also, note that you should never read
3290 # the path attribute if name is blank; it raises an exception.
3291 if bool(html) == bool(pdf):
3292 # One or the other!
3293 raise ValueError("Invalid html/pdf options to Letter.create")
3294 filename_in_storage = os.path.join("letter", basefilename)
3295 abs_filename = os.path.join(
3296 settings.PRIVATE_FILE_STORAGE_ROOT, filename_in_storage
3297 )
3298 os.makedirs(os.path.dirname(abs_filename), exist_ok=True)
3299 if html:
3300 # HTML supplied
3301 if debug_store_html:
3302 with open(abs_filename + ".html", "w") as f:
3303 f.write(html)
3304 make_pdf_on_disk_from_html_with_django_settings(
3305 html,
3306 output_path=abs_filename,
3307 header_html=None,
3308 footer_html=None,
3309 )
3310 else:
3311 # PDF supplied in memory
3312 with open(abs_filename, "wb") as f:
3313 f.write(pdf)
3314 letter = cls(
3315 to_clinician=to_clinician,
3316 to_researcher=to_researcher,
3317 to_patient=to_patient,
3318 rdbm_may_view=rdbm_may_view,
3319 study=study,
3320 contact_request=contact_request,
3321 )
3322 letter.pdf.name = filename_in_storage
3323 letter.save()
3324 return letter
3326 @classmethod
3327 def create_researcher_approval(
3328 cls, contact_request: ContactRequest
3329 ) -> LETTER_FWD_REF:
3330 """
3331 Creates a letter to a researcher giving approval to contact a patient.
3333 Args:
3334 contact_request: a :class:`ContactRequest`
3336 Returns:
3337 a :class:`Letter`
3339 """
3340 basefilename = (
3341 f"cr{contact_request.id}_res_approve_{string_time_now()}.pdf"
3342 )
3343 html = contact_request.get_approval_letter_html()
3344 # noinspection PyTypeChecker
3345 return cls.create(
3346 basefilename,
3347 html=html,
3348 to_researcher=True,
3349 study=contact_request.study,
3350 contact_request=contact_request,
3351 rdbm_may_view=True,
3352 )
3354 @classmethod
3355 def create_researcher_withdrawal(
3356 cls, contact_request: ContactRequest
3357 ) -> LETTER_FWD_REF:
3358 """
3359 Creates a letter to a researcher withdrawing previous approval to
3360 contact a patient.
3362 Args:
3363 contact_request: a :class:`ContactRequest`
3365 Returns:
3366 a :class:`Letter`
3368 """
3369 basefilename = (
3370 f"cr{contact_request.id}_res_withdraw_{string_time_now()}.pdf"
3371 )
3372 html = contact_request.get_withdrawal_letter_html()
3373 # noinspection PyTypeChecker
3374 return cls.create(
3375 basefilename,
3376 html=html,
3377 to_researcher=True,
3378 study=contact_request.study,
3379 contact_request=contact_request,
3380 rdbm_may_view=True,
3381 )
3383 @classmethod
3384 def create_request_to_patient(
3385 cls, contact_request: ContactRequest, rdbm_may_view: bool = False
3386 ) -> LETTER_FWD_REF:
3387 """
3388 Creates a letter to a patient asking them about a specific study.
3390 Args:
3391 contact_request: a :class:`ContactRequest`
3392 rdbm_may_view: is this a request that the Research Database
3393 Manager (RDBM) is allowed to see under our information
3394 governance rules?
3396 Returns:
3397 a :class:`Letter`
3399 """
3400 basefilename = f"cr{contact_request.id}_to_pt_{string_time_now()}.pdf"
3401 pdf = contact_request.get_clinician_pack_pdf()
3402 # noinspection PyTypeChecker
3403 letter = cls.create(
3404 basefilename,
3405 pdf=pdf,
3406 to_patient=True,
3407 study=contact_request.study,
3408 contact_request=contact_request,
3409 rdbm_may_view=rdbm_may_view,
3410 )
3411 if not rdbm_may_view:
3412 # Letter is from clinician directly; clinician will print
3413 letter.mark_sent()
3414 return letter
3416 @classmethod
3417 def create_consent_confirmation_to_patient(
3418 cls, consent_mode: ConsentMode
3419 ) -> LETTER_FWD_REF:
3420 """
3421 Creates a letter to a patient confirming their traffic-light
3422 consent-mode choice.
3424 Args:
3425 consent_mode: a :class:`ConsentMode`
3427 Returns:
3428 a :class:`Letter`
3430 """
3431 basefilename = f"cm{consent_mode.id}_to_pt_{string_time_now()}.pdf"
3432 html = consent_mode.get_confirm_traffic_to_patient_letter_html()
3433 return cls.create(
3434 basefilename, html=html, to_patient=True, rdbm_may_view=True
3435 )
3437 def mark_sent(self) -> None:
3438 """
3439 Mark the letter as having been sent now.
3440 """
3441 self.sent_manually_at = timezone.now()
3442 self.save()
3445# noinspection PyUnusedLocal
3446@receiver(models.signals.post_delete, sender=Letter)
3447def auto_delete_letter_files_on_delete(
3448 sender: Type[Letter], instance: Letter, **kwargs: Any
3449) -> None:
3450 """
3451 Django signal receiver.
3453 Deletes files from filesystem when a :class:`Letter` object is deleted.
3454 """
3455 auto_delete_files_on_instance_delete(instance, ["pdf"])
3458# noinspection PyUnusedLocal
3459@receiver(models.signals.pre_save, sender=Letter)
3460def auto_delete_letter_files_on_change(
3461 sender: Type[Letter], instance: Letter, **kwargs: Any
3462) -> None:
3463 """
3464 Django signal receiver.
3466 Deletes files from filesystem when a :class:`Letter` object is changed.
3467 """
3468 auto_delete_files_on_instance_change(instance, ["pdf"], Letter)
3471# =============================================================================
3472# Record of sent e-mails
3473# =============================================================================
3476def _get_default_email_sender() -> str:
3477 """
3478 Returns the default e-mail sender.
3480 Using a callable, ``default=_get_default_email_sender``, rather than a
3481 value, ``default=settings.EMAIL_SENDER``, makes the Django migration system
3482 stop implementing pointless changes when local settings change.
3484 See
3485 https://docs.djangoproject.com/en/2.1/ref/models/fields/#django.db.models.Field.default
3486 """
3487 return settings.EMAIL_SENDER
3490class Email(models.Model):
3491 """
3492 Represents an e-mail sent (or to be sent) from CRATE.
3493 """
3495 # Let's not record host/port/user. It's configured into the settings.
3496 created_at = models.DateTimeField(
3497 verbose_name="When created", auto_now_add=True
3498 )
3499 sender = models.CharField(
3500 max_length=255, default=_get_default_email_sender
3501 )
3502 recipient = models.CharField(max_length=255)
3503 subject = models.CharField(max_length=255)
3504 msg_text = models.TextField()
3505 msg_html = models.TextField()
3506 # Other flags and links:
3507 to_clinician = models.BooleanField(default=False)
3508 to_researcher = models.BooleanField(default=False)
3509 to_patient = models.BooleanField(default=False)
3510 study = models.ForeignKey(Study, on_delete=models.PROTECT, null=True)
3511 contact_request = models.ForeignKey(
3512 ContactRequest, on_delete=models.PROTECT, null=True
3513 )
3514 letter = models.ForeignKey(Letter, on_delete=models.PROTECT, null=True)
3515 # Transmission attempts are in EmailTransmission.
3516 # Except that filtering in the admin
3518 def __str__(self) -> str:
3519 return f"Email {self.id} to {self.recipient}"
3521 @classmethod
3522 def create_clinician_email(
3523 cls, contact_request: ContactRequest
3524 ) -> EMAIL_FWD_REF:
3525 """
3526 Creates an e-mail to a clinician, asking them to consider a request
3527 from a study about a patient.
3529 Args:
3530 contact_request: a :class:`ContactRequest`
3532 Returns:
3533 an :class:`Email`
3535 """
3536 recipient = contact_request.clinician_email
3537 # noinspection PyUnresolvedReferences
3538 subject = (
3539 "RESEARCH REQUEST on behalf of {researcher}, contact request "
3540 "code {contact_req_code}".format(
3541 researcher=contact_request.study.lead_researcher.profile.get_title_forename_surname(), # noqa: E501
3542 contact_req_code=contact_request.id,
3543 )
3544 )
3545 html = contact_request.get_clinician_email_html()
3546 email = cls(
3547 recipient=recipient,
3548 subject=subject,
3549 msg_html=html,
3550 study=contact_request.study,
3551 contact_request=contact_request,
3552 to_clinician=True,
3553 )
3554 email.save()
3555 return email
3557 @classmethod
3558 def create_clinician_initiated_cr_email(
3559 cls, contact_request: ContactRequest
3560 ) -> EMAIL_FWD_REF:
3561 """
3562 Creates an e-mail to a clinician when they have initiated a contact
3563 request. This email will give them a link to the clinician pack if
3564 they said they'd contact the patient.
3566 Args:
3567 contact_request: a :class:`ContactRequest`
3569 Returns:
3570 an :class:`Email`
3572 """
3573 recipient = contact_request.clinician_email
3574 # noinspection PyUnresolvedReferences
3575 subject = (
3576 f"Confirmation of request for patient to be included in study. "
3577 f"Contact request code {contact_request.id}"
3578 )
3579 html = contact_request.get_clinician_initiated_email_html()
3580 email = cls(
3581 recipient=recipient,
3582 subject=subject,
3583 msg_html=html,
3584 study=contact_request.study,
3585 contact_request=contact_request,
3586 to_clinician=True,
3587 )
3588 email.save()
3589 return email
3591 @classmethod
3592 def create_researcher_approval_email(
3593 cls, contact_request: ContactRequest, letter: Letter
3594 ) -> EMAIL_FWD_REF:
3595 """
3596 Creates an e-mail to a researcher, enclosing a letter giving them
3597 permission to contact a patient.
3599 Args:
3600 contact_request: a :class:`ContactRequest`
3601 letter: a :class:`Letter`
3603 Returns:
3604 an :class:`Email`
3606 """
3607 # noinspection PyUnresolvedReferences
3608 recipient = contact_request.study.lead_researcher.email
3609 subject = (
3610 f"APPROVAL TO CONTACT PATIENT: contact request code "
3611 f"{contact_request.id}"
3612 )
3613 html = contact_request.get_approval_email_html()
3614 email = cls(
3615 recipient=recipient,
3616 subject=subject,
3617 msg_html=html,
3618 study=contact_request.study,
3619 contact_request=contact_request,
3620 letter=letter,
3621 to_researcher=True,
3622 )
3623 email.save()
3624 # noinspection PyTypeChecker
3625 EmailAttachment.create(
3626 email=email, fileobj=letter.pdf, content_type=ContentType.PDF
3627 ) # will save
3628 return email
3630 @classmethod
3631 def create_researcher_withdrawal_email(
3632 cls, contact_request: ContactRequest, letter: Letter
3633 ) -> EMAIL_FWD_REF:
3634 """
3635 Creates an e-mail to a researcher, enclosing a letter withdrawing their
3636 permission to contact a patient.
3638 Args:
3639 contact_request: a :class:`ContactRequest`
3640 letter: a :class:`Letter`
3642 Returns:
3643 an :class:`Email`
3645 """
3646 # noinspection PyUnresolvedReferences
3647 recipient = contact_request.study.lead_researcher.email
3648 subject = (
3649 f"WITHDRAWAL OF APPROVAL TO CONTACT PATIENT: contact request code "
3650 f"{contact_request.id}"
3651 )
3652 html = contact_request.get_withdrawal_email_html()
3653 email = cls(
3654 recipient=recipient,
3655 subject=subject,
3656 msg_html=html,
3657 study=contact_request.study,
3658 contact_request=contact_request,
3659 letter=letter,
3660 to_researcher=True,
3661 )
3662 email.save()
3663 # noinspection PyTypeChecker
3664 EmailAttachment.create(
3665 email=email, fileobj=letter.pdf, content_type=ContentType.PDF
3666 ) # will save
3667 return email
3669 @classmethod
3670 def create_rdbm_email(cls, subject: str, html: str) -> EMAIL_FWD_REF:
3671 """
3672 Create an HTML-based e-mail to the RDBM.
3674 Args:
3675 subject: subject line
3676 html: HTML body
3678 Returns:
3679 an :class:`Email`
3681 """
3682 email = cls(
3683 recipient=settings.RDBM_EMAIL, subject=subject, msg_html=html
3684 )
3685 email.save()
3686 return email
3688 @classmethod
3689 def create_rdbm_text_email(cls, subject: str, text: str) -> EMAIL_FWD_REF:
3690 """
3691 Create an text-based e-mail to the RDBM.
3693 Args:
3694 subject: subject line
3695 text: message body
3697 Returns:
3698 an :class:`Email`
3700 """
3701 email = cls(
3702 recipient=settings.RDBM_EMAIL, subject=subject, msg_text=text
3703 )
3704 email.save()
3705 return email
3707 def has_been_sent(self) -> bool:
3708 """
3709 Has this e-mail been sent?
3711 (Internally: does an :class:`EmailTransmission` for this e-mail
3712 exist with its ``sent`` flag set?)
3713 """
3714 return self.emailtransmission_set.filter(sent=True).exists()
3716 def send(
3717 self, user: settings.AUTH_USER_MODEL = None, resend: bool = False
3718 ) -> Optional[EMAIL_TRANSMISSION_FWD_REF]:
3719 """
3720 Sends the e-mail. Makes a record.
3722 Args:
3723 user: the sender.
3724 resend: say that it's OK to resend one that's already been sent.
3726 Returns:
3727 an :class:`EmailTransmission` object.
3728 """
3729 if self.has_been_sent() and not resend:
3730 log.error(f"Trying to send e-mail twice: ID={self.id}")
3731 return None
3732 if settings.SAFETY_CATCH_ON:
3733 self.recipient = settings.DEVELOPER_EMAIL
3734 try:
3735 if self.msg_html and not self.msg_text:
3736 # HTML-only email
3737 # http://www.masnun.com/2014/01/09/django-sending-html-only-email.html # noqa: E501
3738 msg = EmailMessage(
3739 subject=self.subject,
3740 body=self.msg_html,
3741 from_email=self.sender,
3742 to=[self.recipient],
3743 )
3744 msg.content_subtype = "html" # Main content is now text/html
3745 else:
3746 # Text only, or separate text/HTML
3747 msg = EmailMultiAlternatives(
3748 subject=self.subject,
3749 body=self.msg_text,
3750 from_email=self.sender,
3751 to=[self.recipient],
3752 )
3753 if self.msg_html:
3754 msg.attach_alternative(self.msg_html, "text/html")
3755 for attachment in self.emailattachment_set.all():
3756 # don't use msg.attach_file() if you want to control
3757 # the outbound filename; use msg.attach()
3758 if not attachment.file:
3759 continue
3760 path = attachment.file.path
3761 if not attachment.sent_filename:
3762 attachment.sent_filename = os.path.basename(path)
3763 attachment.save()
3764 with open(path, "rb") as f:
3765 content = f.read()
3766 msg.attach(
3767 attachment.sent_filename,
3768 content,
3769 attachment.content_type or None,
3770 )
3771 msg.send()
3772 sent = True
3773 failure_reason = ""
3774 except Exception as e:
3775 sent = False
3776 failure_reason = str(e)
3777 self.save()
3778 emailtransmission = EmailTransmission(
3779 email=self, by=user, sent=sent, failure_reason=failure_reason
3780 )
3781 emailtransmission.save()
3782 return emailtransmission
3784 def resend(self, user: settings.AUTH_USER_MODEL) -> None:
3785 """
3786 Resend this e-mail.
3787 """
3788 return self.send(user=user, resend=True)
3791EMAIL_ATTACHMENT_FWD_REF = "EmailAttachment"
3794class EmailAttachment(models.Model):
3795 """
3796 E-mail attachment class.
3798 Typically, this does NOT manage its own files (i.e. if the attachment
3799 object is deleted, the files won't be). Use this method for referencing
3800 files already stored elsewhere in the database.
3802 If the :attr:`owns_file` attribute is set, however, the associated file
3803 *is* "owned" by this object, and the file will be deleted when the database
3804 object is.
3805 """
3807 email = models.ForeignKey(Email, on_delete=models.PROTECT)
3808 file = models.FileField(storage=privatestorage)
3809 sent_filename = models.CharField(null=True, max_length=255)
3810 content_type = models.CharField(null=True, max_length=255)
3811 owns_file = models.BooleanField(default=False)
3813 def exists(self) -> bool:
3814 """
3815 Does the attached file exist on disk?
3816 """
3817 if not self.file:
3818 return False
3819 return os.path.isfile(self.file.path)
3821 def size(self) -> int:
3822 """
3823 Returns the size of the attachment in bytes, if it exists on disk
3824 (otherwise 0).
3825 """
3826 if not self.file:
3827 return 0
3828 return os.path.getsize(self.file.path)
3830 @classmethod
3831 def create(
3832 cls,
3833 email: Email,
3834 fileobj: models.FileField,
3835 content_type: str,
3836 sent_filename: str = None,
3837 owns_file=False,
3838 ) -> EMAIL_ATTACHMENT_FWD_REF:
3839 """
3840 Creates an e-mail attachment object and attaches it to an e-mail.
3841 When the e-mail is sent, the file thus referenced will be sent along
3842 with the e-mail; see :meth:`Email.send`.
3844 Args:
3845 email: an :class:`Email`, to which this attachment is attached
3846 fileobj: a :class:`django.db.models.FileField` representing the
3847 file (on disk) to be attached
3848 content_type: HTTP content type string
3849 sent_filename: name of the filename as seen within the e-mail
3850 owns_file: (see class help) Should the file on disk be deleted
3851 if/when this database object is deleted?
3853 Returns:
3854 a :class:`EmailAttachment`
3856 """
3857 if sent_filename is None:
3858 sent_filename = os.path.basename(fileobj.name)
3859 attachment = cls(
3860 email=email,
3861 file=fileobj,
3862 sent_filename=sent_filename,
3863 content_type=content_type,
3864 owns_file=owns_file,
3865 )
3866 attachment.save()
3867 return attachment
3870# noinspection PyUnusedLocal
3871@receiver(models.signals.post_delete, sender=EmailAttachment)
3872def auto_delete_emailattachment_files_on_delete(
3873 sender: Type[EmailAttachment], instance: EmailAttachment, **kwargs: Any
3874) -> None:
3875 """
3876 Django signal receiver.
3878 Deletes files from filesystem when :class:`EmailAttachment` object is
3879 deleted, if its :attr:`owns_file` flag is set.
3880 """
3881 if instance.owns_file:
3882 auto_delete_files_on_instance_delete(instance, ["file"])
3885# noinspection PyUnusedLocal
3886@receiver(models.signals.pre_save, sender=EmailAttachment)
3887def auto_delete_emailattachment_files_on_change(
3888 sender: Type[EmailAttachment], instance: EmailAttachment, **kwargs: Any
3889) -> None:
3890 """
3891 Django signal receiver.
3893 Deletes files from filesystem when :class:`EmailAttachment` object is
3894 changed, if its :attr:`owns_file` flag is set.
3895 """
3896 if instance.owns_file:
3897 auto_delete_files_on_instance_change(
3898 instance, ["file"], EmailAttachment
3899 )
3902class EmailTransmission(models.Model):
3903 """
3904 Represents an e-mail transmission attempt.
3905 """
3907 email = models.ForeignKey(Email, on_delete=models.PROTECT)
3908 at = models.DateTimeField(verbose_name="When sent", auto_now_add=True)
3909 by = models.ForeignKey(
3910 settings.AUTH_USER_MODEL,
3911 on_delete=models.PROTECT,
3912 null=True,
3913 related_name="emailtransmissions",
3914 )
3915 sent = models.BooleanField(default=False)
3916 failure_reason = models.TextField(verbose_name="Reason sending failed")
3918 def __str__(self) -> str:
3919 return "Email transmission at {} by {}: {}".format(
3920 self.at,
3921 self.by or "(system)",
3922 "success" if self.sent else f"failure: {self.failure_reason}",
3923 )
3926# =============================================================================
3927# A dummy set of objects, for template testing.
3928# Linked, so cross-references work.
3929# Don't save() them!
3930# =============================================================================
3933class DummyObjectCollection:
3934 """
3935 A collection of dummy objects within the consent-to-contact system, for
3936 testing templates.
3937 """
3939 def __init__(
3940 self,
3941 contact_request: ContactRequest,
3942 consent_mode: ConsentMode,
3943 patient_lookup: PatientLookup,
3944 study: Study,
3945 clinician_response: ClinicianResponse,
3946 ):
3947 self.contact_request = contact_request
3948 self.consent_mode = consent_mode
3949 self.patient_lookup = patient_lookup
3950 self.study = study
3951 self.clinician_response = clinician_response
3954def make_dummy_objects(
3955 request: HttpRequest, test_id: str = TEST_ID_STR
3956) -> DummyObjectCollection:
3957 """
3958 Returns a collection of dummy objects, for testing consent-to-contact
3959 templates without using live patient data.
3961 Args:
3962 request: the :class:`django.http.request.HttpRequest`
3964 Returns:
3965 a :class:`DummyObjectCollection`
3967 We want to create these objects in memory, without saving to the DB.
3968 However, Django is less good at SQLAlchemy for this, and saves.
3970 - https://stackoverflow.com/questions/7908349/django-making-relationships-in-memory-without-saving-to-db # noqa: E501
3971 - https://code.djangoproject.com/ticket/17253
3972 - https://stackoverflow.com/questions/23372786/django-models-assigning-foreignkey-object-without-saving-to-database # noqa: E501
3973 - https://stackoverflow.com/questions/7121341/django-adding-objects-to-a-related-set-without-saving-to-db # noqa: E501
3975 A simple method works for an SQLite backend database but fails with
3976 an IntegrityError for MySQL/SQL Server. For example:
3978 .. code-block:: none
3980 IntegrityError at /draft_traffic_light_decision_form/-1/html/
3981 (1452, 'Cannot add or update a child row: a foreign key constraint
3982 fails (`crate_django`.`consent_study_researchers`, CONSTRAINT
3983 `consent_study_researchers_study_id_19bb255f_fk_consent_study_id`
3984 FOREIGN KEY (`study_id`) REFERENCES `consent_study` (`id`))')
3986 This occurs in the first creation, of a :class:`Study`, and only if you
3987 specify ``researchers``.
3989 The reason for the crash is that ``researchers`` is a ManyToManyField, and
3990 Django is trying to set the ``user.studies_as_researcher`` back-reference,
3991 but can't do so because the :class:`Study` doesn't have a PK yet.
3993 Since this is a minor thing, and templates are unaffected, and this is only
3994 for debugging, let's ignore it.
3995 """
3996 using_alt = test_id == TEST_ID_TWO_STR
3998 def get_int(query_param_name: str, default: Optional[int]) -> int:
3999 try:
4000 # noinspection PyCallByClass,PyTypeChecker
4001 return int(request.GET.get(query_param_name, default))
4002 except (TypeError, ValueError):
4003 return default
4005 def get_str(query_param_name: str, default: Optional[str]) -> str:
4006 # noinspection PyCallByClass,PyTypeChecker
4007 return request.GET.get(query_param_name, default)
4009 age = get_int("age", 13 if using_alt else 40)
4010 age_months = get_int("age_months", 2)
4011 today = datetime.date.today()
4012 dob = today - relativedelta(years=age, months=age_months)
4014 consent_mode_str = get_str(
4015 "consent_mode", ConsentMode.YELLOW if using_alt else None
4016 )
4017 if consent_mode_str not in (
4018 None,
4019 ConsentMode.RED,
4020 ConsentMode.YELLOW,
4021 ConsentMode.GREEN,
4022 ):
4023 consent_mode_str = None
4025 request_direct_approach = bool(get_int("request_direct_approach", 1))
4026 clinician_involvement = ContactRequest.get_clinician_involvement(
4027 consent_mode_str=consent_mode_str,
4028 request_direct_approach=request_direct_approach,
4029 )
4031 consent_after_discharge = bool(get_int("consent_after_discharge", 0))
4033 nhs_number = 2345678901 if using_alt else 1234567890
4034 study_summary_plaintext = (
4035 "An investigation of the change in blood-oxygen-level-"
4036 "dependent (BOLD) functional magnetic resonance imaging "
4037 "(fMRI) signals during the experience of quaint and "
4038 "fanciful humorous activity. "
4039 "(Incorrectly marked as a CTIMP for illustration only.)"
4040 # "\n"
4041 # "\n"
4042 # "This is paragraph 2.\n"
4043 # "\n"
4044 # "For patients aged >18 and <65."
4045 )
4046 study_summary_html = """
4047 <p>An investigation of the change in <b>blood-oxygen-level-dependent
4048 (BOLD)</b> <i>functional magnetic resonance imaging (fMRI)</i> signals
4049 during the experience of quaint and fanciful humourous activity.</p>
4050 """
4051 # """
4052 #
4053 # <p>Now with extra HTML.</p>
4054 #
4055 # <p>For patients aged >18 and <65.</p>
4056 # """
4057 use_html = False
4058 User = get_user_model()
4059 lead_researcher_profile = UserProfile()
4060 lead_researcher_profile.title = "Prof."
4061 lead_researcher_user = User()
4062 lead_researcher_user.first_name = "Gabrielle"
4063 lead_researcher_user.last_name = "Gnosis"
4064 lead_researcher_user.profile = lead_researcher_profile
4065 study = Study(
4066 id=TEST_ID,
4067 institutional_id=9999999999999,
4068 title="Functional neuroimaging of whimsy",
4069 lead_researcher=lead_researcher_user,
4070 # lead_researcher=request.user,
4071 # researchers=[request.user], # THIS BREAKS IT.
4072 # ... actual crash is in
4073 # django/db/models/fields/related_descriptors.py:500, in
4074 # ReverseManyToOneDescriptor.__set__(), calling
4075 # manager.set(value)
4076 registered_at=datetime.datetime.now(),
4077 summary=study_summary_html if use_html else study_summary_plaintext,
4078 summary_is_html=use_html,
4079 search_methods_planned="Generalized trawl",
4080 patient_contact=True,
4081 include_under_16s=True,
4082 include_lack_capacity=True,
4083 clinical_trial=True,
4084 request_direct_approach=clinician_involvement,
4085 approved_by_rec=True,
4086 rec_reference="blah/999",
4087 approved_locally=True,
4088 local_approval_at=True,
4089 study_details_pdf=None,
4090 subject_form_template_pdf=None,
4091 )
4092 consent_mode = ConsentMode(
4093 id=TEST_ID,
4094 nhs_number=nhs_number,
4095 current=True,
4096 created_by=request.user,
4097 exclude_entirely=False,
4098 consent_mode=consent_mode_str,
4099 consent_after_discharge=consent_after_discharge,
4100 max_approaches_per_year=0,
4101 other_requests="",
4102 prefers_email=False,
4103 changed_by_clinician_override=False,
4104 source="Fictional",
4105 )
4106 patient_lookup = PatientLookup(
4107 id=TEST_ID,
4108 # PatientLookupBase
4109 pt_local_id_description="CPFT#",
4110 pt_local_id_number=987654 if using_alt else 876543,
4111 pt_dob=dob,
4112 pt_dod=None,
4113 pt_dead=False,
4114 pt_discharged=False,
4115 pt_discharge_date=None,
4116 pt_sex=(
4117 PatientLookupBase.FEMALE if using_alt else PatientLookupBase.MALE
4118 ),
4119 pt_title="Miss" if using_alt else "Mr",
4120 pt_first_name="Jane" if using_alt else "John",
4121 pt_last_name="Smith",
4122 pt_address_1="The Farthings",
4123 pt_address_2="1 Penny Lane",
4124 pt_address_3="Mordenville",
4125 pt_address_4="Slowtown",
4126 pt_address_5="Cambridgeshire",
4127 pt_address_6="CB1 0ZZ",
4128 pt_address_7="UK",
4129 pt_telephone="01223 000000",
4130 pt_email="jane@smith.com" if using_alt else "john@smith.com",
4131 gp_title="Dr",
4132 gp_first_name="Gordon",
4133 gp_last_name="Generalist",
4134 gp_address_1="Honeysuckle Medical Practice",
4135 gp_address_2="99 Bloom Street",
4136 gp_address_3="Mordenville",
4137 gp_address_4="Slowtown",
4138 gp_address_5="Cambridgeshire",
4139 gp_address_6="CB1 9QQ",
4140 gp_address_7="UK",
4141 gp_telephone="01223 111111",
4142 gp_email="g.generalist@honeysuckle.nhs.uk",
4143 clinician_title="Dr",
4144 clinician_first_name="Petra",
4145 clinician_last_name="Paroxetine",
4146 clinician_address_1="Union House",
4147 clinician_address_2="37 Union Lane",
4148 clinician_address_3="Chesterton",
4149 clinician_address_4="Cambridge",
4150 clinician_address_5="Cambridgeshire",
4151 clinician_address_6="CB4 1PR",
4152 clinician_address_7="UK",
4153 clinician_telephone="01223 222222",
4154 clinician_email="p.paroxetine@cpft_or_similar.nhs.uk",
4155 clinician_is_consultant=True,
4156 clinician_signatory_title="Consultant psychiatrist",
4157 # PatientLookup
4158 nhs_number=nhs_number,
4159 source_db="Fictional database",
4160 decisions="No real decisions",
4161 secret_decisions="No real secret decisions",
4162 pt_found=True,
4163 gp_found=True,
4164 clinician_found=True,
4165 )
4166 contact_request = ContactRequest(
4167 id=TEST_ID,
4168 request_by=request.user,
4169 study=study,
4170 lookup_rid=9999999,
4171 created_at=timezone.now(),
4172 processed=True,
4173 nhs_number=nhs_number,
4174 patient_lookup=patient_lookup,
4175 consent_mode=consent_mode,
4176 approaches_in_past_year=0,
4177 decisions="No decisions required",
4178 decided_no_action=False,
4179 # decided_send_to_researcher=False,
4180 decided_send_to_researcher=True,
4181 decided_send_to_clinician=True,
4182 clinician_involvement=clinician_involvement,
4183 consent_withdrawn=False,
4184 consent_withdrawn_at=None,
4185 )
4186 clinician_response = ClinicianResponse(
4187 id=TEST_ID,
4188 contact_request=contact_request,
4189 token="dummytoken",
4190 responded=False,
4191 )
4193 return DummyObjectCollection(
4194 contact_request=contact_request,
4195 consent_mode=consent_mode,
4196 patient_lookup=patient_lookup,
4197 study=study,
4198 clinician_response=clinician_response,
4199 )