93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
from rest_framework.views import APIView
|
|
from rest_framework.response import Response
|
|
from rest_framework import status, viewsets
|
|
from rest_framework.permissions import AllowAny, IsAdminUser
|
|
from rest_framework.authentication import SessionAuthentication
|
|
from rest_framework.decorators import api_view, permission_classes
|
|
from drf_spectacular.utils import extend_schema, extend_schema_view
|
|
|
|
from .models import ContactMe
|
|
from .serializer import ContactMeSerializer
|
|
from .tasks import send_contact_me_email_task, send_newly_added_items_to_store_email_task_last_week
|
|
from vontor_cz.turnstile import verify_turnstile
|
|
|
|
|
|
@extend_schema(tags=["advertisement", "public"])
|
|
class ContactMePublicView(APIView):
|
|
permission_classes = [AllowAny]
|
|
# Avoid CSRF for public endpoint by disabling SessionAuthentication
|
|
authentication_classes = []
|
|
|
|
def post(self, request):
|
|
email = request.data.get("email")
|
|
message = request.data.get("message")
|
|
honeypot = request.data.get("hp") # hidden honeypot field
|
|
|
|
# If honeypot is filled, pretend success without processing
|
|
if honeypot:
|
|
return Response({"status": "ok"}, status=status.HTTP_200_OK)
|
|
|
|
turnstile_token = request.data.get("turnstile_token", "")
|
|
remote_ip = request.META.get("HTTP_X_FORWARDED_FOR", request.META.get("REMOTE_ADDR", ""))
|
|
if not verify_turnstile(turnstile_token, remote_ip):
|
|
return Response({"detail": "Ověření CAPTCHA selhalo."}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
if not email or not message:
|
|
return Response({"detail": "Missing email or message."}, status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
# Save to DB
|
|
cm = ContactMe.objects.create(client_email=email, content=message)
|
|
|
|
# Send email via Celery task
|
|
try:
|
|
send_contact_me_email_task.delay(email, message)
|
|
except Exception:
|
|
# Fallback to direct call if Celery is not running in DEV
|
|
send_contact_me_email_task(email, message)
|
|
|
|
return Response({"id": cm.id, "status": "queued"}, status=status.HTTP_201_CREATED)
|
|
|
|
|
|
@extend_schema_view(
|
|
list=extend_schema(tags=["advertisement"], summary="List contact messages (admin)"),
|
|
retrieve=extend_schema(tags=["advertisement"], summary="Retrieve contact message (admin)"),
|
|
create=extend_schema(tags=["advertisement"], summary="Create contact message (admin)"),
|
|
partial_update=extend_schema(tags=["advertisement"], summary="Update contact message (admin)"),
|
|
update=extend_schema(tags=["advertisement"], summary="Replace contact message (admin)"),
|
|
destroy=extend_schema(tags=["advertisement"], summary="Delete contact message (admin)"),
|
|
)
|
|
class ContactMeAdminViewSet(viewsets.ModelViewSet):
|
|
queryset = ContactMe.objects.all().order_by("-sent_at")
|
|
serializer_class = ContactMeSerializer
|
|
permission_classes = [IsAdminUser]
|
|
|
|
|
|
@extend_schema(
|
|
tags=["advertisement"],
|
|
summary="Manually trigger weekly new items email",
|
|
description="Triggers the weekly email task that sends a summary of newly added products from the last week. Only accessible by admin users.",
|
|
methods=["POST"]
|
|
)
|
|
@api_view(['POST'])
|
|
@permission_classes([IsAdminUser])
|
|
def trigger_weekly_email(request):
|
|
"""
|
|
Manually trigger the weekly new items email task.
|
|
Only accessible by admin users.
|
|
"""
|
|
try:
|
|
# Trigger the task asynchronously
|
|
task = send_newly_added_items_to_store_email_task_last_week.delay()
|
|
|
|
return Response({
|
|
'success': True,
|
|
'message': 'Weekly email task triggered successfully',
|
|
'task_id': task.id
|
|
}, status=status.HTTP_200_OK)
|
|
|
|
except Exception as e:
|
|
return Response({
|
|
'success': False,
|
|
'message': f'Failed to trigger weekly email task: {str(e)}'
|
|
}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
|