Files
penracourses/survey/views.py
T
2026-06-29 09:35:12 +01:00

448 lines
18 KiB
Python

import csv
from django.shortcuts import render, get_object_or_404, redirect
from django.http import HttpResponse, JsonResponse, Http404
from django.contrib.auth.decorators import login_required, user_passes_test
from django.db.models import Avg, Count
from django.contrib import messages
from django.contrib.contenttypes.models import ContentType
from django.core.exceptions import PermissionDenied
from generic.models import CidUser
from .models import Survey, SurveyQuestion, SurveyResponse, SurveyAnswer
from .forms import SurveyForm, SurveyQuestionForm, SurveyTakeForm
def is_survey_admin(user):
"""Checks if the user has survey administrative permissions."""
return user.is_authenticated and (user.is_superuser or user.groups.filter(name="cid_user_manager").exists())
def can_manage_survey(user, survey):
"""Checks if a user is authorized to manage/view administrative actions on a survey."""
if user.is_superuser:
return True
if not user.is_authenticated or not user.groups.filter(name="cid_user_manager").exists():
return False
if survey.authors_only:
return survey.is_author(user)
return True
@login_required
@user_passes_test(is_survey_admin)
def survey_list(request):
"""
Scope: Staff/Admin.
Functionality: Displays the dashboard list of all feedback surveys.
"""
if request.user.is_superuser:
surveys = Survey.objects.all()
else:
from django.db.models import Q
surveys = Survey.objects.filter(
Q(authors_only=False) | Q(author=request.user)
).distinct()
return render(request, "survey/survey_list.html", {"surveys": surveys})
@login_required
@user_passes_test(is_survey_admin)
def survey_create(request):
"""
Scope: Staff/Admin.
Functionality: Form to create a new survey.
"""
if request.method == "POST":
form = SurveyForm(request.POST)
if form.is_valid():
survey = form.save(commit=False)
survey.created_by = request.user
survey.save()
form.save_m2m()
survey.add_author(request.user)
messages.success(request, f"Survey '{survey.name}' created successfully.")
return redirect("survey:survey_questions_edit", pk=survey.pk)
else:
form = SurveyForm()
return render(request, "survey/survey_form.html", {"form": form, "action": "Create"})
@login_required
@user_passes_test(is_survey_admin)
def survey_update(request, pk):
"""
Scope: Staff/Admin.
Functionality: Form to edit an existing survey's metadata.
"""
survey = get_object_or_404(Survey, pk=pk)
if not can_manage_survey(request.user, survey):
raise PermissionDenied("You do not have permission to manage this survey.")
if request.method == "POST":
form = SurveyForm(request.POST, instance=survey)
if form.is_valid():
form.save()
messages.success(request, f"Survey '{survey.name}' updated.")
return redirect("survey:survey_list")
else:
form = SurveyForm(instance=survey)
return render(request, "survey/survey_form.html", {"form": form, "action": "Update", "survey": survey})
@login_required
@user_passes_test(is_survey_admin)
def survey_delete(request, pk):
"""
Scope: Staff/Admin.
Functionality: Deletes a survey.
"""
survey = get_object_or_404(Survey, pk=pk)
if not can_manage_survey(request.user, survey):
raise PermissionDenied("You do not have permission to manage this survey.")
if request.method == "POST":
survey.delete()
messages.success(request, "Survey deleted successfully.")
return redirect("survey:survey_list")
return render(request, "survey/survey_confirm_delete.html", {"survey": survey})
@login_required
@user_passes_test(is_survey_admin)
def survey_questions_edit(request, pk):
"""
Scope: Staff/Admin.
Functionality: Question manager. Allows adding, deleting, and reordering questions.
"""
survey = get_object_or_404(Survey, pk=pk)
if not can_manage_survey(request.user, survey):
raise PermissionDenied("You do not have permission to manage this survey.")
questions = survey.questions.all()
if request.method == "POST":
action = request.POST.get("action")
if action == "add":
form = SurveyQuestionForm(request.POST)
if form.is_valid():
q = form.save(commit=False)
q.survey = survey
# set position to end if not specified
if not q.position:
q.position = (questions.count() + 1) * 10
q.save()
messages.success(request, "Question added.")
return redirect("survey:survey_questions_edit", pk=survey.pk)
elif action == "delete":
q_id = request.POST.get("question_id")
q = get_object_or_404(SurveyQuestion, pk=q_id, survey=survey)
q.delete()
messages.success(request, "Question deleted.")
return redirect("survey:survey_questions_edit", pk=survey.pk)
elif action == "reorder":
for key, val in request.POST.items():
if key.startswith("order_") and val.isdigit():
q_pk = key.split("_")[1]
SurveyQuestion.objects.filter(pk=q_pk, survey=survey).update(position=int(val))
messages.success(request, "Question order updated.")
return redirect("survey:survey_questions_edit", pk=survey.pk)
else:
form = SurveyQuestionForm()
return render(request, "survey/survey_questions_edit.html", {
"survey": survey,
"questions": questions,
"form": form
})
def survey_take(request, pk):
"""
Scope: Candidates/Users.
Functionality: Renders survey questions and processes dynamic form submission.
Supports standard login, CID candidate login, and standalone modes.
"""
survey = get_object_or_404(Survey, pk=pk, active=True)
# Retrieve context from GET or POST
pre_or_post = request.GET.get("pre_or_post") or request.POST.get("pre_or_post") or "GENERAL"
content_type_id = request.GET.get("content_type_id") or request.POST.get("content_type_id")
object_id = request.GET.get("object_id") or request.POST.get("object_id")
next_url = request.GET.get("next") or request.POST.get("next") or "/"
mandatory = request.GET.get("mandatory") or request.POST.get("mandatory") or "true"
skip_key = request.GET.get("skip_key") or request.POST.get("skip_key") or ""
# Handle CID auth
cid = request.GET.get("cid") or request.POST.get("cid")
passcode = request.GET.get("passcode") or request.POST.get("passcode")
cid_user = None
if cid and passcode:
cid_user = CidUser.objects.filter(cid=cid).first()
if not cid_user or cid_user.passcode != passcode:
raise Http404("Candidate ID / Passcode combination not found")
elif not request.user.is_authenticated and not survey.anonymous:
# If not anonymous and no CID, require login
return redirect(f"/accounts/login/?next={request.path}")
content_type = None
if content_type_id:
try:
content_type = ContentType.objects.get_for_id(int(content_type_id))
except (ValueError, ContentType.DoesNotExist):
pass
if request.method == "POST":
form = SurveyTakeForm(request.POST, survey=survey)
if form.is_valid():
form.save(
user=request.user if request.user.is_authenticated else None,
cid=int(cid) if cid and str(cid).isdigit() else None,
content_type=content_type,
object_id=int(object_id) if object_id and str(object_id).isdigit() else None,
pre_or_post=pre_or_post
)
return render(request, "survey/survey_thanks.html", {"next_url": next_url, "after_text": survey.after_text})
else:
form = SurveyTakeForm(survey=survey)
return render(request, "survey/survey_take.html", {
"survey": survey,
"form": form,
"cid": cid,
"passcode": passcode,
"pre_or_post": pre_or_post,
"content_type_id": content_type_id,
"object_id": object_id,
"next": next_url,
"mandatory": mandatory,
"skip_key": skip_key,
})
@login_required
@user_passes_test(is_survey_admin)
def survey_results(request, pk):
"""
Scope: Staff/Admin.
Functionality: Renders survey response metrics, analytics, and text answers.
Supports filtering by source context and target collections/exams.
"""
survey = get_object_or_404(Survey, pk=pk)
if not can_manage_survey(request.user, survey):
raise PermissionDenied("You do not have permission to manage this survey.")
questions = survey.questions.all()
responses = survey.responses.all()
# Compile list of distinct exams/collections that have used this survey to build the filter dropdown
targets = []
seen_targets = set()
for resp in responses:
if resp.content_type_id and resp.object_id:
key = (resp.content_type_id, resp.object_id)
if key not in seen_targets:
seen_targets.add(key)
try:
obj = resp.content_object
if obj:
targets.append({
"content_type_id": resp.content_type_id,
"object_id": resp.object_id,
"name": str(obj)
})
except Exception:
pass
# Apply GET filters
filter_context = request.GET.get("filter_context")
if filter_context in [SurveyResponse.SurveyContext.PRE, SurveyResponse.SurveyContext.POST, SurveyResponse.SurveyContext.GENERAL]:
responses = responses.filter(pre_or_post=filter_context)
filter_target = request.GET.get("filter_target")
if filter_target:
try:
ct_id, obj_id = filter_target.split(":")
responses = responses.filter(content_type_id=int(ct_id), object_id=int(obj_id))
except (ValueError, TypeError):
pass
responses_count = responses.count()
analytics = []
for q in questions:
answers = q.answers.filter(response__in=responses)
q_data = {
"question": q,
"answers_count": answers.count()
}
if q.question_type == SurveyQuestion.QuestionType.TEXT:
q_data["text_answers"] = answers.values_list("text_answer", flat=True).exclude(text_answer__isnull=True).exclude(text_answer="")
elif q.question_type == SurveyQuestion.QuestionType.CHOICE:
# Count choices
choice_counts = answers.values("choice_answer").annotate(count=Count("id")).order_by("-count")
q_data["choice_counts"] = []
for cc in choice_counts:
ans_text = cc["choice_answer"]
if ans_text:
pct = round((cc["count"] / q_data["answers_count"]) * 100, 1) if q_data["answers_count"] else 0
q_data["choice_counts"].append({
"choice": ans_text,
"count": cc["count"],
"percentage": pct
})
elif q.question_type == SurveyQuestion.QuestionType.RATING:
# Stats for rating
avg = answers.aggregate(avg=Avg("rating_answer"))["avg"]
q_data["average_rating"] = round(avg, 2) if avg is not None else "N/A"
# Rating counts
rating_counts = answers.values("rating_answer").annotate(count=Count("id"))
counts_dict = {i: 0 for i in range(1, 6)}
for rc in rating_counts:
r = rc["rating_answer"]
if r in counts_dict:
counts_dict[r] = rc["count"]
q_data["rating_breakdown"] = []
for r in range(1, 6):
count = counts_dict[r]
pct = round((count / q_data["answers_count"]) * 100, 1) if q_data["answers_count"] else 0
q_data["rating_breakdown"].append({
"rating": r,
"count": count,
"percentage": pct
})
elif q.question_type == SurveyQuestion.QuestionType.INTEGER:
avg = answers.aggregate(avg=Avg("integer_answer"))["avg"]
q_data["average_integer"] = round(avg, 2) if avg is not None else "N/A"
q_data["integer_answers"] = list(answers.values_list("integer_answer", flat=True).exclude(integer_answer__isnull=True))
elif q.question_type == SurveyQuestion.QuestionType.DATE:
q_data["date_answers"] = list(answers.values_list("date_answer", flat=True).exclude(date_answer__isnull=True))
analytics.append(q_data)
return render(request, "survey/survey_results.html", {
"survey": survey,
"responses_count": responses_count,
"analytics": analytics,
"targets": targets,
"filter_context": filter_context,
"filter_target": filter_target,
})
@login_required
@user_passes_test(is_survey_admin)
def survey_export_csv(request, pk):
"""
Scope: Staff/Admin.
Functionality: Exports filtered responses for a survey as a downloadable CSV.
"""
survey = get_object_or_404(Survey, pk=pk)
if not can_manage_survey(request.user, survey):
raise PermissionDenied("You do not have permission to manage this survey.")
questions = list(survey.questions.all().order_by("position", "id"))
responses = survey.responses.all()
# Apply GET filters to match CSV export output
filter_context = request.GET.get("filter_context")
if filter_context in [SurveyResponse.SurveyContext.PRE, SurveyResponse.SurveyContext.POST, SurveyResponse.SurveyContext.GENERAL]:
responses = responses.filter(pre_or_post=filter_context)
filter_target = request.GET.get("filter_target")
if filter_target:
try:
ct_id, obj_id = filter_target.split(":")
responses = responses.filter(content_type_id=int(ct_id), object_id=int(obj_id))
except (ValueError, TypeError):
pass
response = HttpResponse(content_type="text/csv")
response["Content-Disposition"] = f'attachment; filename="survey_{survey.pk}_responses.csv"'
writer = csv.writer(response)
# Header row
headers = ["Response ID", "Timestamp", "User", "CID", "Context Type", "Context ID", "Pre/Post"]
for q in questions:
headers.append(f"Q: {q.text}")
writer.writerow(headers)
# Data rows
for resp in responses:
user_str = resp.user.username if resp.user else ""
cid_str = str(resp.cid) if resp.cid else ""
ct_str = f"{resp.content_type.app_label}.{resp.content_type.model}" if resp.content_type else ""
object_id_str = str(resp.object_id) if resp.object_id else ""
row = [
resp.pk,
resp.created_at.strftime("%Y-%m-%d %H:%M:%S"),
user_str,
cid_str,
ct_str,
object_id_str,
resp.pre_or_post
]
# Get answers map
answers_map = {ans.question_id: ans for ans in resp.answers.all()}
for q in questions:
ans = answers_map.get(q.pk)
if not ans:
row.append("")
elif q.question_type == SurveyQuestion.QuestionType.TEXT:
row.append(ans.text_answer or "")
elif q.question_type == SurveyQuestion.QuestionType.CHOICE:
row.append(ans.choice_answer or "")
elif q.question_type == SurveyQuestion.QuestionType.RATING:
row.append(str(ans.rating_answer) if ans.rating_answer is not None else "")
elif q.question_type == SurveyQuestion.QuestionType.INTEGER:
row.append(str(ans.integer_answer) if ans.integer_answer is not None else "")
elif q.question_type == SurveyQuestion.QuestionType.DATE:
row.append(ans.date_answer.strftime("%Y-%m-%d") if ans.date_answer else "")
writer.writerow(row)
return response
def check_survey_interception(request, exam_or_collection, cid=None, passcode=None):
"""
Checks if a pre_survey is configured for this collection/exam and if the user/cid
has already responded to it or skipped it. If not, redirects them to take the survey.
"""
if exam_or_collection.pre_survey:
session_key = f"skipped_pre_survey_{exam_or_collection.pk}"
if request.session.get(session_key):
return None
user_param = request.user if request.user.is_authenticated else None
cid_param = int(cid) if cid and str(cid).isdigit() else None
ct = ContentType.objects.get_for_model(exam_or_collection)
has_responded = SurveyResponse.objects.filter(
survey=exam_or_collection.pre_survey,
user=user_param,
cid=cid_param,
content_type=ct,
object_id=exam_or_collection.pk,
pre_or_post=SurveyResponse.SurveyContext.PRE
).exists()
if not has_responded:
from django.urls import reverse
survey_url = reverse("survey:survey_take", kwargs={"pk": exam_or_collection.pre_survey.pk})
is_mandatory = getattr(exam_or_collection, "pre_survey_mandatory", False)
mandatory_str = "true" if is_mandatory else "false"
params = f"?pre_or_post=PRE&content_type_id={ct.id}&object_id={exam_or_collection.pk}&next={request.get_full_path()}&mandatory={mandatory_str}&skip_key={session_key}"
if cid and passcode:
params += f"&cid={cid}&passcode={passcode}"
return redirect(f"{survey_url}{params}")
return None
def survey_skip(request):
"""
Scope: Candidates/Users.
Functionality: Sets a session variable to bypass an optional survey, then redirects.
"""
next_url = request.GET.get("next") or "/"
key = request.GET.get("key")
if key:
request.session[key] = True
return redirect(next_url)