# SPDX-FileCopyrightText: 2017-present Tobias Kunze
# SPDX-License-Identifier: AGPL-3.0-only WITH LicenseRef-Pretalx-AGPL-3.0-Terms
#
# This file contains Apache-2.0 licensed contributions copyrighted by the following contributors:
# SPDX-FileContributor: Florian Moesch
from copy import deepcopy
import rules
from django.conf import settings
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.template.loader import get_template
from django.utils.functional import cached_property
from django.utils.safestring import mark_safe
from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _
from django.utils.translation import override, pgettext_lazy
from django_scopes import ScopedManager
from i18nfield.fields import I18nCharField, I18nTextField
from pretalx.common.exceptions import SendMailException
from pretalx.common.models.mixins import PretalxModel
from pretalx.common.text.formatting import MODE_HTML, MODE_PLAIN, format_map
from pretalx.common.urls import EventUrls
from pretalx.mail.context import get_available_placeholders
from pretalx.mail.placeholders import SimpleFunctionalMailTextPlaceholder
from pretalx.mail.signals import queuedmail_post_send, queuedmail_pre_send
from pretalx.submission.rules import orga_can_change_submissions
class QueuedMailStates(models.TextChoices):
DRAFT = "draft", pgettext_lazy("email status", "Draft")
SENDING = "sending", pgettext_lazy("email status", "Sending")
SENT = "sent", pgettext_lazy("email status", "Sent")
def get_prefixed_subject(event, subject):
if not (prefix := event.mail_settings["subject_prefix"]):
return subject
if not (prefix.startswith("[") and prefix.endswith("]")):
prefix = f"[{prefix}]"
if subject.startswith(prefix):
return subject
return f"{prefix} {subject}"
class MailTemplateRoles(models.TextChoices):
NEW_SUBMISSION = "submission.new", _("Acknowledge proposal submission")
NEW_SUBMISSION_INTERNAL = (
"submission.new.internal",
_("New proposal (organiser notification)"),
)
SUBMISSION_ACCEPT = "submission.state.accepted", _("Proposal accepted")
SUBMISSION_REJECT = "submission.state.rejected", _("Proposal rejected")
NEW_SPEAKER_INVITE = (
"speaker.invite",
_("Add a speaker to a proposal (new account)"),
)
EXISTING_SPEAKER_INVITE = (
"speaker.invite.existing",
_("Add a speaker to a proposal (existing account)"),
)
QUESTION_REMINDER = "question.reminder", _("Custom fields reminder")
DRAFT_REMINDER = "draft.reminder", _("Draft proposal reminder")
NEW_SCHEDULE = "schedule.new", _("New schedule published")
PLACEHOLDER_KWARGS = {
MailTemplateRoles.NEW_SUBMISSION: ["submission", "event", "user", "slot"],
MailTemplateRoles.NEW_SUBMISSION_INTERNAL: ["submission", "event"],
MailTemplateRoles.SUBMISSION_ACCEPT: ["submission", "event", "user"],
MailTemplateRoles.SUBMISSION_REJECT: ["submission", "event", "user"],
MailTemplateRoles.NEW_SPEAKER_INVITE: ["submission", "event", "user"],
MailTemplateRoles.EXISTING_SPEAKER_INVITE: ["submission", "event", "user"],
MailTemplateRoles.QUESTION_REMINDER: ["event", "user"],
MailTemplateRoles.DRAFT_REMINDER: ["submission", "event", "user"],
MailTemplateRoles.NEW_SCHEDULE: ["event", "user"],
}
[docs]
class MailTemplate(PretalxModel):
"""MailTemplates can be used to create.
:class:`~pretalx.mail.models.QueuedMail` objects.
The process does not come with variable substitution except for
special cases, for now.
"""
log_prefix = "pretalx.mail_template"
event = models.ForeignKey(
to="event.Event", on_delete=models.PROTECT, related_name="mail_templates"
)
role = models.CharField(
choices=MailTemplateRoles.choices,
max_length=30,
null=True,
default=None,
editable=False,
)
subject = I18nCharField(
max_length=200, verbose_name=pgettext_lazy("email subject", "Subject")
)
text = I18nTextField(verbose_name=_("Text"))
reply_to = models.CharField(
max_length=200,
blank=True,
null=True,
verbose_name=_("Reply-To"),
help_text=_(
"Change the Reply-To address if you do not want to use the default organiser address"
),
)
bcc = models.CharField(
max_length=1000,
blank=True,
null=True,
verbose_name=_("BCC"),
help_text=_(
"Enter comma separated addresses. Will receive a blind copy of every email sent from this template. This may be a LOT!"
),
)
# Auto-created templates are created when mass emails are sent out. They are only used to re-create similar
# emails, and are never shown in a list of email templates or anywhere else.
is_auto_created = models.BooleanField(default=False)
objects = ScopedManager(event="event")
class Meta:
unique_together = (("event", "role"),)
rules_permissions = {
"list": orga_can_change_submissions,
"view": orga_can_change_submissions,
"create": orga_can_change_submissions,
"update": orga_can_change_submissions,
"delete": orga_can_change_submissions,
}
class urls(EventUrls):
base = edit = "{self.event.orga_urls.mail_templates}{self.pk}/"
delete = "{base}delete/"
def __str__(self):
"""Help with debugging."""
return f"MailTemplate(event={self.event.slug}, subject={self.subject})"
@property
def log_parent(self):
return self.event
[docs]
def to_mail(
self,
user,
event,
locale=None,
safe_extra_context=None,
context_kwargs=None,
skip_queue=False,
commit=True,
allow_empty_address=False,
submissions=None,
attachments=False,
):
"""Creates a :class:`~pretalx.mail.models.QueuedMail` object from a
MailTemplate. This is the canonical and safe way of constructing emails,
particularly emails including user-generated input (e.g. session titles,
user names, etc).
When the template is unsaved (``self.pk is None``), the resulting
``QueuedMail`` will have ``template=None``.
:param user: Either a :class:`~pretalx.person.models.user.User` or an
email address as a string.
:param submissions: A list of submissions to which this email belongs.
This is handled as an addition to any `submission` object present
in ``context_kwargs``.
:param event: The event to which this email belongs. May be ``None``.
:param locale: The locale will be set via the event and the recipient,
but can be overridden with this parameter.
:param safe_extra_context: Per-call overrides for the template
context. Every value must be a
:class:`~django.utils.safestring.SafeString`, a
:class:`~pretalx.common.text.formatting.EmailAlternativeString`,
or a numeric type (``int``, ``float``, ``Decimal``); see
:func:`~pretalx.mail.context.get_mail_context`.
:param context_kwargs: Passed to get_mail_context to retrieve the correct
context when rendering the template.
:param skip_queue: Send directly. If combined with commit=False, this will
remove any logging and traces.
:param commit: Set ``False`` to return an unsaved object.
"""
from pretalx.common.templatetags.rich_text import ( # noqa: PLC0415 -- slow import
render_mail_body,
)
from pretalx.mail.context import ( # noqa: PLC0415 -- avoid circular import
get_mail_context,
)
from pretalx.person.models import User # noqa: PLC0415 -- avoid circular import
if isinstance(user, str):
address = user
users = None
elif isinstance(user, User):
address = None
users = [user]
locale = locale or user.locale
elif not user and allow_empty_address:
address = None
users = None
else:
raise TypeError(
"First argument to to_mail must be a string or a User, not "
+ str(type(user))
)
if users and not commit:
address = ",".join(user.email for user in users)
users = None
event = event or getattr(self, "event", None)
with override(locale):
context_kwargs = context_kwargs or {}
context_kwargs["event"] = event
context = get_mail_context(
safe_extra_context=safe_extra_context, **context_kwargs
)
try:
subject = format_map(self.subject, context, mode=MODE_PLAIN)
text = format_map(self.text, context, mode=MODE_PLAIN)
text_html = render_mail_body(
format_map(self.text, context, mode=MODE_HTML)
)
except KeyError as e:
raise SendMailException(
f"Experienced KeyError when rendering email text: {e!s}"
) from e
if len(subject) > 200:
subject = subject[:198] + "…"
mail = QueuedMail(
event=event,
template=self if self.pk else None,
to=address,
reply_to=self.reply_to,
bcc=self.bcc,
subject=subject,
text=text,
text_html=text_html,
locale=locale,
attachments=attachments,
)
if commit:
mail.save()
submissions = set(submissions or [])
if submission := context_kwargs.get("submission"):
submissions.add(submission)
if submissions:
mail.submissions.set(submissions)
if users:
mail.to_users.set(users)
if skip_queue:
mail.send()
return mail
to_mail.alters_data = True
@cached_property
def valid_placeholders(self):
valid_placeholders = {}
if self.role == MailTemplateRoles.QUESTION_REMINDER:
valid_placeholders["questions"] = SimpleFunctionalMailTextPlaceholder(
"questions",
["user"],
None,
_("- First missing field\n- Second missing field"),
_(
"The list of custom fields that the user has not responded to, as bullet points"
),
)
valid_placeholders["url"] = SimpleFunctionalMailTextPlaceholder(
"url",
["event", "user"],
None,
"https://pretalx.example.com/democon/me/submissions/",
is_visible=False,
)
elif self.role == MailTemplateRoles.NEW_SPEAKER_INVITE:
valid_placeholders["invitation_link"] = SimpleFunctionalMailTextPlaceholder(
"invitation_link",
["event", "user"],
None,
"https://pretalx.example.com/democon/invitation/123abc/",
)
elif self.role == MailTemplateRoles.NEW_SUBMISSION_INTERNAL:
valid_placeholders["orga_url"] = SimpleFunctionalMailTextPlaceholder(
"orga_url",
["event", "submission"],
None,
"https://pretalx.example.com/orga/events/democon/submissions/124ABCD/",
)
kwargs = ["event", "user", "submission", "slot"]
if self.role:
kwargs = PLACEHOLDER_KWARGS[self.role]
valid_placeholders.update(
get_available_placeholders(event=self.event, kwargs=kwargs)
)
return valid_placeholders
@rules.predicate
def can_edit_mail(user, obj):
return getattr(obj, "state", None) == QueuedMailStates.DRAFT
class QueuedMailQuerySet(models.QuerySet):
def prefetch_users(self, event):
from pretalx.person.models.user import ( # noqa: PLC0415 -- avoid circular import
User,
)
return self.prefetch_related(
models.Prefetch("to_users", queryset=User.objects.with_speaker_code(event))
)
def with_computed_state(self):
return self.annotate(
computed_state=models.Case(
models.When(
state=QueuedMailStates.DRAFT,
error_data__isnull=False,
then=models.Value("failed"),
),
default=models.F("state"),
output_field=models.CharField(),
)
)
class QueuedMailManager(models.Manager.from_queryset(QueuedMailQuerySet)):
pass
[docs]
class QueuedMail(PretalxModel):
"""Emails in pretalx are rarely sent directly, hence the name QueuedMail.
This mechanism allows organisers to make sure they send out the right
content, and to include personal changes in emails.
:param sent: ``None`` if the mail has not been sent yet.
:param to_users: All known users to whom this email is addressed.
:param to: A comma-separated list of email addresses to whom this email
is addressed. Does not contain any email addresses known to belong
to users.
"""
log_prefix = "pretalx.mail"
event = models.ForeignKey(
to="event.Event",
on_delete=models.PROTECT,
related_name="queued_mails",
null=True,
blank=True,
)
template = models.ForeignKey(
to=MailTemplate,
related_name="mails",
on_delete=models.SET_NULL,
null=True,
blank=True,
)
to = models.CharField(
max_length=1000,
verbose_name=_("To"),
help_text=_("One email address or several addresses separated by commas."),
null=True,
blank=True,
)
to_users = models.ManyToManyField(to="person.User", related_name="mails")
reply_to = models.CharField(
max_length=1000,
null=True,
blank=True,
verbose_name=_("Reply-To"),
help_text=_("By default, the organiser address is used as Reply-To."),
)
cc = models.CharField(
max_length=1000,
null=True,
blank=True,
verbose_name=_("CC"),
help_text=_("One email address or several addresses separated by commas."),
)
bcc = models.CharField(
max_length=1000,
null=True,
blank=True,
verbose_name=_("BCC"),
help_text=_("One email address or several addresses separated by commas."),
)
subject = models.CharField(
max_length=200, verbose_name=pgettext_lazy("email subject", "Subject")
)
text = models.TextField(verbose_name=_("Text"))
# Set at send-time when the rendered HTML diverges from what
# re-rendering ``text`` would produce; cleared on organiser edit.
text_html = models.TextField(null=True, blank=True)
sent = models.DateTimeField(null=True, blank=True, verbose_name=_("Sent at"))
state = models.CharField(
max_length=10,
choices=QueuedMailStates.choices,
default=QueuedMailStates.DRAFT,
db_index=True,
)
error_data = models.JSONField(null=True, blank=True, default=None)
error_timestamp = models.DateTimeField(null=True, blank=True)
locale = models.CharField(max_length=32, null=True, blank=True)
attachments = models.JSONField(default=None, null=True, blank=True)
submissions = models.ManyToManyField(
to="submission.Submission", related_name="mails"
)
objects = ScopedManager(event="event", _manager_class=QueuedMailManager)
class Meta:
rules_permissions = {
"list": orga_can_change_submissions,
"view": orga_can_change_submissions,
"create": orga_can_change_submissions,
"update": can_edit_mail & orga_can_change_submissions,
"delete": orga_can_change_submissions,
"send": orga_can_change_submissions,
}
class urls(EventUrls):
base = edit = "{self.event.orga_urls.mail}{self.pk}/"
delete = "{base}delete"
send = "{base}send"
copy = "{base}copy"
@property
def has_error(self):
return self.state == QueuedMailStates.DRAFT and self.error_data is not None
def mark_sent(self):
self.state = QueuedMailStates.SENT
self.sent = now()
self.error_data = None
self.error_timestamp = None
self.save(update_fields=["state", "sent", "error_data", "error_timestamp"])
mark_sent.alters_data = True
def mark_failed(self, exception):
from smtplib import ( # noqa: PLC0415 -- lazy import; only needed for error handling
SMTPResponseException,
)
self.state = QueuedMailStates.DRAFT
error_data = {"error": str(exception), "type": type(exception).__name__}
if isinstance(exception, SMTPResponseException):
smtp_message = exception.smtp_error
if isinstance(smtp_message, bytes):
smtp_message = smtp_message.decode("utf-8", errors="replace")
error_data["smtp_code"] = exception.smtp_code
error_data["error"] = smtp_message
self.error_data = error_data
self.error_timestamp = now()
self.save(update_fields=["state", "error_data", "error_timestamp"])
mark_failed.alters_data = True
def __str__(self):
"""Help with debugging."""
return f"QueuedMail(to={self.to}, subject={self.subject}, state={self.state})"
@property
def html_body(self):
"""Return the sanitised HTML body of this mail — the same markup
that will be sent to the recipient, without the outer mail
wrapper template. Used both by :meth:`make_html` (which wraps
this in the full HTML email layout) and by the organiser outbox
/ mail-log previews, so the preview matches the delivered body
byte-for-byte."""
if self.text_html is not None:
# Already rendered at to_mail time (markdown + bleach via
# render_mail_body, with user-controlled substitutions
# escaped and wrapped in <span>/<div>). Stored in the DB as
# a plain string; re-mark safe for template rendering.
return mark_safe(self.text_html) # noqa: S308 -- rendered via render_mail_body at creation
# No placeholder-rendered HTML available — the mail was
# constructed directly (literal string) or the organiser
# edited the draft body after rendering (see
# ``MailDetailForm.save``). Fall back to the legacy
# markdown/bleach pipeline, which autolinks bare URLs in
# organiser-typed text as before. This path is safe only
# for text that is fully organiser- or system-controlled;
# user-content placeholders are already pre-sanitised in
# their plain variant so they cannot re-inject HTML or
# markdown links when this fallback runs.
from pretalx.common.templatetags.rich_text import ( # noqa: PLC0415 -- slow import
render_markdown_abslinks,
)
return render_markdown_abslinks(self.text)
def make_html(self):
event = getattr(self, "event", None)
sig = None
if event:
sig = event.mail_settings["signature"]
if sig.strip().startswith("-- "):
sig = sig.strip()[3:].strip()
html_context = {
"body": self.html_body,
"event": event,
"color": (event.primary_color if event else "")
or settings.DEFAULT_EVENT_PRIMARY_COLOR,
"locale": self.locale,
"rtl": self.locale in settings.LANGUAGES_BIDI,
"subject": self.subject,
"signature": sig,
}
return get_template("mail/mailwrapper.html").render(html_context)
def make_text(self):
event = getattr(self, "event", None)
if not event or not event.mail_settings["signature"]:
return self.text
sig = event.mail_settings["signature"]
if not sig.strip().startswith("-- "):
sig = f"-- \n{sig}"
return f"{self.text}\n{sig}"
@cached_property
def prefixed_subject(self):
event = getattr(self, "event", None)
if not event:
return self.subject
return get_prefixed_subject(event, self.subject)
[docs]
def send(self, requestor=None, orga: bool = True):
"""Sends an email.
:param requestor: The user issuing the command. Used for logging.
:type requestor: :class:`~pretalx.person.models.user.User`
:param orga: Was this email sent as by a privileged user?
"""
if self.state in (QueuedMailStates.SENT, QueuedMailStates.SENDING):
raise ValidationError(
_("This email has been sent already. It cannot be sent again.")
)
has_event = getattr(self, "event", None)
to = self.to.split(",") if self.to else []
with transaction.atomic():
if self.id:
to += [user.email for user in self.to_users.all()]
if has_event:
queuedmail_pre_send.send_robust(sender=self.event, mail=self)
if self.sent is not None or self.state == QueuedMailStates.SENT:
# The pre_send signal must have handled the sending already,
# so there is nothing left for us to do.
self.state = QueuedMailStates.SENT
self.save(update_fields=["state", "sent"])
return
text = self.make_text()
body_html = self.make_html()
task_kwargs = {
"to": to,
"subject": self.prefixed_subject,
"body": text,
"html": body_html,
"reply_to": (self.reply_to or "").split(","),
"event": self.event.pk if has_event else None,
"cc": (self.cc or "").split(","),
"bcc": (self.bcc or "").split(","),
"attachments": self.attachments,
}
if self.pk:
task_kwargs["queued_mail_id"] = self.pk
self.state = QueuedMailStates.SENDING
self.error_data = None
self.error_timestamp = None
self.save(update_fields=["state", "error_data", "error_timestamp"])
self.log_action(
"pretalx.mail.sent",
person=requestor,
orga=orga,
data={
"to_users": [
(user.pk, user.email) for user in self.to_users.all()
]
},
)
# Dispatch the async task outside the transaction so the worker
# sees committed state when it picks up the job.
from kombu.exceptions import ( # noqa: PLC0415 -- lazy import; only needed for error handling
OperationalError,
)
from pretalx.common.mail import ( # noqa: PLC0415 -- avoid circular import
mail_send_task,
)
if self.pk:
try:
mail_send_task.apply_async(kwargs=task_kwargs, ignore_result=True)
except (OSError, OperationalError) as exc:
self.mark_failed(exc)
return
queuedmail_post_send.send(sender=self.event, mail=self)
else:
# Non-persisted mail (commit=False fire-and-forget)
mail_send_task.apply_async(kwargs=task_kwargs, ignore_result=True)
self.sent = now()
self.state = QueuedMailStates.SENT
send.alters_data = True
[docs]
def copy_to_draft(self):
"""Copies an already sent email to a new object and adds it to the
outbox."""
new_mail = deepcopy(self)
new_mail.pk = None
new_mail.sent = None
new_mail.state = QueuedMailStates.DRAFT
new_mail.error_data = None
new_mail.error_timestamp = None
new_mail.save()
for user in self.to_users.all():
new_mail.to_users.add(user)
return new_mail
copy_to_draft.alters_data = True
