333 lines
12 KiB
Python
333 lines
12 KiB
Python
"""
|
|
markdown mail template parsing
|
|
"""
|
|
|
|
import logging
|
|
import warnings
|
|
from typing import Any
|
|
|
|
import frontmatter
|
|
import markdown
|
|
from django.apps import apps
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.mail import EmailMultiAlternatives
|
|
from django.template import Context, Template
|
|
from django.template.base import Variable, VariableNode
|
|
from django.template.loader import get_template, render_to_string
|
|
|
|
from .validation import (
|
|
CourrielError,
|
|
InvalidAddressError,
|
|
InvalidFrontMatter,
|
|
InvalidGabarit,
|
|
InvalidRecipientsError,
|
|
InvalidSenderError,
|
|
MarkdownEmailError,
|
|
UnresolvedVariableError,
|
|
ValidAddress,
|
|
ValidAddressList,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class Courriel(EmailMultiAlternatives):
|
|
"""
|
|
django EmailMultiAlternatives with a french touch.
|
|
shall not accept invalid addresses or from_email.
|
|
And no empty subjects nor body, we're not barbarians.
|
|
"""
|
|
|
|
def __init__(self, *args, **kw):
|
|
super().__setattr__('_from_email', None)
|
|
self._to = ValidAddressList()
|
|
self._cc = ValidAddressList()
|
|
self._bcc = ValidAddressList()
|
|
self._reply_to = ValidAddressList()
|
|
super().__init__(*args, **kw)
|
|
|
|
@property
|
|
def from_email(self):
|
|
return self._from_email
|
|
|
|
@from_email.setter
|
|
def from_email(self, value):
|
|
self._from_email = ValidAddress.factory(value) if value else None
|
|
|
|
|
|
class MarkdownTemplate:
|
|
"""
|
|
Modèle de mail Markdown avec Front Matter YAML.
|
|
S'initialise avec le nom d'un gabarit sans contexte en validant sa structure.
|
|
"""
|
|
|
|
def __init__(self, template_name: str):
|
|
# find template file
|
|
# parse markdown, separate metadata from body
|
|
# validate minimal requirements
|
|
# validate and import hard coded addresses
|
|
# resolve, validate and import group addresses
|
|
self.template_name = template_name
|
|
self.raw_content = self.load_raw_template(template_name)
|
|
self.metadata, self.md_body = self.parse_template(self.raw_content)
|
|
self.validate()
|
|
|
|
@classmethod
|
|
def load_raw_template(cls, template_name: str) -> str:
|
|
"""Charge le code source brut du template sans l'évaluer."""
|
|
template_obj = get_template(template_name)
|
|
raw_content = getattr(getattr(template_obj, "template", template_obj), "source", "")
|
|
if not raw_content and hasattr(template_obj, "render"):
|
|
raw_content = template_obj.render({})
|
|
if not raw_content:
|
|
raise MarkdownEmailError(f"Le gabarit '{template_name}' est introuvable ou vide.")
|
|
return raw_content
|
|
|
|
@classmethod
|
|
def parse_template(cls, raw_content: str) -> tuple[dict[str, Any], str]:
|
|
"""Extrait le Front Matter YAML et le corps Markdown."""
|
|
try:
|
|
post = frontmatter.loads(raw_content)
|
|
return post.metadata or {}, post.content or ""
|
|
except Exception as err:
|
|
raise InvalidFrontMatter(f"Failed to parse Front Matter YAML : {err}") from err
|
|
|
|
def validate(self) -> None:
|
|
"""Vérifie la validité structurelle intrinsèque du modèle (métadonnées existantes)."""
|
|
if not self.metadata:
|
|
raise InvalidFrontMatter(
|
|
f"Le gabarit '{self.template_name}' ne contient pas de Front Matter YAML valide (délimité par ---)."
|
|
)
|
|
elif 'subject' not in self.metadata:
|
|
raise InvalidFrontMatter(f"Missing a 'subject' key in {self.template_name} front matter")
|
|
|
|
@classmethod
|
|
def extract_template_variable_names(cls, template_obj: Template) -> set[str]:
|
|
"""Extrait les noms de variables de premier niveau via l'API native des nœuds de template Django."""
|
|
variables = set()
|
|
for node in template_obj.nodelist.get_nodes_by_type(VariableNode):
|
|
var_obj = node.filter_expression.var
|
|
if isinstance(var_obj, Variable) and var_obj.literal is None and var_obj.var:
|
|
root_var = str(var_obj.var).split(".")[0]
|
|
variables.add(root_var)
|
|
return variables
|
|
|
|
@classmethod
|
|
def render_template_scalar(
|
|
cls,
|
|
template_str: str,
|
|
context: dict[str, Any],
|
|
field_name: str,
|
|
required: bool = False,
|
|
) -> str:
|
|
"""Rend une chaîne scalaire avec le contexte en gérant les variables manquantes."""
|
|
if not isinstance(template_str, str):
|
|
return template_str
|
|
|
|
template = Template(template_str)
|
|
vars_used = cls.extract_template_variable_names(template)
|
|
unresolved = [v for v in vars_used if v not in context or context[v] is None]
|
|
|
|
if unresolved:
|
|
if required:
|
|
raise UnresolvedVariableError(
|
|
f"Variable requise manquante ou non résolue pour '{field_name}': {', '.join(unresolved)}"
|
|
)
|
|
else:
|
|
warnings.warn(
|
|
f"Variable optionnelle non résolue pour '{field_name}': {', '.join(unresolved)}",
|
|
UserWarning,
|
|
stacklevel=3,
|
|
)
|
|
|
|
result = template.render(Context(context)).strip()
|
|
|
|
if required and not result:
|
|
raise UnresolvedVariableError(f"Le champ requis '{field_name}' ne peut être vide.")
|
|
|
|
return result
|
|
|
|
@classmethod
|
|
def resolve_group_emails(cls, group_name: str, required: bool = False) -> list[str]:
|
|
"""Résout les adresses e-mail des utilisateurs actifs appartenant à un groupe Django auth."""
|
|
Group = apps.get_model("auth", "Group")
|
|
User = get_user_model()
|
|
|
|
try:
|
|
group = Group.objects.get(name=group_name)
|
|
except Group.DoesNotExist:
|
|
if required:
|
|
raise UnresolvedVariableError(f"Le groupe Django auth '{group_name}' n'existe pas.")
|
|
warnings.warn(
|
|
f"Le groupe Django auth optionnel '{group_name}' n'existe pas.",
|
|
UserWarning,
|
|
stacklevel=3,
|
|
)
|
|
return []
|
|
|
|
emails = list(
|
|
User.objects.filter(groups=group, is_active=True)
|
|
.exclude(email="")
|
|
.exclude(email__isnull=True)
|
|
.values_list("email", flat=True)
|
|
)
|
|
|
|
if required and not emails:
|
|
raise UnresolvedVariableError(
|
|
f"Le groupe requis '{group_name}' ne contient aucun utilisateur actif avec une adresse e-mail."
|
|
)
|
|
|
|
return emails
|
|
|
|
@classmethod
|
|
def normalize_recipient_list(
|
|
cls,
|
|
raw_val: str | list[str] | None,
|
|
context: dict[str, Any],
|
|
field_name: str,
|
|
required: bool = False,
|
|
) -> list[str]:
|
|
"""Normalise un champ de destinataires (chaîne ou liste) en liste d'adresses résolues."""
|
|
if not raw_val:
|
|
if required:
|
|
raise UnresolvedVariableError(f"Le champ '{field_name}' est obligatoire et vide.")
|
|
return []
|
|
|
|
items = [raw_val] if isinstance(raw_val, str) else list(raw_val)
|
|
resolved_list = []
|
|
for item in items:
|
|
rendered = cls.render_template_scalar(item, context, field_name=field_name, required=required)
|
|
if rendered:
|
|
resolved_list.append(rendered)
|
|
|
|
if required and not resolved_list:
|
|
raise UnresolvedVariableError(f"Aucune adresse valide résolue pour '{field_name}'.")
|
|
|
|
return resolved_list
|
|
|
|
@classmethod
|
|
def resolve_groups_recipients(
|
|
cls,
|
|
raw_groups: str | list[str] | None,
|
|
field_name: str,
|
|
required: bool = False,
|
|
) -> list[str]:
|
|
"""Résout une liste de groupes d'utilisateurs en adresses e-mail uniques."""
|
|
if not raw_groups:
|
|
return []
|
|
|
|
groups = [raw_groups] if isinstance(raw_groups, str) else list(raw_groups)
|
|
resolved_emails = []
|
|
for grp in groups:
|
|
emails = cls.resolve_group_emails(grp, required=required)
|
|
resolved_emails.extend(emails)
|
|
|
|
return list(dict.fromkeys(resolved_emails))
|
|
|
|
def build_message(
|
|
self,
|
|
context: dict[str, Any] | None = None,
|
|
subject: str | None = None,
|
|
body: str | None = None,
|
|
from_email: str | None = None,
|
|
to: list[str] | None = None,
|
|
bcc: list[str] | None = None,
|
|
connection: Any = None,
|
|
attachments: Any = None,
|
|
headers: dict[str, Any] | None = None,
|
|
cc: list[str] | None = None,
|
|
reply_to: list[str] | None = None,
|
|
format: str | None = None,
|
|
) -> Courriel:
|
|
"""Génère et retourne une instance de Courriel prête à l'envoi pour le contexte donné."""
|
|
ctx = context.copy() if context else {}
|
|
|
|
# 1. Sujet
|
|
rendered_subject = subject or self.render_template_scalar(
|
|
self.metadata.get("subject", ""), ctx, field_name="subject", required=True
|
|
)
|
|
|
|
# 2. Destinataires directs et groupes
|
|
direct_to = to or self.normalize_recipient_list(
|
|
self.metadata.get("to"), ctx, field_name="to", required=False
|
|
)
|
|
group_to = self.resolve_groups_recipients(
|
|
self.metadata.get("to_groups"),
|
|
field_name="to_groups",
|
|
required=bool(not direct_to),
|
|
)
|
|
final_to = list(dict.fromkeys(direct_to + group_to))
|
|
if not final_to:
|
|
raise UnresolvedVariableError("Aucun destinataire 'to' spécifié ou résolu.")
|
|
|
|
direct_cc = cc or self.normalize_recipient_list(self.metadata.get("cc"), ctx, field_name="cc", required=False)
|
|
group_cc = self.resolve_groups_recipients(self.metadata.get("cc_groups"), field_name="cc_groups", required=False)
|
|
final_cc = list(dict.fromkeys(direct_cc + group_cc))
|
|
|
|
direct_bcc = bcc or self.normalize_recipient_list(self.metadata.get("bcc"), ctx, field_name="bcc", required=False)
|
|
group_bcc = self.resolve_groups_recipients(self.metadata.get("bcc_groups"), field_name="bcc_groups", required=False)
|
|
final_bcc = list(dict.fromkeys(direct_bcc + group_bcc))
|
|
|
|
final_reply_to = reply_to or self.normalize_recipient_list(
|
|
self.metadata.get("reply_to"), ctx, field_name="reply_to", required=False
|
|
)
|
|
|
|
final_from_email = from_email or self.render_template_scalar(
|
|
self.metadata.get("from_email", ""), ctx, field_name="from_email", required=False
|
|
) or None
|
|
|
|
# 3. Format de message (both, html, text)
|
|
mail_format = format or self.metadata.get("format", "both")
|
|
|
|
# 4. Évaluation du corps Markdown avec le contexte
|
|
rendered_markdown_body = body or Template(self.md_body).render(Context(ctx)).strip()
|
|
|
|
# 5. Rendu HTML Markdown
|
|
extensions = getattr(settings, "MARKDOWN_EMAIL_EXTENSIONS", ["extra", "nl2br"])
|
|
html_body = markdown.markdown(rendered_markdown_body, extensions=extensions)
|
|
|
|
# 6. Layout HTML optionnel
|
|
layout_template = self.metadata.get(
|
|
"layout",
|
|
getattr(settings, "MARKDOWN_EMAIL_DEFAULT_LAYOUT", "courriel/base.dj.html"),
|
|
)
|
|
if layout_template and layout_template != "none":
|
|
html_body = render_to_string(layout_template, {"content_html": html_body, **ctx})
|
|
|
|
# 7. Inlining CSS avec premailer si disponible
|
|
try:
|
|
from premailer import transform
|
|
html_body = transform(html_body)
|
|
except ImportError:
|
|
pass
|
|
|
|
# 8. Instanciation de Courriel
|
|
if mail_format == "text":
|
|
main_body = rendered_markdown_body
|
|
alternatives = []
|
|
elif mail_format == "html":
|
|
main_body = html_body
|
|
alternatives = []
|
|
else: # "both"
|
|
main_body = rendered_markdown_body
|
|
alternatives = [(html_body, "text/html")]
|
|
|
|
message = Courriel(
|
|
subject=rendered_subject,
|
|
body=main_body,
|
|
from_email=final_from_email,
|
|
to=final_to,
|
|
bcc=final_bcc,
|
|
connection=connection,
|
|
attachments=attachments,
|
|
headers=headers,
|
|
cc=final_cc,
|
|
reply_to=final_reply_to if final_reply_to else None,
|
|
)
|
|
|
|
for content, mimetype in alternatives:
|
|
message.attach_alternative(content, mimetype)
|
|
|
|
return message
|