penracourses/generic/models.py

from collections import defaultdict
from django.contrib.auth.mixins import LoginRequiredMixin
from django.db import models
from django.http import HttpRequest
from django.utils import timezone
from smtplib import SMTPException

from django.urls import reverse
from django.views.generic.edit import CreateView
from reversion.views import RevisionMixin

import tagulous
import tagulous.models

from sortedm2m.fields import SortedManyToManyField
from django.conf import settings

from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType

from django.core.mail import send_mail
from django.core.mail import EmailMessage

from django.contrib.auth.models import User

from django.db.models.signals import post_save
from django.dispatch import receiver

class Plane(models.Model):
    plane = models.CharField(max_length=200, unique=True)

    def __str__(self):
        return self.plane

    class Meta:
        ordering = ("plane",)


class Contrast(models.Model):
    contrast = models.CharField(max_length=200, unique=True)

    def __str__(self):
        return self.contrast

    class Meta:
        ordering = ("contrast",)


class Examination(models.Model):
    examination = models.CharField(max_length=200, unique=True)

    def __str__(self):
        return self.examination

    class Meta:
        ordering = ("examination",)


class Site(models.Model):
    """Model to hold site details"""
    full_name = models.CharField(max_length=255, blank=True, help_text="Name of the site")
    short_code = models.CharField(max_length=255, blank=True, help_text="Shortcode/name of the site")


    def __str__(self):
        return self.short_code


class Condition(tagulous.models.TagModel):
    pass


class Sign(tagulous.models.TagModel):
    pass

class QuestionBase(models.Model):

    authors_only = models.BooleanField(
        help_text="If true only question authors will be able to view.",
        default=False,
    )

    class Meta:
        abstract = True


class ExamBase(models.Model):
    name = models.CharField(max_length=200, help_text="Name of the exam")
    # exam_questions = SortedManyToManyField(Long, related_name="exams", blank="true")

    active = models.BooleanField(
        help_text="If an exam should be available to take", default=False
    )

    exam_mode = models.BooleanField(
        help_text="If an exam should be taken in exam mode (users results will be submited to the server for marking)",
        default=False,
    )

    include_history = models.BooleanField(
        help_text="If an exam should include history when taking",
        default=False,
    )

    publish_results = models.BooleanField(
        help_text="If an exams results should be available", default=False
    )

    recreate_json = models.BooleanField(
        help_text="If the json cache needs updating", default=False
    )

    json_creation_time = models.DateTimeField(blank=True, default=None, null=True)
    exam_json_id = models.IntegerField(
        default=1, help_text="auto incrementing field when json recreated"
    )

    archive = models.BooleanField(
        help_text="Archived exams will remain on the test system but will not be displayed by default",
        default=False,
    )

    open_access = models.BooleanField(
        help_text="If the exam is freely accessible (to view and edit on the test system)",
        default=False,
    )

    authors_only = models.BooleanField(
        help_text="If true only exam authors will be able to view.",
        default=False,
    )

    stats_mean = models.FloatField(default=0)
    stats_mode = models.CharField(default=0, max_length=25)
    stats_median = models.FloatField(default=0)

    stats_candidates = models.FloatField(default=0)

    stats_min = models.FloatField(default=0)
    stats_max = models.FloatField(default=0)

    stats_max_possible = models.FloatField(default=0)

    #stats_graph = models.TextField(default=0)

    user_scores = models.JSONField(default=dict)

    # Is this actually used?
    cid_users = GenericRelation("generic.CidUserExam")

    exam_results_emailed = models.DateTimeField(default=None, null=True)


    class Meta:
        abstract = True

    def save(self, *args, recreate_json=True, **kwargs):
        self.recreate_json = recreate_json
        super().save(*args, **kwargs)

    def get_exam_stats(self, name=True):
        if self.stats_candidates < int(4):
            return f"- Candidates: {int(self.stats_candidates)} (too few to generate stats)"

        text = f"""- Candidates: {int(self.stats_candidates)}
- Mean: {self.stats_mean:.2f} Median: {self.stats_median} Mode: {self.stats_mode} [Min: {self.stats_min} / Max: {self.stats_max}]
"""
        if name:
            text = f"""{self.name}
{"-"*len(self.name)}
{text}
"""
        return text

    def __str__(self):
        return self.name

    def get_absolute_url(self):
        return reverse("{}:exam_overview".format(self.app_name), kwargs={"pk": self.pk})

    def get_take_url(self):
        return f"{settings.REMOTE_URL}/rts"

    def get_exam_name(self):
        return self.name

    def get_json_url(self):
        return reverse("{}:exam_json".format(self.app_name), args=(self.pk,))

    def get_question_index(self, question):
        return list(self.exam_questions.all()).index(question)

    def get_authors(self):
        """Returns a comma seperated text list of authors"""
        authors = ", ".join([i.username for i in self.author.all()])
        return authors

    def get_author_objects(self):
        """Returns a comma seperated text list of authors"""
        authors = [i for i in self.author.all()]
        return authors

    def check_cid_user(self, cid, passcode, request=None, user_id=None):
        if request is not None and request.user.is_superuser:
            return True

        if not self.valid_cid_users.exists() and not self.valid_user_users.exists():
            return True

        if user_id is not None:
            if self.valid_user_users.filter(pk=user_id).exists():
                return True

        if self.valid_cid_users.exists():

            cid_user = self.valid_cid_users.filter(cid=cid).first()

            if cid_user and cid_user.passcode == passcode:
                return True

            return False

        return False

    def get_or_create_cid_user_exam(self, cid_user=None, user_user=None, start_time=None):
        content_type = ContentType.objects.get_for_model(self)

        if cid_user is not None:
            c = CidUserExam.objects.filter(
                content_type=content_type, object_id=self.pk, cid_user=cid_user
            ).first()
            if c:
                return c

        if user_user is not None:
            c = CidUserExam.objects.filter(
                content_type=content_type, object_id=self.pk, user_user=user_user
            ).first()
            if c:
                return c


        if start_time is None:
            start_time = timezone.now()


        if cid_user is not None:
            new = CidUserExam(
                content_type=content_type,
                object_id=self.pk,
                cid_user=cid_user,
                start_time=start_time,
            )

        if user_user is not None:
            new = CidUserExam(
                content_type=content_type,
                object_id=self.pk,
                user_user=user_user,
                start_time=start_time,
            )

        new.save()
        return new

    def get_cid_user_exams(self, cid_user=None, user_user=None):
        content_type = ContentType.objects.get_for_model(self)
        if cid_user is None and user_user is None:
            return CidUserExam.objects.filter(
                content_type=content_type,
                object_id=self.pk,
            )
        else:
            if cid_user is not None:
                return CidUserExam.objects.filter(
                    content_type=content_type, object_id=self.pk, cid_user=cid_user
                )
            else:
                return CidUserExam.objects.filter(
                    content_type=content_type, object_id=self.pk, user_user=user_user
                )

    def get_cid_user_score(self, cid_user):
        c = "c/"+str(cid_user)
        if c in self.user_scores:
            return self.user_scores[c]
        else:
            return False

    def generate_user_report(self, user):
        exam_text = [f"Candidate {user.first_name} [{user.email}]"]

        exam_name = self.name

        exam_text.append(f"{'='*len(exam_name)}\n{exam_name}\n{'='*len(exam_name)}")

        try:
            user_score = self.user_scores["u/"+str(user.pk)]["score"]

            if self.app_name == "rapids":
                callstate = self.user_scores["u/"+str(user.pk)]["callstates"]
        except KeyError:
            user_score = "Score not generated"

        try:
            percentage = (
                f" ({user_score / int(self.stats_max_possible) * 100:.2f}%)"
            )
        except TypeError:
            percentage = ""

        
        exam_text.append(
                f"""Results\n{'-'*len('Results')}
# Score: {user_score} / {int(self.stats_max_possible)}{percentage}
"""
            )
        stats = self.get_exam_stats(name=False)
        exam_text.append(f"Stats\n{'-'*len('Stats')}\n{stats}")


        if self.app_name == "rapids":
            exam_text.append(
                f"Answers breakdown: {callstate}"
            )

        msg = "\n\n".join(exam_text)

        return msg

    def email_user_results(self, user, resend=False, additional_emails=None):
        #if self.results_email_sent and not resend:
        #    return False, "Already sent."

        # Get a list of taken exams
        msg = self.generate_user_report(user)

        if not user.email:
            return [False, "User has no email"]

        emails = [user.email]

        supervisor_email = user.userprofile.supervisor.email

        if supervisor_email:
            emails.append(supervisor_email)


        if additional_emails is not None:
            emails.extend(additional_emails)

        try:
            send_mail(
                f"{self.name} results",
                msg,
                "no-reply@penracourses.org.uk",
                emails,
                fail_silently=False,
                # html_message=html_msg,
            )
        except SMTPException as e:
            return [False, e]

        #self.results_email_sent = True
        #self.save()
        return [True, ""]


class NoteType(models.Model):
    note_type = models.CharField(max_length=200)

    def __str__(self):
        return self.note_type


class QuestionNote(models.Model):
    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
    object_id = models.PositiveIntegerField()
    question = GenericForeignKey("content_type", "object_id")

    author = models.CharField(max_length=200, blank=True)

    user_author = models.ForeignKey(
        settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True
    )

    note = models.TextField(blank=True)
    created_on = models.DateTimeField(auto_now_add=True)

    complete = models.BooleanField(default=False)

    note_type = models.ForeignKey(
        NoteType, on_delete=models.CASCADE, null=True, blank=True
    )

    # def get_absolute_url(self):
    #    self.pk = self.rapid_id
    #    return reverse('rapids:question_detail', kwargs={'pk': self.pk})

    def __str__(self):
        return "{}: {} [{}] {} / {}".format(
            self.content_type,
            self.get_author_str(),
            self.created_on,
            self.note_type,
            self.note,
        )

    def get_absolute_url(self):
        # TODO: return to question if logged in
        return "/"

    def get_object_url(self):
        return self.question.get_absolute_url()

    def get_author_str(self):
        if self.user_author is not None:
            return self.user_author.username
        else:
            return self.author

    def get_short_str(self):
        return f"{self.note_type} - {self.note}"

EXAM_TYPES = (
    ("physics", "physics_exams"),
    ("rapids", "rapid_exams"),
    ("sbas", "sba_exams"),
    ("anatomy", "anatomy_exams"),
    ("longs", "longs_exams"),
    ("casecollection", "casecollection_exams"),
)

class CidUser(models.Model):

    cid = models.IntegerField(blank=True, null=True, unique=True)
    passcode = models.CharField(blank=True, max_length=50)
    active = models.BooleanField(default=True)
    internal_candidate = models.BooleanField(default=False)
    name = models.CharField(blank=True, max_length=255)
    email = models.EmailField(blank=True)
    supervisor = models.ForeignKey("Supervisor", on_delete=models.SET_NULL, blank=True, null=True, related_name="cid_user")

    login_email_sent = models.BooleanField(default=False)
    results_email_sent = models.BooleanField(default=False)

    group = models.ForeignKey("CidUserGroup", on_delete=models.SET_NULL, null=True, blank=True)

    class Meta:
        indexes = [ 
            models.Index(fields=["cid"])
        ]

    def __str__(self) -> str:
        return str(self.cid)

    def get_absolute_url(self):
        return reverse("generic:manage_cids")

    def check_passcode(self, passcode) -> bool:
        return self.passcode == passcode

    def get_cid_exams(self):
        available_exams = []

        for n, t in EXAM_TYPES:
            exam_rel = getattr(self, t)
            if exam_rel.exists():
                exams = exam_rel.filter(exam_mode=True, archive=False).order_by("name")
                available_exams.append((n, exams))

        return available_exams

    def generate_exam_report(self):
        exam_list = self.get_cid_exams()

        name = ""
        if self.name:
            name = f"{self.name} "

        exam_text = [f"Candidate {self.cid} {name}[{self.email}]"]

        for exam_type, exams in exam_list:
            exam_text.append(f"{'='*len(exam_type)}\n{exam_type}\n{'='*len(exam_type)}")

            for exam in exams:
                try:
                    user_score = exam.user_scores["c/"+str(self.cid)]
                except KeyError:
                    user_score = "Score not generated"
                t = exam.get_exam_stats()

                try:
                    percentage = (
                        f" ({user_score / int(exam.stats_max_possible) * 100:.2f}%)"
                    )
                except TypeError:
                    percentage = ""

                exam_text.append(
                    f"""{t}
# Score: {user_score} / {int(exam.stats_max_possible)}{percentage}
"""
                )

        msg = "\n\n".join(exam_text)

        return msg

        html_msg = f""""""

    def email_results(self, resend=False, additional_emails=[]):
        if not self.internal_candidate:
            return False, "Not internal candidate."

        if self.results_email_sent and not resend:
            return False, "Already sent."

        # Get a list of taken exams
        msg = self.generate_exam_report()

        emails = [self.email, self.supervisor.email]

        if additional_emails:
            emails.extend(additional_emails)

        try:
            send_mail(
                f"{self.group} results",
                msg,
                "test@xkjq.uk",
                emails,
                fail_silently=False,
                # html_message=html_msg,
            )
        except SMTPException as e:
            return False, e

        self.results_email_sent = True
        self.save()
        return True, ""

    def email_details(self, resend=False):
        if self.login_email_sent and not resend:
            return False, "Already sent."

        login_url = reverse("cid_selector")

        msg = f"""
        See below for your details for: {self.group}

        ---------
        CID: {self.cid}
        Passcode: {self.passcode}
        ---------

        You will need these to access to the test system.

        Direct link to login:
        {settings.REMOTE_URL}{login_url}?cid={self.cid}&passcode={self.passcode}

        The above details are unique to you. Please do not share as otherwise others will be able to access and submit your results.



        Please note this inbox is not monitored. Direct all enquiries to plh-tr.radiology-academy@nhs.net
        """

        try:
            # send_mail(
            #    "Candidate Details",
            #    msg,
            #    "test@xkjq.uk",
            #    [self.email],
            #    fail_silently=False,
            # )
            email = EmailMessage(
                "Candidate Details",
                msg,
                "no-reply@penracourses.org.uk",
                [self.email],
                #["ross.kruger@nhs.net"],
            )
            email.send(fail_silently=False)
        except SMTPException as e:
            return False, e

        self.login_email_sent = True
        self.save()
        return True, ""

    def get_email_details_url(self):
        return reverse("generic:candidate_email_details", kwargs={"cid": self.cid})

    def get_email_results_url(self):
        return reverse("generic:candidate_email_results", kwargs={"cid": self.cid})

    def get_email_results_resend_url(self):
        return reverse("generic:candidate_email_results_resend", kwargs={"cid": self.cid})

def get_next_cid():
    """Returns the next available CID"""
    return CidUser.objects.order_by("-cid").first().cid + 1



class CidUserExam(models.Model):
    content_type = models.ForeignKey(ContentType, on_delete=models.CASCADE)
    object_id = models.PositiveIntegerField()
    exam = GenericForeignKey("content_type", "object_id")

    start_time = models.DateTimeField(blank=True, null=True)
    end_time = models.DateTimeField(blank=True, null=True)

    cid_user = models.ForeignKey(CidUser, on_delete=models.CASCADE, blank=True, null=True, related_name="cid_userexam")
    user_user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True, related_name="user_userexam")

    manual_submission = models.BooleanField(default=False)

    # TODO switch to json field?
    results_emailed_status = models.CharField(max_length=255, blank=True)

CID_GROUP_EXAMS = (
    ("SBAs", "sba_cid_user_groups"),
    ("Physics", "physics_cid_user_groups"),
    ("Anatomy", "anatomy_cid_user_groups"),
    ("Rapids", "rapid_cid_user_groups"),
    ("Longs", "longs_cid_user_groups"),
)

class CidUserGroup(models.Model):
    name = models.CharField(blank=True, max_length=50)
    archive = models.BooleanField(default=False)

    def __str__(self) -> str:
        return str(self.name)

    def GetGroupUsers(self):
        cid_users = self.ciduser_set.all()

        s = ", ".join([f"{user.id} - {user.name}" for user in cid_users])
        return s

    def GetGroupExams(self):
        exams = defaultdict(list)
        for t, rel in CID_GROUP_EXAMS:
            exam_rel = getattr(self, rel)
            if exam_rel.exists():
                exams[t].extend(exam_rel.all())

        return dict(exams)


    def get_absolute_url(self):
        return reverse("generic:cid_group_update", kwargs={"pk": self.pk})

    
USER_GROUP_EXAMS = (
    ("SBAs", "sba_user_user_groups"),
    ("Physics", "physics_user_user_groups"),
    ("Anatomy", "anatomy_user_user_groups"),
    ("Rapids", "rapid_user_user_groups"),
    ("Longs", "longs_user_user_groups"),
)

class UserUserGroup(models.Model):
    name = models.CharField(blank=True, max_length=50, help_text="Name of the User Group")
    archive = models.BooleanField(default=False, help_text="Archived groups remain on the test system but are not displayed by default")

    users = models.ManyToManyField(
        settings.AUTH_USER_MODEL,
        blank=True,
        related_name="user_groups",
    )

    def __str__(self) -> str:
        return str(self.name)

    def GetGroupUsers(self):
        users = self.users.all()

        s = ", ".join([f"{user.id} - {user.username} ({user.email})" for user in users])
        return s

    def GetGroupExams(self):
        exams = defaultdict(list)
        for t, rel in USER_GROUP_EXAMS:
            exam_rel = getattr(self, rel)
            if exam_rel.exists():
                exams[t].extend(exam_rel.all())

        return dict(exams)

    def get_absolute_url(self):
        return reverse("generic:user_group_update", kwargs={"pk": self.pk})
    
class UserGrades(models.Model):
    name = models.CharField(max_length=100)

    def __str__(self):
        return self.name

USER_EXAM_TYPES = (
    ("Physics", "user_physics_exams"),
    ("Rapids", "user_rapid_exams"),
    ("SBAs", "user_sba_exams"),
    ("Anatomy", "user_anatomy_exams"),
    ("Longs", "user_longs_exams"),
    #("CaseCollection", "user_casecollection_exams"),
)

class UserProfile(models.Model):
    user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    supervisor = models.ForeignKey("Supervisor", on_delete=models.SET_NULL, blank=True, null=True, related_name="trainee")
    registration_number = models.CharField(max_length=25, blank=True)
    grade = models.ForeignKey(UserGrades, null=True, blank=True, help_text="User grade", on_delete=models.CASCADE)
    peninsula_trainee = models.BooleanField(default=False)


    def getusername(self):
        return self.user.username

    def __str__(self):
        return f"Userprofile {self.user}"

    username = property(getusername)

    def get_exams(self):
        available_exams = defaultdict(list)

        for n, t in USER_EXAM_TYPES:
            exam_rel = getattr(self.user, t)
            if exam_rel.exists():
                exams = exam_rel.filter(exam_mode=True, archive=False).order_by("name")
                available_exams[n].extend(exams)

        return dict(available_exams)

class Supervisor(models.Model):
    """Model to hold individual supervisor details (email and name)
    
    This can be linked with a user account.
    """
    email = models.EmailField(unique=True, help_text="The (nhs.net) email address of the supervisor")
    name = models.CharField(max_length=255, help_text="Name of the supervisor")
    user = models.OneToOneField(settings.AUTH_USER_MODEL, on_delete=models.CASCADE, null=True, blank=True, help_text="If the supervisor has an account on the test system that can be associated here")
    site = models.ForeignKey("Site", on_delete=models.SET_NULL, null=True, blank=True, help_text="Hospital site at which the supervisor is based")
    active = models.BooleanField(default=True)

    def __str__(self) -> str:
        return f"{self.name} ({self.email})"

    def get_absolute_url(self):
        return reverse('generic:supervisor_detail', kwargs={'pk': self.pk})

    def clean(self):
        self.email = self.email.strip()
        self.name = self.name.strip()
    

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        UserProfile.objects.create(user=instance)

@receiver(post_save, sender=User)
def save_user_profile(sender, instance, **kwargs):
    instance.userprofile.save()