Introduce notifications app and integrate emails

Add a new notifications app (models, serializers, views, admin, tasks, consumers, routing, urls, tests) with Notification.notify for in-app websocket pushes and optional email delivery. Centralize email sending in notifications.tasks.send_email_with_context and re-export it from account.tasks for compatibility. Update account and commerce and advertisement tasks to use Notification.notify/_notify_order and the new email helper; adjust order/notification tasks to consolidate logic. Wire notifications into ASGI routing, settings and URL conf. Misc: handle OSError when importing weasyprint, add tmp/ to .gitignore, add local .claude PowerShell checks, add social.blog skeleton, and remove legacy ews-component test files.
This commit is contained in:
David Bruno Vontor
2026-06-09 16:18:41 +02:00
parent 46bc131a56
commit 2592a69790
74 changed files with 666 additions and 13194 deletions

View File

@@ -1,121 +1,80 @@
from celery import shared_task
from celery.utils.log import get_task_logger
from django.core.mail import send_mail
from django.conf import settings
from django.utils.http import urlsafe_base64_encode
from django.utils.encoding import force_bytes
from django.template.loader import render_to_string
from .tokens import *
from .models import CustomUser
# Re-exported so existing imports from account.tasks still work
from notifications.tasks import send_email_with_context # noqa: F401
logger = get_task_logger(__name__)
def send_email_with_context(recipients, subject, template_path=None, context=None, message: str | None = None):
"""
Send emails rendering a single HTML template.
- `template_name` is a simple base name without extension, e.g. "email/test".
- Renders only HTML (".html"), no ".txt" support.
- Converts `user` in context to a plain dict to avoid passing models to templates.
"""
if isinstance(recipients, str):
recipients = [recipients]
html_message = None
if template_path:
ctx = dict(context or {})
# Render base layout and include the provided template as the main content.
# The included template receives the same context as the base.
html_message = render_to_string(
"email/components/base.html",
{"content_template": template_path, **ctx},
)
try:
send_mail(
subject=subject,
message=message or "",
from_email=None,
recipient_list=recipients,
fail_silently=False,
html_message=html_message,
)
if settings.EMAIL_BACKEND == 'django.core.mail.backends.console.EmailBackend' and message:
logger.debug(f"\nEMAIL OBSAH:\n{message}\nKONEC OBSAHU")
return True
except Exception as e:
logger.error(f"E-mail se neodeslal: {e}")
return False
#----------------------------------------------------------------------------------------------------
# This function sends an email to the user for email verification after registration.
@shared_task
def send_email_verification_task(user_id):
from notifications.models import Notification
try:
user = CustomUser.objects.get(pk=user_id)
except CustomUser.DoesNotExist:
logger.info(f"Task send_email_verification has failed. Invalid User ID was sent.")
return 0
logger.info("send_email_verification_task: invalid user_id %s", user_id)
return
uid = urlsafe_base64_encode(force_bytes(user.pk))
# {changed} generate and store a per-user token
token = user.generate_email_verification_token()
verify_url = f"{settings.FRONTEND_URL}/email-verification/?uidb64={uid}&token={token}"
context = {
"user": user,
"action_url": verify_url,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Ověřit email",
}
send_email_with_context(
recipients=user.email,
subject="Ověření emailu",
Notification.notify(
user=user,
title="Ověření emailu",
text="Prosím ověřte svou e-mailovou adresu kliknutím na odkaz v e-mailu.",
action_url=verify_url,
template_path="email/email_verification.html",
context=context,
email_context={
"action_url": verify_url,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Ověřit email",
},
)
@shared_task
def send_email_test_task(email):
context = {
"action_url": settings.FRONTEND_URL,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Otevřít aplikaci",
}
send_email_with_context(
recipients=email,
subject="Testovací email",
template_path="email/test.html",
context=context,
context={
"action_url": settings.FRONTEND_URL,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Otevřít aplikaci",
},
)
@shared_task
def send_password_reset_email_task(user_id):
from notifications.models import Notification
try:
user = CustomUser.objects.get(pk=user_id)
except CustomUser.DoesNotExist:
logger.info(f"Task send_password_reset_email has failed. Invalid User ID was sent.")
return 0
logger.info("send_password_reset_email_task: invalid user_id %s", user_id)
return
uid = urlsafe_base64_encode(force_bytes(user.pk))
token = password_reset_token.make_token(user)
reset_url = f"{settings.FRONTEND_URL}/reset-password/{uid}/{token}"
context = {
"user": user,
"action_url": reset_url,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Obnovit heslo",
}
send_email_with_context(
recipients=user.email,
subject="Obnova hesla",
Notification.notify(
user=user,
title="Obnova hesla",
text="Byl vyžádán reset vašeho hesla. Klikněte na odkaz pro nastavení nového hesla.",
action_url=reset_url,
template_path="email/password_reset.html",
context=context,
)
email_context={
"action_url": reset_url,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Obnovit heslo",
},
)

View File

@@ -237,7 +237,9 @@ class UserView(viewsets.ModelViewSet):
def get_serializer_class(self):
user = getattr(self.request, 'user', None)
is_admin = user and (getattr(user, 'role', None) == 'admin' or getattr(user, 'is_superuser', False))
if self.action in ['retrieve', 'list'] and not is_admin:
return PublicUserSerializer
return CustomUserSerializer
@@ -277,11 +279,15 @@ class ChangePasswordView(APIView):
def post(self, request):
serializer = ChangePasswordSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = request.user
if not user.check_password(serializer.validated_data['current_password']):
return Response({"current_password": "Nesprávné heslo."}, status=status.HTTP_400_BAD_REQUEST)
user.set_password(serializer.validated_data['new_password'])
user.save()
return Response({"detail": "Heslo bylo úspěšně změněno."})