Ross
1745 lines
59 KiB
Python
1745 lines
59 KiB
Python
import secrets
|
|
from typing import Any, Optional
|
|
import os
|
|
import re
|
|
from io import StringIO
|
|
from datetime import datetime
|
|
from django.conf import settings
|
|
|
|
from django_tables2 import SingleTableMixin
|
|
from atlas.models import CaseCollection, CidReportAnswer, Case as AtlasCase
|
|
from generic.decorators import user_is_cid_user_manager
|
|
from generic.filters import UserUserFilter
|
|
from generic.tables import UserUserTable
|
|
from generic.views import (
|
|
CidManagerRequiredMixin,
|
|
RedirectMixin,
|
|
get_question_and_content_type,
|
|
get_user_exams,
|
|
get_exam_model_from_app_name,
|
|
)
|
|
from django.core.exceptions import PermissionDenied, ObjectDoesNotExist
|
|
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django import forms
|
|
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import validate_email
|
|
|
|
from django.utils.html import format_html
|
|
from django.utils.safestring import mark_safe
|
|
|
|
# from django.contrib.auth.models import User
|
|
from django.contrib.auth.decorators import login_required, user_passes_test
|
|
from django.contrib.auth.forms import PasswordResetForm
|
|
from django.contrib.auth.models import User
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
|
|
from django.views.generic.edit import CreateView, UpdateView, DeleteView
|
|
from django.views.generic import ListView, TemplateView
|
|
|
|
from django.db.models.functions import Lower
|
|
|
|
from django.core.cache import cache
|
|
|
|
from django.urls import reverse_lazy, reverse
|
|
|
|
from django.http import Http404, JsonResponse
|
|
|
|
from django.http import HttpResponseRedirect, HttpResponse
|
|
from django.views.decorators.http import require_POST
|
|
from django.core.management import call_command, CommandError
|
|
|
|
from physics.models import UserAnswer as PhysicsUserAnswer
|
|
from physics.models import Exam as PhysicsExam
|
|
|
|
from anatomy.models import UserAnswer as AnatomyUserAnswer
|
|
from anatomy.models import Exam as AnatomyExam
|
|
from anatomy.models import AnatomyQuestion
|
|
from anatomy.models import Answer as AnatomyAnswer
|
|
from rad.filters import UserListFilter
|
|
|
|
from helpers.cimar import CimarAPI, InvalidCredentials, NoSession
|
|
from rapids.models import UserAnswer as RapidsUserAnswer
|
|
from rapids.models import Exam as RapidsExam
|
|
from rapids.models import Rapid
|
|
from rapids.models import Answer as RapidAnswer
|
|
|
|
from longs.models import UserAnswer as LongsUserAnswer
|
|
from longs.models import Exam as LongsExam
|
|
from longs.models import Long
|
|
|
|
from sbas.models import UserAnswer as SbasUserAnswer
|
|
from sbas.models import Exam as SbasExam
|
|
|
|
from anatomy.views import GenericExamViews as AnatomyExamViews
|
|
from rapids.views import GenericExamViews as RapidsExamsViews
|
|
from shorts.views import GenericExamViews as ShortsExamsViews
|
|
from longs.views import GenericExamViews as LongsExamsViews
|
|
|
|
from generic.forms import QuestionNoteForm
|
|
from generic.models import (
|
|
CidUser,
|
|
CimarCase,
|
|
ExamCollection,
|
|
QuestionNote,
|
|
Supervisor,
|
|
UserGrades,
|
|
UserProfile,
|
|
USER_EXAM_TYPES,
|
|
)
|
|
from django.contrib.auth.models import Group
|
|
from generic.models import UserUserGroup
|
|
|
|
from django_filters.views import FilterView
|
|
|
|
import json
|
|
|
|
from django.template import RequestContext
|
|
|
|
from django.db.models import Q
|
|
|
|
from dal import autocomplete
|
|
|
|
from loguru import logger
|
|
from .logging_setup import get_log_file_path
|
|
|
|
import psutil
|
|
|
|
|
|
|
|
|
|
def _media_file_inventory(path: str) -> dict[str, int]:
|
|
"""Return a mapping of file path to byte size for files under a directory."""
|
|
if not path or not os.path.exists(path):
|
|
return {}
|
|
|
|
inventory: dict[str, int] = {}
|
|
for root, _, files in os.walk(path):
|
|
for filename in files:
|
|
file_path = os.path.join(root, filename)
|
|
try:
|
|
inventory[file_path] = os.path.getsize(file_path)
|
|
except OSError:
|
|
continue
|
|
|
|
return inventory
|
|
|
|
|
|
def feedback_checker(user):
|
|
return user.groups.filter(name="feedback_checker").exists()
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def server(request):
|
|
return render(request, "server.html")
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def people(request):
|
|
"""Render the people management overview or search users via HTMX.
|
|
|
|
Scope: managers and superusers.
|
|
Functionality: provides details on user roles/groups, and a single search
|
|
input that filters users dynamically via HTMX GET requests.
|
|
"""
|
|
if request.htmx:
|
|
q = request.GET.get("q", "").strip()
|
|
if not q:
|
|
return HttpResponse("")
|
|
|
|
users = (
|
|
User.objects.filter(
|
|
Q(first_name__icontains=q)
|
|
| Q(last_name__icontains=q)
|
|
| Q(email__icontains=q)
|
|
| Q(username__icontains=q)
|
|
)
|
|
.prefetch_related(
|
|
"userprofile", "userprofile__grade", "userprofile__supervisor", "user_groups"
|
|
)
|
|
.order_by("first_name", "last_name", "username")[:20]
|
|
)
|
|
|
|
return render(
|
|
request,
|
|
"rad/partials/_people_search_results.html",
|
|
{"users": users, "q": q},
|
|
)
|
|
|
|
return render(request, "people.html", {})
|
|
|
|
|
|
|
|
@login_required
|
|
def profile(request):
|
|
"""Render the current user's profile page with group management context.
|
|
|
|
Scope: authenticated user's own profile view.
|
|
Functionality: provides user details, available groups, and supervisor
|
|
account state for template actions.
|
|
"""
|
|
user = request.user
|
|
available_groups = UserUserGroup.objects.filter(archive=False).exclude(
|
|
pk__in=user.user_groups.values_list("pk", flat=True)
|
|
)
|
|
available_auth_groups = Group.objects.exclude(
|
|
pk__in=user.groups.values_list("pk", flat=True)
|
|
)
|
|
is_cid_user_manager = request.user.groups.filter(name="cid_user_manager").exists()
|
|
supervisor_account = Supervisor.objects.filter(user=user).first()
|
|
return render(
|
|
request,
|
|
"profile.html",
|
|
{
|
|
"user": user,
|
|
"available_groups": available_groups,
|
|
"available_auth_groups": available_auth_groups,
|
|
"is_cid_user_manager": is_cid_user_manager,
|
|
"supervisor_account": supervisor_account,
|
|
"supervisor_action": request.GET.get("supervisor_action", ""),
|
|
"password_reset_action": request.GET.get("password_reset_action", ""),
|
|
},
|
|
)
|
|
|
|
|
|
def index(request):
|
|
collections = []
|
|
|
|
if request.user.is_authenticated:
|
|
collections = ExamCollection.objects.filter(archive=False, author=request.user)
|
|
|
|
rcr_assessor = request.user.groups.filter(
|
|
Q(name="rcr_radiology_assessor")
|
|
| Q(name="rcr_oncology_assessor")
|
|
| Q(name="rcr_assessor")
|
|
).exists()
|
|
return render(
|
|
request,
|
|
"index.html",
|
|
{"rcr_assessor": rcr_assessor, "collections": collections},
|
|
)
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def account_profile(request, slug):
|
|
"""Render an account profile page for administrators.
|
|
|
|
Scope: cid_user_manager and superusers viewing another user's profile.
|
|
Functionality: exposes profile details, group controls, and supervisor
|
|
account creation/link status for admin actions.
|
|
"""
|
|
user = get_object_or_404(User, username=slug)
|
|
available_groups = UserUserGroup.objects.filter(archive=False).exclude(
|
|
pk__in=user.user_groups.values_list("pk", flat=True)
|
|
)
|
|
available_auth_groups = Group.objects.exclude(
|
|
pk__in=user.groups.values_list("pk", flat=True)
|
|
)
|
|
is_cid_user_manager = request.user.groups.filter(name="cid_user_manager").exists()
|
|
supervisor_account = Supervisor.objects.filter(user=user).first()
|
|
return render(
|
|
request,
|
|
"profile.html",
|
|
{
|
|
"user": user,
|
|
"available_groups": available_groups,
|
|
"available_auth_groups": available_auth_groups,
|
|
"is_cid_user_manager": is_cid_user_manager,
|
|
"supervisor_account": supervisor_account,
|
|
"supervisor_action": request.GET.get("supervisor_action", ""),
|
|
"password_reset_action": request.GET.get("password_reset_action", ""),
|
|
},
|
|
)
|
|
|
|
|
|
@require_POST
|
|
@user_is_cid_user_manager
|
|
def account_set_supervisor(request, slug):
|
|
"""Create or link a Supervisor record for an existing user.
|
|
|
|
Scope: cid_user_manager and superusers only.
|
|
Functionality: one-click action from profile pages to assign a user as a
|
|
supervisor account, reusing an existing supervisor by email when safe.
|
|
"""
|
|
user = get_object_or_404(User, username=slug)
|
|
|
|
next_url = request.POST.get("next") or reverse("account_profile", kwargs={"slug": user.username})
|
|
|
|
email = (user.email or "").strip().lower()
|
|
if not email:
|
|
return redirect(f"{next_url}?supervisor_action=no_email")
|
|
|
|
existing_for_user = Supervisor.objects.filter(user=user).first()
|
|
if existing_for_user:
|
|
return redirect(f"{next_url}?supervisor_action=already_linked")
|
|
|
|
supervisor = Supervisor.objects.filter(email__iexact=email).first()
|
|
if supervisor:
|
|
if supervisor.user and supervisor.user != user:
|
|
return redirect(f"{next_url}?supervisor_action=email_in_use")
|
|
|
|
supervisor.user = user
|
|
if not supervisor.name:
|
|
supervisor.name = user.get_full_name().strip() or user.username
|
|
supervisor.save()
|
|
return redirect(f"{next_url}?supervisor_action=linked")
|
|
|
|
name = user.get_full_name().strip() or user.username
|
|
Supervisor.objects.create(email=email, name=name, user=user)
|
|
return redirect(f"{next_url}?supervisor_action=created")
|
|
|
|
|
|
@require_POST
|
|
@user_is_cid_user_manager
|
|
def account_send_password_reset(request, slug):
|
|
"""Send a password reset email for an existing user from their profile page."""
|
|
user = get_object_or_404(User, username=slug)
|
|
|
|
next_url = request.POST.get("next") or reverse("account_profile", kwargs={"slug": user.username})
|
|
email = (user.email or user.username or "").strip()
|
|
|
|
if not email:
|
|
return redirect(f"{next_url}?password_reset_action=no_email")
|
|
|
|
if not user.is_active:
|
|
return redirect(f"{next_url}?password_reset_action=inactive")
|
|
|
|
form = PasswordResetForm({"email": email})
|
|
if not form.is_valid():
|
|
return redirect(f"{next_url}?password_reset_action=invalid")
|
|
|
|
try:
|
|
form.save(
|
|
request=request,
|
|
use_https=request.is_secure(),
|
|
from_email=getattr(settings, "DEFAULT_FROM_EMAIL", None),
|
|
)
|
|
except Exception:
|
|
logger.exception("Failed to send password reset email for user {}", user.pk)
|
|
return redirect(f"{next_url}?password_reset_action=error")
|
|
|
|
return redirect(f"{next_url}?password_reset_action=sent")
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def logs_view(request):
|
|
"""Display production logs with search/filter controls.
|
|
|
|
Scope: superusers only.
|
|
Functionality: reads the configured application log file and displays entries with optional text search filter.
|
|
Supports ?all=1 to show all entries without pagination limit.
|
|
"""
|
|
log_file_path = str(get_log_file_path())
|
|
search_query = request.GET.get("q", "").strip()
|
|
level_filter = request.GET.get("level", "").strip().upper()
|
|
show_all = request.GET.get("all") == "1"
|
|
raw_mode = request.GET.get("raw") == "1"
|
|
include_noise = request.GET.get("include_noise") == "1"
|
|
date_from_raw = request.GET.get("date_from", "").strip()
|
|
date_to_raw = request.GET.get("date_to", "").strip()
|
|
time_from_raw = request.GET.get("time_from", "").strip()
|
|
time_to_raw = request.GET.get("time_to", "").strip()
|
|
action_message = ""
|
|
|
|
if request.method == "POST" and request.POST.get("action") == "clear":
|
|
try:
|
|
os.makedirs(os.path.dirname(log_file_path), exist_ok=True)
|
|
with open(log_file_path, "w", encoding="utf-8"):
|
|
pass
|
|
action_message = "Log file cleared."
|
|
except Exception as exc:
|
|
action_message = f"Failed to clear log file: {exc}"
|
|
|
|
def _parse_log_timestamp(timestamp_text: str) -> datetime | None:
|
|
if not timestamp_text:
|
|
return None
|
|
for timestamp_format in ("%d/%b/%Y %H:%M:%S", "%Y-%m-%d %H:%M:%S"):
|
|
try:
|
|
return datetime.strptime(timestamp_text, timestamp_format)
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
def _parse_date_input(value: str):
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.strptime(value, "%Y-%m-%d").date()
|
|
except ValueError:
|
|
return None
|
|
|
|
def _parse_time_input(value: str):
|
|
if not value:
|
|
return None
|
|
for time_format in ("%H:%M", "%H:%M:%S"):
|
|
try:
|
|
return datetime.strptime(value, time_format).time()
|
|
except ValueError:
|
|
continue
|
|
return None
|
|
|
|
date_from = _parse_date_input(date_from_raw)
|
|
date_to = _parse_date_input(date_to_raw)
|
|
time_from = _parse_time_input(time_from_raw)
|
|
time_to = _parse_time_input(time_to_raw)
|
|
|
|
if raw_mode:
|
|
if not os.path.exists(log_file_path):
|
|
return HttpResponse(
|
|
f"Log file not found at {log_file_path}\n",
|
|
content_type="text/plain; charset=utf-8",
|
|
status=404,
|
|
)
|
|
|
|
try:
|
|
with open(log_file_path, "r", encoding="utf-8", errors="ignore") as f:
|
|
raw_content = f.read()
|
|
except Exception as e:
|
|
return HttpResponse(
|
|
f"Error reading log file: {str(e)}\n",
|
|
content_type="text/plain; charset=utf-8",
|
|
status=500,
|
|
)
|
|
|
|
return HttpResponse(raw_content, content_type="text/plain; charset=utf-8")
|
|
|
|
log_header_re = re.compile(
|
|
r"^\[(?P<timestamp>[^\]]+)\]\s+(?P<level>[A-Z]+)\s+\[(?P<logger>[^:\]]+):(?P<line>\d+)\]\s+(?P<message>.*)$"
|
|
)
|
|
|
|
parsed_entries: list[dict[str, Any]] = []
|
|
if os.path.exists(log_file_path):
|
|
try:
|
|
with open(log_file_path, "r", encoding="utf-8", errors="ignore") as f:
|
|
lines = f.readlines()
|
|
|
|
current_entry = None
|
|
for raw_line in lines:
|
|
line = raw_line.rstrip("\n")
|
|
if not line.strip():
|
|
if current_entry is not None:
|
|
current_entry["details"].append("")
|
|
continue
|
|
|
|
header_match = log_header_re.match(line)
|
|
if header_match:
|
|
if current_entry is not None:
|
|
parsed_entries.append(current_entry)
|
|
|
|
current_entry = {
|
|
"timestamp": header_match.group("timestamp"),
|
|
"level": header_match.group("level"),
|
|
"logger": header_match.group("logger"),
|
|
"source_line": header_match.group("line"),
|
|
"message": header_match.group("message"),
|
|
"details": [],
|
|
}
|
|
continue
|
|
|
|
if current_entry is not None:
|
|
current_entry["details"].append(line)
|
|
else:
|
|
parsed_entries.append(
|
|
{
|
|
"timestamp": "",
|
|
"level": "RAW",
|
|
"logger": "",
|
|
"source_line": "",
|
|
"message": line,
|
|
"details": [],
|
|
}
|
|
)
|
|
|
|
if current_entry is not None:
|
|
parsed_entries.append(current_entry)
|
|
|
|
except Exception as e:
|
|
parsed_entries = [
|
|
{
|
|
"timestamp": "",
|
|
"level": "ERROR",
|
|
"logger": "logs_view",
|
|
"source_line": "",
|
|
"message": f"Error reading log file: {str(e)}",
|
|
"details": [],
|
|
}
|
|
]
|
|
else:
|
|
parsed_entries = [
|
|
{
|
|
"timestamp": "",
|
|
"level": "ERROR",
|
|
"logger": "logs_view",
|
|
"source_line": "",
|
|
"message": f"Log file not found at {log_file_path}",
|
|
"details": [],
|
|
}
|
|
]
|
|
|
|
# Most recent first.
|
|
parsed_entries = list(reversed(parsed_entries))
|
|
|
|
available_levels = sorted({entry["level"] for entry in parsed_entries if entry["level"]})
|
|
|
|
filtered_entries = []
|
|
search_lower = search_query.lower()
|
|
noise_suppressed = 0
|
|
for entry in parsed_entries:
|
|
# Django template debug logs can be very noisy and usually do not indicate
|
|
# fatal issues. Hide them by default, with an explicit opt-in toggle.
|
|
is_template_debug_noise = (
|
|
entry.get("level") == "DEBUG"
|
|
and entry.get("logger") == "django.template"
|
|
and (
|
|
"Exception while resolving variable" in entry.get("message", "")
|
|
or any("VariableDoesNotExist" in d for d in entry.get("details", []))
|
|
)
|
|
)
|
|
if is_template_debug_noise and not include_noise:
|
|
noise_suppressed += 1
|
|
continue
|
|
|
|
if level_filter and entry["level"] != level_filter:
|
|
continue
|
|
|
|
haystack = "\n".join(
|
|
[
|
|
entry.get("timestamp", ""),
|
|
entry.get("level", ""),
|
|
entry.get("logger", ""),
|
|
entry.get("message", ""),
|
|
"\n".join(entry.get("details", [])),
|
|
]
|
|
).lower()
|
|
if search_lower and search_lower not in haystack:
|
|
continue
|
|
|
|
entry_timestamp = _parse_log_timestamp(entry.get("timestamp", ""))
|
|
if date_from or date_to or time_from or time_to:
|
|
if entry_timestamp is None:
|
|
continue
|
|
if date_from and entry_timestamp.date() < date_from:
|
|
continue
|
|
if date_to and entry_timestamp.date() > date_to:
|
|
continue
|
|
if time_from and entry_timestamp.time() < time_from:
|
|
continue
|
|
if time_to and entry_timestamp.time() > time_to:
|
|
continue
|
|
|
|
filtered_entries.append(entry)
|
|
|
|
total = len(filtered_entries)
|
|
limit = None if show_all else 500
|
|
displayed_entries = filtered_entries[:limit] if limit else filtered_entries
|
|
|
|
return render(
|
|
request,
|
|
"logs_viewer.html",
|
|
{
|
|
"entries": displayed_entries,
|
|
"total_entries": total,
|
|
"search_query": search_query,
|
|
"level_filter": level_filter,
|
|
"available_levels": available_levels,
|
|
"log_file_path": log_file_path,
|
|
"show_all": show_all,
|
|
"limit": limit,
|
|
"raw_url": f"{reverse('logs_view')}?raw=1",
|
|
"include_noise": include_noise,
|
|
"noise_suppressed": noise_suppressed,
|
|
"date_from": date_from_raw,
|
|
"date_to": date_to_raw,
|
|
"time_from": time_from_raw,
|
|
"time_to": time_to_raw,
|
|
"action_message": action_message,
|
|
},
|
|
)
|
|
|
|
|
|
@login_required
|
|
@user_is_cid_user_manager
|
|
def media_cleanup(request):
|
|
"""Run django-unused-media cleanup from a manager-facing maintenance page.
|
|
|
|
Scope: cid_user_manager group members and superusers.
|
|
Functionality: supports safe preview (dry-run) and confirmed deletion runs,
|
|
returning HTMX fragments for inline results.
|
|
"""
|
|
context: dict[str, Any] = {
|
|
"ran": False,
|
|
"action": "preview",
|
|
"minimum_file_age": 60,
|
|
"remove_empty_dirs": True,
|
|
"exclude": "",
|
|
}
|
|
|
|
if request.method == "POST":
|
|
action = request.POST.get("action", "preview")
|
|
dry_run = action != "cleanup"
|
|
exclude_raw = (request.POST.get("exclude") or "").strip()
|
|
exclude = [line.strip() for line in exclude_raw.splitlines() if line.strip()]
|
|
remove_empty_dirs = request.POST.get("remove_empty_dirs") == "on"
|
|
|
|
min_age_raw = (request.POST.get("minimum_file_age") or "60").strip()
|
|
try:
|
|
minimum_file_age = max(0, int(min_age_raw))
|
|
except (TypeError, ValueError):
|
|
minimum_file_age = 0
|
|
|
|
out = StringIO()
|
|
success = True
|
|
|
|
command_kwargs: dict[str, Any] = {
|
|
"dry_run": dry_run,
|
|
"no_input": True,
|
|
"interactive": False,
|
|
"minimum_file_age": minimum_file_age,
|
|
"remove_empty_dirs": remove_empty_dirs,
|
|
"verbosity": 2,
|
|
"stdout": out,
|
|
}
|
|
if exclude:
|
|
command_kwargs["exclude"] = exclude
|
|
|
|
error_message = ""
|
|
media_inventory_before = _media_file_inventory(settings.MEDIA_ROOT)
|
|
media_inventory_after = media_inventory_before
|
|
try:
|
|
call_command("cleanup_unused_media", **command_kwargs)
|
|
except CommandError as exc:
|
|
success = False
|
|
error_message = str(exc)
|
|
except Exception as exc: # pragma: no cover
|
|
success = False
|
|
error_message = str(exc)
|
|
|
|
media_inventory_after = _media_file_inventory(settings.MEDIA_ROOT)
|
|
|
|
media_size_before = sum(media_inventory_before.values())
|
|
media_size_after = sum(media_inventory_after.values())
|
|
removed_files = set(media_inventory_before) - set(media_inventory_after)
|
|
|
|
output = out.getvalue().strip()
|
|
# Use before/after total delta so reported recovered space matches real disk impact.
|
|
recovered_size = max(0, media_size_before - media_size_after)
|
|
context.update(
|
|
{
|
|
"ran": True,
|
|
"success": success,
|
|
"action": action,
|
|
"dry_run": dry_run,
|
|
"minimum_file_age": minimum_file_age,
|
|
"remove_empty_dirs": remove_empty_dirs,
|
|
"exclude": exclude_raw,
|
|
"output": output,
|
|
"error_message": error_message,
|
|
"media_size_before": media_size_before,
|
|
"media_size_after": media_size_after,
|
|
"recovered_size": recovered_size,
|
|
"removed_file_count": len(removed_files),
|
|
}
|
|
)
|
|
|
|
if request.htmx:
|
|
return render(request, "rad/partials/_media_cleanup_result.html", context)
|
|
|
|
return render(request, "rad/media_cleanup.html", context)
|
|
|
|
|
|
def cid_selector(request):
|
|
return render(
|
|
request,
|
|
"cid_selector.html",
|
|
)
|
|
|
|
|
|
@login_required
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def cid_scores_admin(request, cid):
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
|
|
if not cid_user:
|
|
raise Http404("CID not found")
|
|
|
|
return cid_user_overview(request, cid_user.cid, cid_user.passcode)
|
|
|
|
|
|
@login_required
|
|
# @user_passes_test(lambda u: u.is_superuser)
|
|
@user_is_cid_user_manager
|
|
def cid_results(request, cid):
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
|
|
if not cid_user:
|
|
raise Http404("CID not found")
|
|
|
|
report = cid_user.generate_exam_report()
|
|
|
|
return render(
|
|
request,
|
|
"cid_results.html",
|
|
{
|
|
"report": report,
|
|
},
|
|
)
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def user_scores_admin(request, user_id):
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
raise Http404("Cid not found")
|
|
|
|
return user_scores(request, user=user)
|
|
|
|
|
|
@login_required
|
|
def user_scores(request, user=None):
|
|
admin = True
|
|
if user is None:
|
|
user = request.user
|
|
admin = False
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from generic.models import UserHiddenItem
|
|
|
|
show_hidden = request.GET.get("show_hidden") in ("true", "1")
|
|
|
|
# Fetch user's hidden items map: {content_type_id: {object_ids}}
|
|
hidden_map = {}
|
|
for ct_id, obj_id in UserHiddenItem.objects.filter(user=user).values_list("content_type_id", "object_id"):
|
|
hidden_map.setdefault(ct_id, set()).add(obj_id)
|
|
|
|
# Filter Exam Results list (all_exams)
|
|
exams = get_user_exams(user)
|
|
filtered_exams = []
|
|
for exam_type, exams_qs in exams:
|
|
exam_model = exams_qs.model
|
|
ct_id = ContentType.objects.get_for_model(exam_model).id
|
|
hidden_ids = hidden_map.get(ct_id, set())
|
|
|
|
if not show_hidden:
|
|
exams_qs = exams_qs.exclude(id__in=hidden_ids)
|
|
|
|
exams_list = list(exams_qs)
|
|
for exam in exams_list:
|
|
exam.is_hidden = exam.id in hidden_ids
|
|
|
|
if exams_list:
|
|
filtered_exams.append((exam_type, exams_list))
|
|
|
|
# Filter Assigned Exams list (available_exams)
|
|
available_exams = []
|
|
for n, t in USER_EXAM_TYPES:
|
|
exam_rel = getattr(user, t)
|
|
if exam_rel.exists():
|
|
temp_exams = exam_rel.filter(exam_mode=True, archive=False).order_by("name")
|
|
exam_model = temp_exams.model
|
|
ct_id = ContentType.objects.get_for_model(exam_model).id
|
|
hidden_ids = hidden_map.get(ct_id, set())
|
|
|
|
if not show_hidden:
|
|
temp_exams = temp_exams.exclude(id__in=hidden_ids)
|
|
|
|
temp_exams_list = list(temp_exams)
|
|
for exam in temp_exams_list:
|
|
exam.is_hidden = exam.id in hidden_ids
|
|
|
|
if temp_exams_list:
|
|
available_exams.append((n, temp_exams_list))
|
|
|
|
return render(
|
|
request,
|
|
"user_scores.html",
|
|
{
|
|
"all_exams": filtered_exams,
|
|
"cid_user": user,
|
|
"available_exams": available_exams,
|
|
"case_collections": [],
|
|
"admin": admin,
|
|
"show_hidden": show_hidden,
|
|
},
|
|
)
|
|
|
|
|
|
|
|
@login_required
|
|
def user_marking(request):
|
|
"""Main page for user marking overview. Uses HTMX to fetch per-app partials."""
|
|
apps = ["physics", "anatomy", "rapids", "shorts", "longs", "sbas"]
|
|
return render(request, "rad/user_marking.html", {"apps": apps})
|
|
|
|
|
|
@login_required
|
|
def user_marking_partial(request, exam_type: str):
|
|
"""Return an HTMX partial listing exams the user can mark for a given exam type.
|
|
|
|
Each exam is checked for whether it has any unmarked answers (using
|
|
question.get_unmarked_user_answer_count when available).
|
|
"""
|
|
user = request.user
|
|
|
|
try:
|
|
ExamModel = get_exam_model_from_app_name(exam_type)
|
|
except Exception:
|
|
return HttpResponse("Unknown exam type", status=400)
|
|
|
|
# Exams where user is an author or an authorised marker
|
|
exams_qs = (
|
|
ExamModel.objects.filter(archive=False, exam_mode=True)
|
|
.filter(Q(author=user) | Q(markers=user))
|
|
.distinct()
|
|
.order_by("name")
|
|
.prefetch_related("author", "markers")
|
|
)
|
|
|
|
exams_data = []
|
|
for exam in exams_qs:
|
|
has_unmarked = False
|
|
unmarked_count = 0
|
|
# iterate questions and use their helper to count unmarked answers when available
|
|
for q in exam.get_questions():
|
|
count = 0
|
|
# Prefer question-provided helper if present
|
|
helper = getattr(q, "get_unmarked_user_answer_count", None)
|
|
if callable(helper):
|
|
try:
|
|
count = helper(exam_pk=exam.pk, marker=user)
|
|
except TypeError:
|
|
count = helper(exam_pk=exam.pk)
|
|
except Exception:
|
|
count = 0
|
|
else:
|
|
# Fallback: inspect related answer manager if present
|
|
rel_manager = None
|
|
for rel_name in ("cid_user_answers", "user_answers", "answers", "useranswer_set"):
|
|
if hasattr(q, rel_name):
|
|
rel_manager = getattr(q, rel_name)
|
|
break
|
|
|
|
if rel_manager is not None:
|
|
try:
|
|
ans_model = rel_manager.model
|
|
# If the answer model has a `score` field, count answers that are unmarked
|
|
field_names = [f.name for f in ans_model._meta.get_fields()]
|
|
if "score" in field_names:
|
|
# Try to detect an UNMARKED constant on the model
|
|
unmarked_vals = [None, ""]
|
|
model_score_const = getattr(ans_model, "ScoreOptions", None) or getattr(ans_model, "MarkOptions", None)
|
|
if model_score_const is not None:
|
|
# attempt to use any UNMARKED attribute
|
|
unmarked_val = getattr(model_score_const, "UNMARKED", None)
|
|
if unmarked_val is not None:
|
|
unmarked_vals.append(unmarked_val)
|
|
|
|
count = rel_manager.filter(exam__id=exam.pk).filter(score__in=unmarked_vals).count()
|
|
else:
|
|
# No score field — likely auto-marked; treat as 0
|
|
count = 0
|
|
except Exception:
|
|
count = 0
|
|
|
|
try:
|
|
c = int(count)
|
|
except Exception:
|
|
c = 0
|
|
if c:
|
|
has_unmarked = True
|
|
unmarked_count += c
|
|
|
|
exams_data.append({"exam": exam, "has_unmarked": has_unmarked, "unmarked_count": unmarked_count})
|
|
|
|
return render(request, "generic/partials/user_marking_exam_list.html", {"exams_data": exams_data, "exam_type": exam_type})
|
|
|
|
|
|
def cid_user_overview(request, cid, passcode):
|
|
# exam = get_object_or_404(Exam, pk=pk)
|
|
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
if not cid_user or cid_user.passcode != passcode:
|
|
raise Http404("CID / Passcode combination not found")
|
|
|
|
EXAM_ANSWER_MAP = {
|
|
"physics": (PhysicsUserAnswer, PhysicsExam),
|
|
"anatomy": (AnatomyUserAnswer, AnatomyExam),
|
|
"rapids": (RapidsUserAnswer, RapidsExam),
|
|
"longs": (LongsUserAnswer, LongsExam),
|
|
"sbas": (SbasUserAnswer, SbasExam),
|
|
}
|
|
kwargs = {"exam_mode": True, "archive": False}
|
|
|
|
exams = []
|
|
for exam_type in EXAM_ANSWER_MAP:
|
|
UserAnswer, Exam = EXAM_ANSWER_MAP[exam_type]
|
|
exam_answers = UserAnswer.objects.filter(cid=cid)
|
|
exam_ids = exam_answers.values_list("exam").distinct()
|
|
exams_to_add = Exam.objects.filter(id__in=exam_ids, **kwargs)
|
|
if exams_to_add:
|
|
exams.append((exam_type, exams_to_add))
|
|
|
|
case_collections = cid_user.casecollection_exams.all()
|
|
|
|
available_exams = cid_user.get_cid_exams(include_case_collections=False)
|
|
|
|
return render(
|
|
request,
|
|
"cid_user_overview.html",
|
|
{
|
|
# "physics_exams": physics_exams,
|
|
# "anatomy_exams": anatomy_exams,
|
|
# "rapid_exams": rapid_exams,
|
|
# "longs_exams": longs_exams,
|
|
# "sba_exams": sba_exams,
|
|
"all_exams": exams,
|
|
"cid": cid,
|
|
"passcode": passcode,
|
|
"cid_user": cid_user,
|
|
"available_exams": available_exams,
|
|
"case_collections": case_collections,
|
|
},
|
|
)
|
|
|
|
|
|
def active_exams(request, cid=None, passcode=None, based=True):
|
|
active_exams = {}
|
|
|
|
if cid is not None:
|
|
active_exams["cid"] = cid
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
if not cid_user or cid_user.passcode != passcode:
|
|
return JsonResponse({"status": "invalid"}, status=401)
|
|
|
|
anatomy_exams = AnatomyExamViews.active_exams(
|
|
request, False, cid=cid, passcode=passcode
|
|
)
|
|
rapid_exams = RapidsExamsViews.active_exams(
|
|
request, False, cid=cid, passcode=passcode
|
|
)
|
|
short_exams = ShortsExamsViews.active_exams(
|
|
request, False, cid=cid, passcode=passcode
|
|
)
|
|
long_exams = LongsExamsViews.active_exams(request, False, cid=cid, passcode=passcode)
|
|
|
|
exams = []
|
|
exams.extend(anatomy_exams)
|
|
exams.extend(rapid_exams)
|
|
exams.extend(short_exams)
|
|
exams.extend(long_exams)
|
|
|
|
active_exams["exams"] = exams
|
|
|
|
if request.user:
|
|
active_exams["user"] = request.user.username
|
|
active_exams["user_id"] = request.user.pk
|
|
else:
|
|
active_exams["user"] = False
|
|
|
|
return JsonResponse(active_exams)
|
|
|
|
|
|
def active_exams_unbased(request, cid=None, passcode=None):
|
|
return active_exams(request, cid, passcode, based=False)
|
|
|
|
|
|
@csrf_exempt
|
|
def exam_submit(request):
|
|
if request.method == "POST":
|
|
exam_type, exam_id = request.POST.get("eid").split("/")
|
|
|
|
try:
|
|
if exam_type == "anatomy":
|
|
return AnatomyExamViews.post_exam_answers(request)
|
|
elif exam_type == "rapid":
|
|
return RapidsExamsViews.post_exam_answers(request)
|
|
elif exam_type == "long":
|
|
return LongsExamsViews.post_exam_answers(request)
|
|
elif exam_type == "short":
|
|
return ShortsExamsViews.post_exam_answers(request)
|
|
|
|
return JsonResponse({"success": False, "error": "Invalid exam type"})
|
|
except ObjectDoesNotExist:
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
# post_exam_answers
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
|
|
def feedback_create(request, question_type, pk):
|
|
if question_type == "anatomy":
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
question = Rapid
|
|
elif question_type == "long":
|
|
question = Long
|
|
|
|
return JsonResponse({"success": False, "error": "Invalid exam type"})
|
|
|
|
|
|
class AddQuestionNote(CreateView):
|
|
model = QuestionNote
|
|
form_class = QuestionNoteForm
|
|
|
|
# fields = ("author", "note_type", "note")
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super(AddQuestionNote, self).get_form_kwargs()
|
|
# update the kwargs for the form init method with yours
|
|
kwargs.update(self.kwargs) # self.kwargs contains all url conf params
|
|
kwargs.update({"user": self.request.user})
|
|
return kwargs
|
|
|
|
def get_initial(self):
|
|
pk = self.kwargs["pk"]
|
|
self.question_type = self.kwargs["question_type"]
|
|
|
|
question, content_type = get_question_and_content_type(self.question_type)
|
|
|
|
self.question = get_object_or_404(question, pk=pk)
|
|
return {"content_type": content_type, "object_id": pk}
|
|
|
|
def get_context_data(self, **kwargs):
|
|
# This can also be accessed via view.**** in the template
|
|
context = super(AddQuestionNote, self).get_context_data(**kwargs)
|
|
context["question"] = self.question
|
|
context["question_type"] = self.question_type
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
model = form.save(commit=False)
|
|
|
|
if self.request.user.is_authenticated:
|
|
model.user_author = self.request.user
|
|
|
|
model.save()
|
|
|
|
response = super().form_valid(form)
|
|
|
|
return render(self.request, "feedback_submitted.html")
|
|
return response
|
|
|
|
def get_success_url(self) -> str:
|
|
return super().get_success_url()
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def feedback_note_delete(request, pk):
|
|
q = get_object_or_404(QuestionNote, pk=pk)
|
|
q.delete()
|
|
|
|
return redirect(q.get_object_url())
|
|
|
|
|
|
@login_required
|
|
def feedback_mark_complete(request, pk):
|
|
q = get_object_or_404(QuestionNote, pk=pk)
|
|
|
|
if (
|
|
request.user not in q.question.author.all()
|
|
and not request.user.groups.filter(name="feedback_checker").exists()
|
|
):
|
|
raise PermissionDenied()
|
|
|
|
q.complete = True
|
|
q.save()
|
|
return redirect(q.get_object_url())
|
|
|
|
|
|
@login_required
|
|
def answer_suggestion_submit(request):
|
|
"""Submit a proposed answer to a question for review.
|
|
|
|
Scope: authenticated users only.
|
|
Functionality: accepts a JSON POST containing the question ID, proposed answer
|
|
string, and status; creates a new Answer record marked as proposed=True for
|
|
moderator review. Only 'anatomy' and 'rapid' question types are supported.
|
|
"""
|
|
if request.method != "POST":
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
post_data = json.loads(request.body)
|
|
question_type, qid = post_data["qid"].split("/")
|
|
answer_string = post_data["answer"]
|
|
status = post_data["status"]
|
|
|
|
if str(status) not in " 012":
|
|
return JsonResponse({"success": False, "error": "Invalid status"})
|
|
|
|
if question_type == "anatomy":
|
|
AnswerModel = AnatomyAnswer
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
AnswerModel = RapidAnswer
|
|
question = Rapid
|
|
else:
|
|
return JsonResponse({"success": False, "error": "Invalid question type"})
|
|
|
|
q = get_object_or_404(question, pk=qid)
|
|
if AnswerModel.objects.filter(answer=answer_string, question=q):
|
|
return JsonResponse({"success": False, "error": "Answer already exists"})
|
|
|
|
a = AnswerModel(question=q, answer=answer_string, status=status, proposed=True)
|
|
a.save()
|
|
return JsonResponse({"success": True, "error": "Answer submitted"})
|
|
|
|
|
|
@login_required
|
|
def answer_submit(request):
|
|
if request.method == "POST":
|
|
qid = request.POST.get("qid")
|
|
question_type = request.POST.get("question_type")
|
|
answer_string = request.POST.get("answer")
|
|
status = request.POST.get("status")
|
|
|
|
if str(status) not in "012":
|
|
return JsonResponse({"success": False, "error": "Invalid status"})
|
|
|
|
if question_type == "anatomy":
|
|
AnswerModel = AnatomyAnswer
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
AnswerModel = RapidAnswer
|
|
question = Rapid
|
|
else:
|
|
return JsonResponse({"success": False, "error": "Invalid question type"})
|
|
|
|
q = get_object_or_404(question, pk=qid)
|
|
if AnswerModel.objects.filter(answer=answer_string, question=q):
|
|
return JsonResponse({"success": False, "error": "Answer already exists"})
|
|
|
|
a = AnswerModel(question=q, answer=answer_string, status=status)
|
|
a.save()
|
|
return JsonResponse({"success": True, "error": "Answer added"})
|
|
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
|
|
@login_required
|
|
def answer_suggestion_confirm(request):
|
|
if request.method == "POST":
|
|
question_type = request.POST.get("question_type")
|
|
aid = request.POST.get("aid")
|
|
status = request.POST.get("status")
|
|
|
|
if question_type == "anatomy":
|
|
AnswerModel = AnatomyAnswer
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
AnswerModel = RapidAnswer
|
|
question = Rapid
|
|
else:
|
|
return JsonResponse({"success": False, "error": "Invalid question type"})
|
|
|
|
answer = get_object_or_404(AnswerModel, pk=aid)
|
|
answer.proposed = False
|
|
answer.status = status
|
|
answer.save()
|
|
return JsonResponse({"success": True, "error": "Answer changed"})
|
|
|
|
|
|
@login_required
|
|
@user_passes_test(feedback_checker)
|
|
def view_feedback(request):
|
|
# exam = get_object_or_404(Exam, pk=pk)
|
|
|
|
feedback_notes = QuestionNote.objects.filter(complete=False)
|
|
|
|
return render(
|
|
request,
|
|
"view_feedback.html",
|
|
{
|
|
"notes": feedback_notes,
|
|
},
|
|
)
|
|
|
|
|
|
def privacy_view(request):
|
|
return render(request, "privacy.html")
|
|
|
|
|
|
def cimar(request):
|
|
if "link_id" in request.GET:
|
|
link_id = request.GET["link_id"]
|
|
return render(request, "cimar.html", {"link_id": link_id})
|
|
return render(request, "cimar.html")
|
|
|
|
|
|
def about_view(request):
|
|
return render(request, "about.html")
|
|
|
|
|
|
def stats(request):
|
|
collection_count = CaseCollection.objects.filter(archive=False).count()
|
|
case_count = AtlasCase.objects.filter().count()
|
|
|
|
anatomy_exam_count = AnatomyExam.objects.filter(archive=False).count()
|
|
anatomy_question_count = AnatomyQuestion.objects.filter().count()
|
|
return render(
|
|
request,
|
|
"stats.html",
|
|
{
|
|
"collection_count": collection_count,
|
|
"case_count": case_count,
|
|
"anatomy_exam_count": anatomy_exam_count,
|
|
"anatomy_question_count": anatomy_question_count,
|
|
},
|
|
)
|
|
|
|
|
|
# HTTP Error 400
|
|
def page_not_found(request, exception):
|
|
# Provide the exception message (reason) to the template so callers
|
|
# can show a user-friendly explanation when available.
|
|
context = {"reason": str(exception) if exception is not None else "Not found"}
|
|
response = render(request, "404.html", context=context)
|
|
|
|
response.status_code = 404
|
|
|
|
return response
|
|
|
|
|
|
def page_forbidden(request, exception):
|
|
response = render(request, "403.html", context={"exception": exception})
|
|
|
|
response.status_code = 403
|
|
|
|
return response
|
|
|
|
|
|
def server_error(request):
|
|
response = render(request, "500.html")
|
|
|
|
response.status_code = 500
|
|
|
|
return response
|
|
|
|
|
|
class UserListView(CidManagerRequiredMixin, FilterView):
|
|
model = User
|
|
template_name = "user_list_view.html"
|
|
|
|
filterset_class = UserListFilter
|
|
|
|
|
|
class UserListTableView(CidManagerRequiredMixin, SingleTableMixin, FilterView):
|
|
model = User
|
|
table_class = UserUserTable
|
|
template_name = "generic/user_view.html"
|
|
|
|
filterset_class = UserUserFilter
|
|
|
|
# table_pagination = {"per_page": 5}
|
|
|
|
def get_paginate_by(self, table_data) -> int | None:
|
|
return 100
|
|
# return super().get_paginate_by(table_data)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
# user = self.request.user
|
|
|
|
# filters = {"archive": False, "exam_mode": True}
|
|
|
|
# physics_exams = [(i.name, i.pk) for i in PhysicsExam.objects.filter(**filters)]
|
|
# rapid_exams = [(i.name, i.pk) for i in RapidsExam.objects.filter(**filters)]
|
|
# sba_exams = [(i.name, i.pk) for i in SbasExam.objects.filter(**filters)]
|
|
# longs_exams = [(i.name, i.pk) for i in LongsExam.objects.filter(**filters)]
|
|
# anatomy_exams = [(i.name, i.pk) for i in AnatomyExam.objects.filter(**filters)]
|
|
# casecollection_exams = [
|
|
# (i.name, i.pk)
|
|
# for i in CaseCollection.objects.filter(archive=False, collection_type__gt=0)
|
|
# ]
|
|
|
|
# context["physics_exams"] = physics_exams
|
|
# context["rapid_exams"] = rapid_exams
|
|
# context["sba_exams"] = sba_exams
|
|
# context["longs_exams"] = longs_exams
|
|
# context["anatomy_exams"] = anatomy_exams
|
|
# context["casecollection_exams"] = casecollection_exams
|
|
|
|
# cid_user_groups = [
|
|
# (i.name, i.pk) for i in CidUserGroup.objects.filter(archive=False)
|
|
# ]
|
|
|
|
# context["cid_user_groups"] = cid_user_groups
|
|
return context
|
|
|
|
|
|
class DeleteUserView(CidManagerRequiredMixin, DeleteView):
|
|
model = User
|
|
template_name: str = "confirm_delete.html"
|
|
success_url = reverse_lazy("accounts_list")
|
|
|
|
|
|
class UpdateUserView(CidManagerRequiredMixin, UpdateView):
|
|
model = User
|
|
fields = [
|
|
"first_name",
|
|
"last_name",
|
|
"email",
|
|
"is_active",
|
|
] # Keep listing whatever fields
|
|
# the combined UserProfile and User exposes.
|
|
template_name = "user_update.html"
|
|
slug_field = "username"
|
|
slug_url_kwarg = "slug"
|
|
|
|
def get_success_url(self):
|
|
view_name = "account_profile"
|
|
# No need for reverse_lazy here, because it's called inside the method
|
|
return reverse(view_name, kwargs={"slug": self.object.username})
|
|
|
|
|
|
class UpdateUserProfileAdminView(CidManagerRequiredMixin, UpdateView):
|
|
model = UserProfile
|
|
fields = [
|
|
"supervisor",
|
|
"grade",
|
|
"registration_number",
|
|
"peninsula_trainee",
|
|
] # Keep listing whatever fields
|
|
template_name = "user_update_profile.html"
|
|
slug_field = "user__username"
|
|
slug_url_kwarg = "slug"
|
|
|
|
def get_success_url(self):
|
|
view_name = "account_profile"
|
|
# No need for reverse_lazy here, because it's called inside the method
|
|
return reverse(view_name, kwargs={"slug": self.object.username})
|
|
|
|
|
|
class UpdateUserProfileView(UpdateView):
|
|
model = UserProfile
|
|
fields = [
|
|
"supervisor",
|
|
"grade",
|
|
"registration_number",
|
|
# "peninsula_trainee",
|
|
] # Keep listing whatever fields
|
|
template_name = "user_update_profile.html"
|
|
slug_field = "user__username"
|
|
slug_url_kwarg = "slug"
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if (
|
|
request.user.is_superuser
|
|
or request.user.groups.filter(name="cid_user_manager").exists()
|
|
):
|
|
self.fields = [
|
|
"supervisor",
|
|
"grade",
|
|
"registration_number",
|
|
"site",
|
|
"peninsula_trainee",
|
|
]
|
|
# Check permissions for the request.user here
|
|
elif request.user != self.get_object().user:
|
|
raise Http404
|
|
elif request.user.userprofile.peninsula_trainee:
|
|
self.fields = ["supervisor", "grade", "registration_number", "site"]
|
|
|
|
else:
|
|
self.fields = ["grade", "registration_number", "site"]
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
def get_context_data(self, **kwargs: reverse_lazy) -> dict[str, Any]:
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
form = context["form"]
|
|
|
|
if "supervisor" in form.fields:
|
|
# form.fields["supervisor"].queryset = Supervisor.objects.all()
|
|
form.fields["supervisor"] = forms.ModelChoiceField(
|
|
Supervisor.objects.all().order_by("name"),
|
|
required=False,
|
|
widget=autocomplete.ModelSelect2(url="generic:supervisor-autocomplete"),
|
|
)
|
|
return context
|
|
|
|
def get_success_url(self):
|
|
if "redirect" in self.request.GET:
|
|
return self.request.GET["redirect"]
|
|
|
|
if (
|
|
self.request.user.is_superuser
|
|
or self.request.user.groups.filter(name="cid_user_manager").exists()
|
|
):
|
|
view_name = "account_profile"
|
|
# No need for reverse_lazy here, because it's called inside the method
|
|
return reverse(view_name, kwargs={"slug": self.object.username})
|
|
else:
|
|
return reverse("profile")
|
|
|
|
|
|
# class UpdateUser(TemplateView):
|
|
#
|
|
# user_form_class = UserForm
|
|
# comment_form_class = UserProfileForm
|
|
# template_name = 'user_update.html'
|
|
# slug_field = 'username'
|
|
# slug_url_kwarg = 'slug'
|
|
#
|
|
# def post(self, request):
|
|
# post_data = request.POST or None
|
|
# user_form = self.user_form_class(post_data, prefix='user')
|
|
# profile_form = self.profile_form_class(post_data, prefix='userprofile')
|
|
#
|
|
# context = self.get_context_data(post_form=user_form,
|
|
# profile_form=profile_form)
|
|
#
|
|
# if user_form.is_valid():
|
|
# self.form_save(user_form)
|
|
# if profile_form.is_valid():
|
|
# self.form_save(profile_form)
|
|
#
|
|
# return self.render_to_response(context)
|
|
#
|
|
# def form_save(self, form):
|
|
# obj = form.save()
|
|
# #messages.success(self.request, "{} saved successfully".format(obj))
|
|
# return obj
|
|
#
|
|
# def get(self, request, *args, **kwargs):
|
|
# return self.post(request, *args, **kwargs)
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def accounts_check_users(request):
|
|
if request.method == "POST":
|
|
users = json.loads(request.POST.get("user_list"))
|
|
|
|
existing_users = []
|
|
|
|
for email in users:
|
|
if User.objects.filter(username=email).exists():
|
|
existing_users.append(email)
|
|
|
|
return JsonResponse({"users": existing_users})
|
|
|
|
return
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def accounts_bulk_create_check(request):
|
|
if not request.method == "POST":
|
|
return HttpResponse("Invalid request")
|
|
|
|
user = {}
|
|
|
|
user["email"] = request.POST.get("email")
|
|
user["first_name"] = request.POST.get("first_name")
|
|
user["last_name"] = request.POST.get("last_name")
|
|
user["grade"] = request.POST.get("grade")
|
|
user["supervisor_email"] = request.POST.get("supervisor_email")
|
|
user["supervisor_name"] = request.POST.get("supervisor_name")
|
|
|
|
errors = []
|
|
info = []
|
|
|
|
if user["email"] == "undefined":
|
|
errors.append("No email provided")
|
|
|
|
if User.objects.filter(username=user["email"]).exists():
|
|
errors.append("User already exists")
|
|
|
|
if not UserGrades.objects.filter(name=user["grade"]):
|
|
errors.append("Invalid grade")
|
|
|
|
# Treat empty, missing or the string 'undefined' as no supervisor email provided
|
|
if not user.get("supervisor_email") or user["supervisor_email"] == "undefined":
|
|
# Supervisor email not provided; try to match by name
|
|
if not user.get("supervisor_name"):
|
|
errors.append("No supervisor provided")
|
|
else:
|
|
supervisor = Supervisor.objects.filter(name__iexact=user["supervisor_name"])
|
|
if not supervisor.exists():
|
|
info.append("Supervisor does not exist (will be created)")
|
|
else:
|
|
info.append("Supervisor exists (matched by name)")
|
|
if supervisor.count() > 1:
|
|
errors.append("More than one supervisor with that name")
|
|
# no further name check needed since matched by name
|
|
else:
|
|
supervisor = Supervisor.objects.filter(email=user["supervisor_email"])
|
|
|
|
if not supervisor:
|
|
info.append("Supervisor does not exist (will be created)")
|
|
else:
|
|
info.append("Supervisor exists")
|
|
|
|
if supervisor.count() > 1:
|
|
errors.append("More than one supervisor with that email")
|
|
|
|
# Compare names case-insensitively and trimmed to avoid false mismatches
|
|
try:
|
|
sup_name = (supervisor.first().name or '').strip().lower()
|
|
posted_name = (user.get("supervisor_name") or '').strip().lower()
|
|
if posted_name and sup_name != posted_name:
|
|
errors.append("Supervisor name does not match")
|
|
except Exception:
|
|
pass
|
|
|
|
if errors:
|
|
return HttpResponse(
|
|
format_html("<span class='error'>{}</span>", "<br/>".join(errors))
|
|
)
|
|
|
|
return HttpResponse(format_html("<span class='info'>{}</span>", "<br/>".join(info)))
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def accounts_bulk_create(request):
|
|
if request.method == "POST":
|
|
context = {}
|
|
|
|
users = json.loads(request.POST.get("usersjson"))
|
|
|
|
created_users = []
|
|
existing_users = []
|
|
error_text = ""
|
|
|
|
try:
|
|
users = json.loads(request.POST.get("usersjson"))
|
|
|
|
for user in users:
|
|
try:
|
|
existing_user = User.objects.get(username=user["email"])
|
|
existing_users.append(user["email"])
|
|
|
|
user_profile = UserProfile.objects.get(user=existing_user)
|
|
|
|
# Update grade and supervisor details
|
|
# This code is dup[licated]
|
|
if "grade" in user:
|
|
if not user["grade"].startswith("ST"):
|
|
user["grade"] = f"ST{user['grade']}"
|
|
|
|
grade = UserGrades.objects.get(name=user["grade"])
|
|
user_profile.grade = grade
|
|
|
|
if "supervisor_name" in user and user.get("supervisor_name"):
|
|
sup_email = user.get("supervisor_email")
|
|
# If supervisor email provided, prefer that. Otherwise try to match by name or create placeholder
|
|
if sup_email and sup_email != 'undefined':
|
|
s, created = Supervisor.objects.get_or_create(email=sup_email, defaults={"name": user.get("supervisor_name")})
|
|
else:
|
|
# try match by name
|
|
s_qs = Supervisor.objects.filter(name__iexact=user.get("supervisor_name"))
|
|
if s_qs.exists():
|
|
s = s_qs.first()
|
|
created = False
|
|
else:
|
|
import uuid
|
|
placeholder = f"no-email-{uuid.uuid4().hex}@local.invalid"
|
|
s = Supervisor.objects.create(email=placeholder, name=user.get("supervisor_name"))
|
|
created = True
|
|
|
|
if created:
|
|
s.save()
|
|
|
|
user_profile.supervisor = s
|
|
user_profile.save()
|
|
except User.DoesNotExist:
|
|
# Start by checking the email
|
|
try:
|
|
validate_email(user["email"])
|
|
except ValidationError:
|
|
error_text = (
|
|
error_text
|
|
+ f"<p>{user['email']} does not appear to be a valid email (account for {user['first_name']} {user['last_name']} not created).</p>"
|
|
)
|
|
continue
|
|
|
|
# Try to create the user
|
|
try:
|
|
user_dict = {
|
|
"username": user["email"],
|
|
"first_name": user["first_name"],
|
|
"last_name": user["last_name"],
|
|
"email": user["email"],
|
|
"password": secrets.token_hex(nbytes=16),
|
|
}
|
|
new_user = User.objects.create_user(**user_dict)
|
|
except Exception as error:
|
|
error_text = (
|
|
error_text
|
|
+ f"<p>Error creating user: {user['email']}<br/>{error}</p>"
|
|
)
|
|
# No need to try and create a profile if we can't create a user
|
|
continue
|
|
|
|
try:
|
|
user_profile = UserProfile.objects.get(user=new_user)
|
|
user_profile.peninsula_trainee = True
|
|
# TODO: check supervisor details are correct
|
|
if "supervisor_name" in user and user.get("supervisor_name"):
|
|
sup_email = user.get("supervisor_email")
|
|
if sup_email and sup_email != 'undefined':
|
|
s, created = Supervisor.objects.get_or_create(email=sup_email, defaults={"name": user.get("supervisor_name")})
|
|
else:
|
|
s_qs = Supervisor.objects.filter(name__iexact=user.get("supervisor_name"))
|
|
if s_qs.exists():
|
|
s = s_qs.first()
|
|
created = False
|
|
else:
|
|
import uuid
|
|
placeholder = f"no-email-{uuid.uuid4().hex}@local.invalid"
|
|
s = Supervisor.objects.create(email=placeholder, name=user.get("supervisor_name"))
|
|
created = True
|
|
|
|
if created:
|
|
s.save()
|
|
|
|
user_profile.supervisor = s
|
|
|
|
if "grade" in user:
|
|
if not user["grade"].startswith("ST"):
|
|
user["grade"] = f"ST{user['grade']}"
|
|
|
|
grade = UserGrades.objects.get(name=user["grade"])
|
|
user_profile.grade = grade
|
|
|
|
user_profile.save()
|
|
|
|
created_users.append(user["email"])
|
|
except Exception as error:
|
|
# If we fail to create the profile delete the user
|
|
new_user.delete()
|
|
error_text = (
|
|
error_text
|
|
+ f"<p>Error creating user profile: {user['email']}<br/>{error} (user not created)</p>"
|
|
)
|
|
|
|
except Exception as error:
|
|
error_text = error_text + f"<p>Error creating users.<br/>{error}</p>"
|
|
|
|
context["error"] = mark_safe(error_text)
|
|
context["created_users"] = created_users
|
|
|
|
if existing_users:
|
|
context["message"] = "The following users already exist: " + ", ".join(
|
|
existing_users
|
|
)
|
|
|
|
return render(request, "accounts_bulk_create.html", context)
|
|
else:
|
|
return render(request, "accounts_bulk_create.html", {})
|
|
|
|
|
|
def request_cid_details(request):
|
|
email = request.POST.get("email")
|
|
try:
|
|
validate_email(email)
|
|
except ValidationError:
|
|
return HttpResponse("Invalid email.")
|
|
|
|
cid_users = CidUser.objects.filter(email=email)
|
|
|
|
for cid_user in cid_users:
|
|
cid_user.email_details(resend=True)
|
|
cid_user.email_details()
|
|
|
|
return HttpResponse(
|
|
f"Candidate details will be sent to the requested email (if it is found in the system). If you have not received an email after a few minutes you may need to contact <a href='mailto:{settings.CONTACT_EMAIL}'>{settings.CONTACT_EMAIL}</a>"
|
|
)
|
|
|
|
|
|
def cimar_study_dicom_json(request, uuid="c9417b53-87aa-4c40-aca0-3fa34c225536"):
|
|
with CimarAPI(request.user.userprofile.cimar_sid) as api:
|
|
# studies = api.list_studies()
|
|
# test_study = studies["studies"][0]
|
|
|
|
# schema = api.get_study_schema(test_study)
|
|
|
|
schema = api.get_study_schema_by_uuid("c9417b53-87aa-4c40-aca0-3fa34c225536")
|
|
|
|
return JsonResponse(api.get_study_dicom_json(schema))
|
|
|
|
|
|
def cimar_study_viewer(request, uuid="c9417b53-87aa-4c40-aca0-3fa34c225536"):
|
|
case = get_object_or_404(CimarCase, uuid=uuid)
|
|
|
|
viewer_link = case.viewer_link + f"?sid={request.user.userprofile.cimar_sid}"
|
|
return render(request, "cimar_embed.html", {"viewer_link": viewer_link})
|
|
|
|
|
|
def cimar_study_viewer_embed(request, uuid="c9417b53-87aa-4c40-aca0-3fa34c225536"):
|
|
case = get_object_or_404(CimarCase, uuid=uuid)
|
|
|
|
viewer_link = case.viewer_link + f"?sid={request.user.userprofile.cimar_sid}"
|
|
return render(request, "cimar_embed.html#embed", {"viewer_link": viewer_link})
|
|
|
|
|
|
def cimar_study_series_block(request, uuid):
|
|
case = get_object_or_404(CimarCase, uuid=uuid)
|
|
|
|
return render(
|
|
request,
|
|
"cimar_series.html",
|
|
{
|
|
"series_data": case.series_data["series"],
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|
|
|
|
|
|
def cimar_status(request):
|
|
login_status = CimarAPI(request.user.userprofile.cimar_sid).check_login_status()
|
|
|
|
if login_status:
|
|
return HttpResponse("")
|
|
else:
|
|
return render(
|
|
request,
|
|
"cimar_login.html#login-block",
|
|
{
|
|
"login_status": login_status,
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|
|
|
|
|
|
def cimar_logout(request):
|
|
CimarAPI(request.user.userprofile.cimar_sid).logout()
|
|
return HttpResponse("Logged out")
|
|
|
|
|
|
def cimar_login(request):
|
|
if request.htmx:
|
|
username = request.POST.get("username")
|
|
password = request.POST.get("password")
|
|
|
|
if not username and not password:
|
|
return HttpResponse("Please fill in login details")
|
|
|
|
try:
|
|
sid = CimarAPI().login(username, password)
|
|
|
|
request.user.userprofile.cimar_sid = sid
|
|
request.user.userprofile.save()
|
|
except InvalidCredentials:
|
|
return HttpResponse("Invalid login details")
|
|
|
|
return HttpResponse("Logged in")
|
|
|
|
else:
|
|
try:
|
|
login_status = CimarAPI(
|
|
request.user.userprofile.cimar_sid
|
|
).check_login_status()
|
|
except NoSession:
|
|
login_status = False
|
|
return render(
|
|
request,
|
|
"cimar_login.html",
|
|
{
|
|
"login_status": login_status,
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|