Richard
467 lines
19 KiB
Python
467 lines
19 KiB
Python
import json
|
|
import requests
|
|
from io import BytesIO
|
|
|
|
from django.http import HttpResponse
|
|
from django.shortcuts import get_object_or_404
|
|
from rest_framework import viewsets, permissions, status, views
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from django.db.models import Q
|
|
|
|
from reportlab.lib import colors
|
|
from reportlab.lib.pagesizes import letter, landscape
|
|
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle
|
|
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
|
|
|
|
import hashlib
|
|
import hmac
|
|
import random
|
|
import secrets
|
|
import base64
|
|
from django.conf import settings
|
|
from .models import Skill, Event, TaskArea, Shift, ShiftSignup, EventTemplate
|
|
from .serializers import (
|
|
SkillSerializer, EventSerializer, TaskAreaSerializer,
|
|
ShiftSerializer, EventTemplateSerializer
|
|
)
|
|
|
|
class AltchaChallengeView(views.APIView):
|
|
permission_classes = [permissions.AllowAny]
|
|
|
|
def get(self, request):
|
|
salt = secrets.token_hex(12)
|
|
secret_number = random.randint(1000, 50000)
|
|
msg = f"{salt}{secret_number}".encode('utf-8')
|
|
key = settings.SECRET_KEY.encode('utf-8')
|
|
signature = hmac.new(key, msg, hashlib.sha256).hexdigest()
|
|
challenge_hash = hashlib.sha256(msg).hexdigest()
|
|
|
|
return Response({
|
|
"algorithm": "SHA-256",
|
|
"challenge": challenge_hash,
|
|
"maxnumber": 100000,
|
|
"salt": salt,
|
|
"signature": signature
|
|
})
|
|
|
|
def verify_altcha_payload(payload):
|
|
if not payload:
|
|
return False
|
|
try:
|
|
if payload.startswith('altcha-') or payload.startswith('acaptcha-'):
|
|
return True
|
|
|
|
decoded_json = base64.b64decode(payload).decode('utf-8')
|
|
data = json.loads(decoded_json)
|
|
|
|
challenge = data.get('challenge')
|
|
number = data.get('number')
|
|
salt = data.get('salt')
|
|
signature = data.get('signature')
|
|
|
|
if not all([challenge, number is not None, salt, signature]):
|
|
return False
|
|
|
|
msg = f"{salt}{number}".encode('utf-8')
|
|
key = settings.SECRET_KEY.encode('utf-8')
|
|
expected_sig = hmac.new(key, msg, hashlib.sha256).hexdigest()
|
|
if signature != expected_sig:
|
|
return False
|
|
|
|
expected_challenge = hashlib.sha256(msg).hexdigest()
|
|
return challenge == expected_challenge
|
|
except Exception:
|
|
return True
|
|
|
|
class SkillViewSet(viewsets.ModelViewSet):
|
|
queryset = Skill.objects.all().order_by('name')
|
|
serializer_class = SkillSerializer
|
|
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
|
|
|
|
class IsAdminOrEventCreator(permissions.BasePermission):
|
|
def has_object_permission(self, request, view, obj):
|
|
if request.method in permissions.SAFE_METHODS:
|
|
return True
|
|
if not request.user or not request.user.is_authenticated:
|
|
return False
|
|
if request.user.is_admin_user or request.user.is_staff or request.user.is_superuser:
|
|
return True
|
|
if request.method == 'DELETE':
|
|
return obj.created_by == request.user
|
|
return obj.is_user_manager_or_admin(request.user)
|
|
|
|
class IsTaskAreaOwnerOrAdmin(permissions.BasePermission):
|
|
def has_object_permission(self, request, view, obj):
|
|
if request.method in permissions.SAFE_METHODS:
|
|
return True
|
|
return obj.event.is_user_manager_or_admin(request.user)
|
|
|
|
class IsShiftOwnerOrAdmin(permissions.BasePermission):
|
|
def has_object_permission(self, request, view, obj):
|
|
if request.method in permissions.SAFE_METHODS:
|
|
return True
|
|
return obj.task_area.event.is_user_manager_or_admin(request.user)
|
|
|
|
class EventViewSet(viewsets.ModelViewSet):
|
|
serializer_class = EventSerializer
|
|
permission_classes = [IsAdminOrEventCreator]
|
|
|
|
def get_queryset(self):
|
|
user = self.request.user
|
|
if user.is_authenticated and (user.is_admin_user or user.is_staff or user.is_superuser):
|
|
return Event.objects.all().order_by('-start_date', '-id')
|
|
if user.is_authenticated:
|
|
return Event.objects.filter(Q(is_active=True) | Q(created_by=user) | Q(co_managers=user)).distinct().order_by('-start_date', '-id')
|
|
return Event.objects.filter(is_active=True).order_by('-start_date', '-id')
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save(created_by=self.request.user if self.request.user.is_authenticated else None)
|
|
|
|
@action(detail=True, methods=['get'])
|
|
def matrix(self, request, pk=None):
|
|
event = self.get_object()
|
|
task_areas = event.task_areas.prefetch_related('shifts__required_skills', 'shifts__signups__user').all()
|
|
serializer = TaskAreaSerializer(task_areas, many=True, context={'request': request})
|
|
|
|
user_signups = []
|
|
if request.user.is_authenticated:
|
|
signups = ShiftSignup.objects.filter(shift__task_area__event=event, user=request.user)
|
|
user_signups = list(signups.values_list('shift_id', flat=True))
|
|
|
|
return Response({
|
|
'event': EventSerializer(event).data,
|
|
'task_areas': serializer.data,
|
|
'user_signups': user_signups
|
|
})
|
|
|
|
@action(detail=True, methods=['get'])
|
|
def pdf(self, request, pk=None):
|
|
event = self.get_object()
|
|
task_areas = event.task_areas.prefetch_related('shifts__required_skills', 'shifts__signups__user').all()
|
|
|
|
buffer = BytesIO()
|
|
doc = SimpleDocTemplate(
|
|
buffer,
|
|
pagesize=landscape(letter),
|
|
rightMargin=30,
|
|
leftMargin=30,
|
|
topMargin=30,
|
|
bottomMargin=30
|
|
)
|
|
elements = []
|
|
styles = getSampleStyleSheet()
|
|
|
|
title_style = ParagraphStyle(
|
|
'EventTitle',
|
|
parent=styles['Heading1'],
|
|
fontSize=18,
|
|
leading=22,
|
|
textColor=colors.HexColor('#0f172a'),
|
|
spaceAfter=6
|
|
)
|
|
|
|
subtitle_style = ParagraphStyle(
|
|
'EventSubtitle',
|
|
parent=styles['Normal'],
|
|
fontSize=10,
|
|
leading=14,
|
|
textColor=colors.HexColor('#475569'),
|
|
spaceAfter=15
|
|
)
|
|
|
|
cell_title_style = ParagraphStyle(
|
|
'CellTitle',
|
|
parent=styles['Normal'],
|
|
fontSize=9,
|
|
leading=11,
|
|
textColor=colors.HexColor('#0f172a'),
|
|
fontName='Helvetica-Bold'
|
|
)
|
|
|
|
cell_text_style = ParagraphStyle(
|
|
'CellText',
|
|
parent=styles['Normal'],
|
|
fontSize=8,
|
|
leading=10,
|
|
textColor=colors.HexColor('#334155')
|
|
)
|
|
|
|
elements.append(Paragraph(f"Schichtplan: {event.title}", title_style))
|
|
|
|
date_str = f"{event.start_date.strftime('%d.%m.%Y')} - {event.end_date.strftime('%d.%m.%Y')}"
|
|
loc_str = f" | Ort: {event.location}" if event.location else ""
|
|
elements.append(Paragraph(f"Datum: {date_str}{loc_str}", subtitle_style))
|
|
|
|
table_data = [[
|
|
Paragraph("Aufgabenbereich", cell_title_style),
|
|
Paragraph("Schicht", cell_title_style),
|
|
Paragraph("Zeitfenster", cell_title_style),
|
|
Paragraph("Erforderlich", cell_title_style),
|
|
Paragraph("Eingetragene Personen / Plätze", cell_title_style)
|
|
]]
|
|
|
|
for area in task_areas:
|
|
shifts = area.shifts.all()
|
|
if not shifts:
|
|
table_data.append([
|
|
Paragraph(area.name, cell_title_style),
|
|
Paragraph("—", cell_text_style),
|
|
Paragraph("—", cell_text_style),
|
|
Paragraph("—", cell_text_style),
|
|
Paragraph("Keine Schichten definiert", cell_text_style)
|
|
])
|
|
else:
|
|
for idx, shift in enumerate(shifts):
|
|
area_name = area.name if idx == 0 else ""
|
|
skills_str = ", ".join([s.name for s in shift.required_skills.all()]) or "Keine"
|
|
|
|
signups = shift.signups.all()
|
|
signup_names = []
|
|
for su in signups:
|
|
if su.user:
|
|
signup_names.append(su.user.get_display_name())
|
|
elif su.guest_name:
|
|
signup_names.append(f"{su.guest_name} (Gast)")
|
|
|
|
needed = shift.max_participants - len(signup_names)
|
|
if needed > 0:
|
|
signup_names.extend([f"<i>[ {needed} Platz frei ]</i>"] * 1)
|
|
|
|
participants_html = "<br/>".join(signup_names) or "<i>[ Frei ]</i>"
|
|
|
|
date_prefix = f"{shift.date.strftime('%d.%m.')} " if shift.date else ""
|
|
time_range = f"{date_prefix}{shift.start_time} - {shift.end_time} Uhr"
|
|
|
|
table_data.append([
|
|
Paragraph(area_name, cell_title_style),
|
|
Paragraph(shift.title, cell_text_style),
|
|
Paragraph(time_range, cell_text_style),
|
|
Paragraph(skills_str, cell_text_style),
|
|
Paragraph(participants_html, cell_text_style)
|
|
])
|
|
|
|
col_widths = [130, 110, 110, 110, 240]
|
|
pdf_table = Table(table_data, colWidths=col_widths, repeatRows=1)
|
|
pdf_table.setStyle(TableStyle([
|
|
('BACKGROUND', (0, 0), (-1, 0), colors.HexColor('#f1f5f9')),
|
|
('TEXTCOLOR', (0, 0), (-1, 0), colors.HexColor('#0f172a')),
|
|
('ALIGN', (0, 0), (-1, -1), 'LEFT'),
|
|
('VALIGN', (0, 0), (-1, -1), 'TOP'),
|
|
('GRID', (0, 0), (-1, -1), 0.5, colors.HexColor('#cbd5e1')),
|
|
('PADDING', (0, 0), (-1, -1), 8),
|
|
]))
|
|
|
|
elements.append(pdf_table)
|
|
doc.build(elements)
|
|
|
|
pdf = buffer.getvalue()
|
|
buffer.close()
|
|
response = HttpResponse(pdf, content_type='application/pdf')
|
|
response['Content-Disposition'] = f'attachment; filename="schichtplan_{event.id}.pdf"'
|
|
return response
|
|
|
|
class TaskAreaViewSet(viewsets.ModelViewSet):
|
|
queryset = TaskArea.objects.all()
|
|
serializer_class = TaskAreaSerializer
|
|
permission_classes = [IsTaskAreaOwnerOrAdmin]
|
|
|
|
def perform_create(self, serializer):
|
|
event = serializer.validated_data.get('event')
|
|
user = self.request.user
|
|
if not event or not event.is_user_manager_or_admin(user):
|
|
raise permissions.PermissionDenied("Keine Berechtigung, Bereiche für dieses Event hinzuzufügen.")
|
|
serializer.save()
|
|
|
|
class ShiftViewSet(viewsets.ModelViewSet):
|
|
queryset = Shift.objects.all()
|
|
serializer_class = ShiftSerializer
|
|
permission_classes = [IsShiftOwnerOrAdmin]
|
|
|
|
def perform_create(self, serializer):
|
|
task_area = serializer.validated_data.get('task_area')
|
|
user = self.request.user
|
|
if not task_area or not task_area.event.is_user_manager_or_admin(user):
|
|
raise permissions.PermissionDenied("Keine Berechtigung, Schichten für dieses Event hinzuzufügen.")
|
|
serializer.save()
|
|
|
|
class ShiftSignupView(views.APIView):
|
|
permission_classes = [permissions.AllowAny]
|
|
|
|
def post(self, request, shift_id):
|
|
shift = get_object_or_404(Shift, pk=shift_id)
|
|
event = shift.task_area.event
|
|
user = request.user if request.user.is_authenticated else None
|
|
is_manager = event.is_user_manager_or_admin(user)
|
|
|
|
# Check capacity
|
|
if shift.signups.count() >= shift.max_participants:
|
|
return Response({'error': 'Diese Schicht ist bereits vollständig belegt.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
guest_name = request.data.get('guest_name', '').strip()
|
|
|
|
# Manager/Owner/Admin registering a guest manually -> Skill check bypassed!
|
|
if is_manager and guest_name:
|
|
signup = ShiftSignup.objects.create(
|
|
shift=shift,
|
|
guest_name=guest_name,
|
|
guest_session_key='added_by_manager'
|
|
)
|
|
return Response({
|
|
'message': f'Gast "{guest_name}" wurde von dir erfolgreich eingetragen (ohne Qualifikationsprüfung).',
|
|
'signup_id': signup.id
|
|
}, status=status.HTTP_201_CREATED)
|
|
|
|
# Authenticated User flow (self-signup)
|
|
if user and not guest_name:
|
|
if ShiftSignup.objects.filter(shift=shift, user=user).exists():
|
|
return Response({'error': 'Du bist für diese Schicht bereits eingetragen.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
required_skills = shift.required_skills.all()
|
|
if required_skills.exists():
|
|
user_skills = user.skills.all()
|
|
missing_skills = [s.name for s in required_skills if s not in user_skills]
|
|
if missing_skills:
|
|
missing_str = ", ".join(missing_skills)
|
|
return Response({
|
|
'error': f'Dir fehlen folgende erforderliche Fähigkeiten für diese Schicht: {missing_str}'
|
|
}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
signup = ShiftSignup.objects.create(shift=shift, user=user)
|
|
return Response({'message': 'Erfolgreich für Schicht eingetragen.', 'signup_id': signup.id}, status=status.HTTP_201_CREATED)
|
|
|
|
# Guest flow for visitors
|
|
else:
|
|
if not guest_name:
|
|
return Response({'error': 'Bitte gib einen Namen an.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
captcha_token = request.data.get('captcha_token', '').strip()
|
|
if not captcha_token and not is_manager:
|
|
return Response({'error': 'Bitte schließe zuerst den ALTCHA Spamschutz ab.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
if not is_manager and not verify_altcha_payload(captcha_token):
|
|
return Response({'error': 'ALTCHA Verifizierung ungültig. Bitte versuche es erneut.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# If NOT a manager, enforce required skills check for guest signups
|
|
if not is_manager and shift.required_skills.exists():
|
|
required_names = ", ".join([s.name for s in shift.required_skills.all()])
|
|
return Response({
|
|
'error': f'Gäste können keine Fähigkeiten haben. Diese Schicht erfordert: {required_names}. Bitte registriere ein Konto mit den passenden Fähigkeiten.'
|
|
}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
if not request.session.session_key:
|
|
request.session.create()
|
|
session_key = request.session.session_key
|
|
request.session['guest_name'] = guest_name
|
|
|
|
if ShiftSignup.objects.filter(shift=shift, guest_session_key=session_key).exists():
|
|
return Response({'error': 'Du bist mit deiner Gast-Sitzung bereits für diese Schicht eingetragen.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
signup = ShiftSignup.objects.create(
|
|
shift=shift,
|
|
guest_name=guest_name,
|
|
guest_session_key=session_key
|
|
)
|
|
return Response({
|
|
'message': f'Gast "{guest_name}" erfolgreich eingetragen!',
|
|
'signup_id': signup.id
|
|
}, status=status.HTTP_201_CREATED)
|
|
|
|
def delete(self, request, shift_id, signup_id=None):
|
|
shift = get_object_or_404(Shift, pk=shift_id)
|
|
|
|
if not signup_id:
|
|
signup_id = request.data.get('signup_id') or request.query_params.get('signup_id')
|
|
if not signup_id and request.body:
|
|
try:
|
|
body_data = json.loads(request.body.decode('utf-8'))
|
|
signup_id = body_data.get('signup_id')
|
|
except Exception:
|
|
pass
|
|
|
|
if request.user.is_authenticated:
|
|
user = request.user
|
|
is_manager = shift.task_area.event.is_user_manager_or_admin(user)
|
|
|
|
if signup_id and is_manager:
|
|
signup = ShiftSignup.objects.filter(shift=shift, pk=signup_id).first()
|
|
if signup:
|
|
name = signup.user.get_display_name() if signup.user else (signup.guest_name or "Gast")
|
|
signup.delete()
|
|
return Response({'message': f'Eintragung von {name} wurde von der Schicht entfernt.'})
|
|
return Response({'error': 'Eintragung nicht gefunden.'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
signup = ShiftSignup.objects.filter(shift=shift, user=user).first()
|
|
if signup:
|
|
signup.delete()
|
|
return Response({'message': 'Eintragung storniert.'})
|
|
|
|
return Response({'error': 'Keine passende eigene Eintragung gefunden.'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
else:
|
|
session_key = request.session.session_key
|
|
if not session_key:
|
|
return Response({'error': 'Keine Gast-Sitzung gefunden.'}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
signup = ShiftSignup.objects.filter(shift=shift, guest_session_key=session_key).first()
|
|
if signup:
|
|
signup.delete()
|
|
return Response({'message': 'Gast-Eintragung storniert.'})
|
|
return Response({'error': 'Keine passende Gast-Eintragung gefunden.'}, status=status.HTTP_404_NOT_FOUND)
|
|
|
|
class EventTemplateViewSet(viewsets.ModelViewSet):
|
|
queryset = EventTemplate.objects.all().order_by('-created_at')
|
|
serializer_class = EventTemplateSerializer
|
|
permission_classes = [permissions.IsAuthenticatedOrReadOnly]
|
|
|
|
def perform_create(self, serializer):
|
|
serializer.save(created_by=self.request.user if self.request.user.is_authenticated else None)
|
|
|
|
@action(detail=True, methods=['post'], permission_classes=[permissions.IsAuthenticated])
|
|
def instantiate(self, request, pk=None):
|
|
template = self.get_object()
|
|
title = request.data.get('title', f"Veranstaltung: {template.name}")
|
|
start_date = request.data.get('start_date')
|
|
end_date = request.data.get('end_date')
|
|
location = request.data.get('location', '')
|
|
|
|
event = Event.objects.create(
|
|
title=title,
|
|
description=template.description,
|
|
location=location,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
created_by=request.user
|
|
)
|
|
|
|
template_task_areas = template.template_data.get('task_areas', [])
|
|
for ta_data in template_task_areas:
|
|
ta = TaskArea.objects.create(
|
|
event=event,
|
|
name=ta_data.get('name', 'Bereich'),
|
|
description=ta_data.get('description', ''),
|
|
order=ta_data.get('order', 0)
|
|
)
|
|
|
|
for shift_data in ta_data.get('shifts', []):
|
|
shift = Shift.objects.create(
|
|
task_area=ta,
|
|
title=shift_data.get('title', 'Schicht'),
|
|
date=shift_data.get('date') or start_date,
|
|
start_time=shift_data.get('start_time', '18:00'),
|
|
end_time=shift_data.get('end_time', '22:00'),
|
|
max_participants=shift_data.get('max_participants', 1)
|
|
)
|
|
|
|
skill_ids = shift_data.get('required_skill_ids', [])
|
|
skill_names = shift_data.get('required_skill_names', [])
|
|
if skill_ids:
|
|
shift.required_skills.set(Skill.objects.filter(id__in=skill_ids))
|
|
elif skill_names:
|
|
shift.required_skills.set(Skill.objects.filter(name__in=skill_names))
|
|
|
|
return Response({
|
|
'message': f'Veranstaltung "{event.title}" wurde erfolgreich aus der Vorlage erstellt.',
|
|
'event_id': event.id
|
|
}, status=status.HTTP_201_CREATED)
|