569 lines
20 KiB
Python
569 lines
20 KiB
Python
"""
|
|
Public survey views - Token-based survey forms (no login required)
|
|
"""
|
|
|
|
import json
|
|
import time
|
|
|
|
from django.contrib import messages
|
|
from django.db.models import Q
|
|
from django.http import JsonResponse
|
|
from django.shortcuts import get_object_or_404, redirect, render
|
|
from django.utils import timezone
|
|
from django.views.decorators.http import require_http_methods
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from user_agents import parse
|
|
|
|
from apps.core.services import AuditService
|
|
|
|
from .models import SurveyInstance, SurveyQuestion, SurveyResponse, SurveyTracking
|
|
from .analytics import track_survey_open, track_survey_completion
|
|
from .routing import get_next_question, resolve_question_path
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def survey_form(request, token):
|
|
"""
|
|
Public survey form - accessible via secure token link.
|
|
|
|
Features:
|
|
- No login required
|
|
- Token-based access
|
|
- Mobile-first responsive design
|
|
- Bilingual support (AR/EN)
|
|
- Progress indicator
|
|
- Question type rendering
|
|
- Form validation
|
|
"""
|
|
# Get survey instance by token
|
|
# Allow access until survey is completed or token expires (2 days by default)
|
|
try:
|
|
survey = (
|
|
SurveyInstance.objects.select_related("survey_template", "patient", "journey_instance")
|
|
.prefetch_related("survey_template__questions")
|
|
.get(
|
|
access_token=token,
|
|
status__in=["pending", "sent", "viewed", "in_progress"],
|
|
token_expires_at__gt=timezone.now(),
|
|
)
|
|
)
|
|
except SurveyInstance.DoesNotExist:
|
|
return render(request, "surveys/invalid_token.html", {"error": "invalid_or_expired"})
|
|
|
|
# Track survey open - increment count and record tracking event
|
|
# Get device info from user agent
|
|
user_agent_str = request.META.get("HTTP_USER_AGENT", "")
|
|
ip_address = request.META.get("REMOTE_ADDR", "")
|
|
|
|
# Parse user agent for device info
|
|
user_agent = parse(user_agent_str)
|
|
device_type = "mobile" if user_agent.is_mobile else ("tablet" if user_agent.is_tablet else "desktop")
|
|
browser = f"{user_agent.browser.family} {user_agent.browser.version_string}"
|
|
|
|
# Update survey instance tracking fields
|
|
survey.open_count += 1
|
|
survey.last_opened_at = timezone.now()
|
|
|
|
# Update status based on current state
|
|
if not survey.opened_at:
|
|
survey.opened_at = timezone.now()
|
|
survey.status = "viewed"
|
|
elif survey.status == "sent":
|
|
survey.status = "viewed"
|
|
|
|
survey.save(update_fields=["open_count", "last_opened_at", "opened_at", "status"])
|
|
|
|
# Track page view event
|
|
SurveyTracking.track_event(
|
|
survey,
|
|
"page_view",
|
|
user_agent=user_agent_str[:500] if user_agent_str else "",
|
|
ip_address=ip_address,
|
|
device_type=device_type,
|
|
browser=browser,
|
|
metadata={
|
|
"referrer": request.META.get("HTTP_REFERER", ""),
|
|
"language": request.GET.get("lang", "en"),
|
|
},
|
|
)
|
|
|
|
# Get questions — filter by patient's experienced events
|
|
patient_events = set(survey.metadata.get("event_types", []))
|
|
questions = survey.survey_template.questions.filter(Q(is_base=True) | Q(event_type__in=patient_events)).order_by(
|
|
"order"
|
|
)
|
|
|
|
routing_rules = list(
|
|
survey.survey_template.routing_rules.select_related("source_question", "target_question").order_by(
|
|
"source_question__order", "order"
|
|
)
|
|
)
|
|
|
|
questions_data = resolve_question_path(questions, routing_rules)
|
|
|
|
if request.method == "POST":
|
|
# Process survey responses
|
|
language = request.POST.get("language", "en")
|
|
errors = []
|
|
responses_data = []
|
|
|
|
# Validate and collect responses
|
|
for question in questions:
|
|
field_name = f"question_{question.id}"
|
|
|
|
# Check if required
|
|
if question.is_required and not request.POST.get(field_name):
|
|
errors.append(f"Question {question.order + 1} is required")
|
|
continue
|
|
|
|
# Get response value based on question type
|
|
if question.question_type in ["rating", "likert"]:
|
|
numeric_value = request.POST.get(field_name)
|
|
if numeric_value:
|
|
responses_data.append(
|
|
{
|
|
"question": question,
|
|
"numeric_value": float(numeric_value),
|
|
"text_value": "",
|
|
"choice_value": "",
|
|
}
|
|
)
|
|
|
|
elif question.question_type == "nps":
|
|
numeric_value = request.POST.get(field_name)
|
|
if numeric_value:
|
|
responses_data.append(
|
|
{
|
|
"question": question,
|
|
"numeric_value": float(numeric_value),
|
|
"text_value": "",
|
|
"choice_value": "",
|
|
}
|
|
)
|
|
|
|
elif question.question_type == "yes_no":
|
|
choice_value = request.POST.get(field_name)
|
|
if choice_value:
|
|
# Convert yes/no to numeric for scoring
|
|
numeric_value = 5.0 if choice_value == "yes" else 1.0
|
|
responses_data.append(
|
|
{
|
|
"question": question,
|
|
"numeric_value": numeric_value,
|
|
"text_value": "",
|
|
"choice_value": choice_value,
|
|
}
|
|
)
|
|
|
|
elif question.question_type == "multiple_choice":
|
|
choice_value = request.POST.get(field_name)
|
|
if choice_value:
|
|
# Find the selected choice to get its label
|
|
selected_choice = None
|
|
for choice in question.choices_json:
|
|
if str(choice.get("value", "")) == str(choice_value):
|
|
selected_choice = choice
|
|
break
|
|
|
|
# Get the label based on language
|
|
language = request.POST.get("language", "en")
|
|
if language == "ar" and selected_choice and selected_choice.get("label_ar"):
|
|
text_value = selected_choice["label_ar"]
|
|
elif selected_choice and selected_choice.get("label"):
|
|
text_value = selected_choice["label"]
|
|
else:
|
|
text_value = choice_value
|
|
|
|
# Try to convert choice value to numeric for scoring
|
|
try:
|
|
numeric_value = float(choice_value)
|
|
except (ValueError, TypeError):
|
|
numeric_value = None
|
|
|
|
responses_data.append(
|
|
{
|
|
"question": question,
|
|
"numeric_value": numeric_value,
|
|
"text_value": text_value,
|
|
"choice_value": choice_value,
|
|
}
|
|
)
|
|
|
|
elif question.question_type in ["text", "textarea"]:
|
|
text_value = request.POST.get(field_name, "")
|
|
if text_value:
|
|
responses_data.append(
|
|
{"question": question, "numeric_value": None, "text_value": text_value, "choice_value": ""}
|
|
)
|
|
|
|
# Get optional comment
|
|
comment = request.POST.get("comment", "").strip()
|
|
|
|
# If validation errors, show form again
|
|
if errors:
|
|
context = {
|
|
"survey": survey,
|
|
"questions": questions,
|
|
"errors": errors,
|
|
"language": language,
|
|
}
|
|
return render(request, "surveys/public_form.html", context)
|
|
|
|
# Save responses
|
|
for response_data in responses_data:
|
|
SurveyResponse.objects.update_or_create(
|
|
survey_instance=survey,
|
|
question=response_data["question"],
|
|
defaults={
|
|
"numeric_value": response_data["numeric_value"],
|
|
"text_value": response_data["text_value"],
|
|
"choice_value": response_data["choice_value"],
|
|
},
|
|
)
|
|
|
|
# Update survey status
|
|
survey.status = "completed"
|
|
survey.completed_at = timezone.now()
|
|
|
|
# Calculate time spent (from opened_at to completed_at)
|
|
if survey.opened_at:
|
|
time_spent = (timezone.now() - survey.opened_at).total_seconds()
|
|
survey.time_spent_seconds = int(time_spent)
|
|
|
|
survey.completed_language = language
|
|
survey.save(update_fields=["status", "completed_at", "time_spent_seconds", "completed_language"])
|
|
|
|
# Track completion event
|
|
SurveyTracking.track_event(
|
|
survey,
|
|
"survey_completed",
|
|
total_time_spent=survey.time_spent_seconds,
|
|
user_agent=user_agent_str[:500] if user_agent_str else "",
|
|
ip_address=ip_address,
|
|
metadata={
|
|
"response_count": len(responses_data),
|
|
"language": language,
|
|
},
|
|
)
|
|
|
|
# Calculate score
|
|
score = survey.calculate_score()
|
|
|
|
# Log completion
|
|
AuditService.log_event(
|
|
event_type="survey_completed",
|
|
description=f"Survey completed: {survey.survey_template.name}",
|
|
user=None,
|
|
content_object=survey,
|
|
metadata={
|
|
"score": float(score) if score else None,
|
|
"is_negative": survey.is_negative,
|
|
"response_count": len(responses_data),
|
|
"has_comment": bool(comment),
|
|
},
|
|
)
|
|
|
|
# Save comment and trigger AI analysis if present
|
|
if comment:
|
|
survey.comment = comment
|
|
survey.save(update_fields=["comment"])
|
|
|
|
# Trigger background task for comment analysis
|
|
from apps.surveys.tasks import analyze_survey_comment
|
|
|
|
try:
|
|
analyze_survey_comment.delay(str(survey.id))
|
|
except Exception as e:
|
|
# Log but don't fail the survey submission
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.error(f"Failed to trigger comment analysis: {str(e)}")
|
|
|
|
# Create PX action if negative
|
|
if survey.is_negative:
|
|
from apps.surveys.tasks import create_action_from_negative_survey
|
|
|
|
try:
|
|
create_action_from_negative_survey.delay(str(survey.id))
|
|
except Exception as e:
|
|
# Log but don't fail the survey submission
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger.error(f"Failed to trigger action creation: {str(e)}")
|
|
|
|
# Redirect to thank you page with language
|
|
return redirect(f"/surveys/s/{token}/thank-you/?lang={language}")
|
|
|
|
# GET request - show form
|
|
# Determine language from query param or browser
|
|
language = request.GET.get("lang", "en")
|
|
|
|
# Serialize routing rules for client-side evaluation
|
|
routing_rules_json = json.dumps(
|
|
[
|
|
{
|
|
"id": str(r.id),
|
|
"source_question": str(r.source_question_id),
|
|
"operator": r.operator,
|
|
"value": r.value,
|
|
"action": r.action,
|
|
"target_question": str(r.target_question_id) if r.target_question_id else None,
|
|
"order": r.order,
|
|
}
|
|
for r in routing_rules
|
|
]
|
|
)
|
|
|
|
questions_json = json.dumps(questions_data)
|
|
|
|
context = {
|
|
"survey": survey,
|
|
"questions": questions,
|
|
"questions_json": questions_json,
|
|
"routing_rules_json": routing_rules_json,
|
|
"language": language,
|
|
"total_questions": len(questions_data),
|
|
}
|
|
|
|
return render(request, "surveys/public_form.html", context)
|
|
|
|
|
|
def thank_you(request, token):
|
|
"""Thank you page after survey completion"""
|
|
try:
|
|
survey = SurveyInstance.objects.select_related("survey_template", "patient").get(
|
|
access_token=token, status="completed"
|
|
)
|
|
except SurveyInstance.DoesNotExist:
|
|
return render(request, "surveys/invalid_token.html", {"error": "not_found"})
|
|
|
|
language = request.GET.get("lang", "en")
|
|
|
|
context = {
|
|
"survey": survey,
|
|
"language": language,
|
|
}
|
|
|
|
return render(request, "surveys/thank_you.html", context)
|
|
|
|
|
|
def invalid_token(request):
|
|
"""Invalid or expired token page"""
|
|
return render(request, "surveys/invalid_token.html")
|
|
|
|
|
|
@csrf_exempt
|
|
@require_http_methods(["POST"])
|
|
def track_survey_start(request, token):
|
|
"""
|
|
API endpoint to track when patient starts answering survey.
|
|
|
|
Called via AJAX when patient first interacts with the form.
|
|
Updates status from 'viewed' to 'in_progress'.
|
|
"""
|
|
try:
|
|
# Get survey instance
|
|
survey = SurveyInstance.objects.get(
|
|
access_token=token, status__in=["viewed", "in_progress"], token_expires_at__gt=timezone.now()
|
|
)
|
|
|
|
# Only update if not already in_progress
|
|
if survey.status == "viewed":
|
|
survey.status = "in_progress"
|
|
survey.save(update_fields=["status"])
|
|
|
|
# Track survey started event
|
|
SurveyTracking.track_event(
|
|
survey,
|
|
"survey_started",
|
|
user_agent=request.META.get("HTTP_USER_AGENT", "")[:500] if request.META.get("HTTP_USER_AGENT") else "",
|
|
ip_address=request.META.get("REMOTE_ADDR", ""),
|
|
metadata={
|
|
"referrer": request.META.get("HTTP_REFERER", ""),
|
|
},
|
|
)
|
|
|
|
return JsonResponse(
|
|
{
|
|
"status": "success",
|
|
"survey_status": survey.status,
|
|
}
|
|
)
|
|
|
|
except SurveyInstance.DoesNotExist:
|
|
return JsonResponse({"status": "error", "message": "Survey not found or invalid token"}, status=404)
|
|
|
|
|
|
@csrf_exempt
|
|
@require_http_methods(["POST"])
|
|
def submit_answer(request, token):
|
|
"""
|
|
AJAX endpoint for step-by-step survey answering.
|
|
|
|
Accepts a single answer, saves it, evaluates routing rules,
|
|
and returns the next question or completion status.
|
|
|
|
POST body (JSON):
|
|
{
|
|
"question_id": "uuid",
|
|
"answer": "value",
|
|
"response_time_seconds": 3.2
|
|
}
|
|
"""
|
|
try:
|
|
survey = SurveyInstance.objects.select_related("survey_template").get(
|
|
access_token=token,
|
|
status__in=["pending", "sent", "viewed", "in_progress"],
|
|
token_expires_at__gt=timezone.now(),
|
|
)
|
|
except SurveyInstance.DoesNotExist:
|
|
return JsonResponse({"status": "error", "message": "Survey not found or expired"}, status=404)
|
|
|
|
if survey.status == "viewed":
|
|
survey.status = "in_progress"
|
|
survey.save(update_fields=["status"])
|
|
|
|
try:
|
|
data = json.loads(request.body)
|
|
except json.JSONDecodeError:
|
|
return JsonResponse({"status": "error", "message": "Invalid JSON"}, status=400)
|
|
|
|
question_id = data.get("question_id")
|
|
answer_value = data.get("answer")
|
|
response_time = data.get("response_time_seconds")
|
|
|
|
if not question_id:
|
|
return JsonResponse({"status": "error", "message": "question_id is required"}, status=400)
|
|
|
|
try:
|
|
question = survey.survey_template.questions.get(id=question_id)
|
|
except SurveyQuestion.DoesNotExist:
|
|
return JsonResponse({"status": "error", "message": "Question not found"}, status=404)
|
|
|
|
numeric_value = None
|
|
text_value = ""
|
|
choice_value = ""
|
|
|
|
if question.question_type in ["rating", "likert"]:
|
|
if answer_value is not None:
|
|
numeric_value = float(answer_value)
|
|
elif question.question_type == "nps":
|
|
if answer_value is not None:
|
|
numeric_value = float(answer_value)
|
|
elif question.question_type == "yes_no":
|
|
choice_value = str(answer_value)
|
|
numeric_value = 5.0 if choice_value == "yes" else 1.0
|
|
elif question.question_type == "multiple_choice":
|
|
choice_value = str(answer_value)
|
|
try:
|
|
numeric_value = float(choice_value)
|
|
except (ValueError, TypeError):
|
|
numeric_value = None
|
|
for choice in question.choices_json or []:
|
|
if str(choice.get("value", "")) == choice_value:
|
|
text_value = choice.get("label", choice_value)
|
|
break
|
|
elif question.question_type in ["text", "textarea"]:
|
|
text_value = str(answer_value) if answer_value else ""
|
|
|
|
if question.is_required and not answer_value:
|
|
return JsonResponse({"status": "error", "message": "This question is required"}, status=400)
|
|
|
|
SurveyResponse.objects.update_or_create(
|
|
survey_instance=survey,
|
|
question=question,
|
|
defaults={
|
|
"numeric_value": numeric_value,
|
|
"text_value": text_value,
|
|
"choice_value": choice_value,
|
|
"response_time_seconds": response_time,
|
|
},
|
|
)
|
|
|
|
SurveyTracking.track_event(
|
|
survey,
|
|
"question_answered",
|
|
current_question=question.order,
|
|
metadata={"question_id": str(question.id), "question_type": question.question_type},
|
|
)
|
|
|
|
patient_events = set(survey.metadata.get("event_types", []))
|
|
all_questions = list(
|
|
survey.survey_template.questions.filter(Q(is_base=True) | Q(event_type__in=patient_events)).order_by("order")
|
|
)
|
|
routing_rules = list(
|
|
survey.survey_template.routing_rules.select_related("source_question", "target_question").order_by(
|
|
"source_question__order", "order"
|
|
)
|
|
)
|
|
|
|
questions_data = resolve_question_path(all_questions, routing_rules)
|
|
rules_serialized = [
|
|
{
|
|
"id": str(r.id),
|
|
"source_question": str(r.source_question_id),
|
|
"operator": r.operator,
|
|
"value": r.value,
|
|
"action": r.action,
|
|
"target_question": str(r.target_question_id) if r.target_question_id else None,
|
|
"order": r.order,
|
|
}
|
|
for r in routing_rules
|
|
]
|
|
|
|
routing_result = get_next_question(
|
|
current_question_id=str(question.id),
|
|
answer_value=answer_value,
|
|
all_questions=questions_data,
|
|
routing_rules=rules_serialized,
|
|
)
|
|
|
|
if routing_result.is_complete:
|
|
return JsonResponse(
|
|
{
|
|
"status": "complete",
|
|
"is_complete": True,
|
|
"progress": 100,
|
|
}
|
|
)
|
|
|
|
next_q = None
|
|
if routing_result.next_question_id:
|
|
for q in questions_data:
|
|
if q["id"] == routing_result.next_question_id:
|
|
next_q = q
|
|
break
|
|
elif routing_result.action == "next":
|
|
current_idx = None
|
|
for i, q in enumerate(questions_data):
|
|
if q["id"] == str(question.id):
|
|
current_idx = i
|
|
break
|
|
if current_idx is not None and current_idx + 1 < len(questions_data):
|
|
next_q = questions_data[current_idx + 1]
|
|
|
|
if next_q is None:
|
|
return JsonResponse(
|
|
{
|
|
"status": "complete",
|
|
"is_complete": True,
|
|
"progress": 100,
|
|
}
|
|
)
|
|
|
|
answered_count = len(questions_data) - len(
|
|
[q for q in questions_data[current_idx + 1 :] if q["id"] == next_q["id"]]
|
|
)
|
|
progress = round((answered_count / len(questions_data)) * 100) if questions_data else 100
|
|
|
|
return JsonResponse(
|
|
{
|
|
"status": "ok",
|
|
"next_question": next_q,
|
|
"is_complete": False,
|
|
"progress": min(progress, 100),
|
|
}
|
|
)
|