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)