#!/usr/bin/env python3 """Envoie un email adalan par Mailgun, images intégrées au message. Le HTML renvoie à ses images par cid: : le script joint chaque fichier en image intégrée (paramètre inline de Mailgun), remplit les variables {{nom}} pour chaque destinataire, retire les commentaires internes et ajoute une version texte. Sans --envoyer, rien ne part : le script vérifie l'email et écrit l'aperçu .eml du premier destinataire, à ouvrir dans Apple Mail ou Outlook. Exemples : python3 envoyer_email.py email.html --objet "Matinale IA" \\ --de "adalan " --repondre-a contact@adalan.fr \\ --destinataires invites.csv --var lien_inscription=https://www.adalan.fr/contact/ python3 envoyer_email.py email.html --objet "Matinale IA" --de "…" \\ --a moi@adalan.fr --var prenom=Camille --envoyer Liste des destinataires : CSV en UTF-8, séparé par des virgules ou des points-virgules, avec une colonne email et une colonne par variable. Envoi : variables d'environnement MAILGUN_API_KEY, MAILGUN_DOMAIN et MAILGUN_REGION (us par défaut, eu pour un domaine créé dans la région Europe). Python 3.9 ou plus, bibliothèque standard seulement. """ import argparse import base64 import csv import html import json import mimetypes import os import re import sys import urllib.error import urllib.request import uuid from email.message import EmailMessage from email.utils import formatdate from html.parser import HTMLParser from pathlib import Path IMAGES = Path(__file__).resolve().parent.parent / "assets" / "email" DESINSCRIPTION = "%unsubscribe_url%" # remplacé par Mailgun, suivi des désinscriptions activé RE_VAR = re.compile(r"\{\{\s*([A-Za-z_]\w*)\s*\}\}") RE_CID = re.compile(r"cid:([\w.\-]+)") RE_TROU = re.compile(r"\[[^\[\]<>\n]{1,80}\]") RE_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") # commentaires internes, toujours ouverts par « , ) restent RE_COMMENTAIRE = re.compile(r"\n?") class EchecMailgun(Exception): pass def erreur(message): sys.exit(f"erreur : {message}") class VersionTexte(HTMLParser): """Version texte du message : paragraphes, liens entre parenthèses.""" BLOCS = {"p", "h1", "h2", "h3", "tr", "table", "div", "li", "br"} IGNORES = {"head", "style", "title", "script"} def __init__(self): super().__init__(convert_charrefs=True) self.morceaux, self.ignore, self.cache, self.liens = [], 0, 0, [] def handle_starttag(self, tag, attrs): attrs = dict(attrs) if tag in self.IGNORES: self.ignore += 1 elif tag == "div" and (self.cache or "display:none" in (attrs.get("style") or "").replace(" ", "")): self.cache += 1 elif tag == "a": self.liens.append((attrs.get("href") or "", len(self.morceaux))) if tag in self.BLOCS: self.morceaux.append("\n") elif tag == "td": self.morceaux.append(" ") def handle_endtag(self, tag): if tag in self.IGNORES: self.ignore = max(0, self.ignore - 1) elif tag == "div" and self.cache: self.cache -= 1 elif tag == "a" and self.liens: lien, debut = self.liens.pop() texte = "".join(self.morceaux[debut:]).strip() if lien and not lien.startswith(("mailto:", "#")) and lien not in texte: self.morceaux.append(f" ({lien})") if tag in self.BLOCS: self.morceaux.append("\n") def handle_data(self, data): if not self.ignore and not self.cache: self.morceaux.append(re.sub(r"\s+", " ", data)) def texte(self): lignes = [ligne.strip() for ligne in "".join(self.morceaux).split("\n")] return re.sub(r"\n{3,}", "\n\n", "\n".join(lignes)).strip() + "\n" def version_texte(source): analyse = VersionTexte() analyse.feed(re.sub(r"|", "", source)) return analyse.texte() def lire_liste(chemin): if not chemin.is_file(): erreur(f"liste des destinataires introuvable : {chemin}") try: with open(chemin, newline="", encoding="utf-8-sig") as f: echantillon = f.read(4096) f.seek(0) try: dialecte = csv.Sniffer().sniff(echantillon, delimiters=",;\t") except csv.Error: dialecte = csv.excel lignes = list(csv.DictReader(f, dialect=dialecte)) except UnicodeDecodeError: erreur(f"{chemin} n'est pas en UTF-8 : l'enregistrer depuis Excel en « CSV UTF-8 ».") destinataires, vus, invalides = [], set(), [] for numero, ligne in enumerate(lignes, start=2): ligne = {(k or "").strip().lower(): (v or "").strip() for k, v in ligne.items()} if "email" not in ligne: erreur(f"{chemin} n'a pas de colonne email (colonnes lues : {', '.join(ligne) or 'aucune'}).") adresse = ligne["email"] if not adresse: continue if not RE_EMAIL.match(adresse): invalides.append(f"ligne {numero} : {adresse}") elif adresse.lower() not in vus: vus.add(adresse.lower()) destinataires.append(ligne) if invalides: erreur("adresses invalides dans la liste :\n " + "\n ".join(invalides)) if not destinataires: erreur(f"aucun destinataire dans {chemin}.") return destinataires def images(source, dossier): fichiers = [] for nom in sorted(set(RE_CID.findall(source))): chemin = dossier / nom if not chemin.is_file(): erreur(f"cid:{nom} renvoie à {chemin}, introuvable. Préciser le dossier des images avec --images.") type_mime = mimetypes.guess_type(nom)[0] if not type_mime or not type_mime.startswith("image/"): erreur(f"cid:{nom} n'est pas une image reconnue (png, jpg ou gif).") fichiers.append((nom, chemin.read_bytes(), type_mime)) return fichiers def remplir(modele, valeurs, echapper): def remplace(m): valeur = valeurs.get(m.group(1)) if valeur is None: return m.group(0) return html.escape(valeur) if echapper and valeur != DESINSCRIPTION else valeur return RE_VAR.sub(remplace, modele) def message_eml(expediteur, destinataire, objet, corps_html, corps_texte, fichiers, repondre_a): message = EmailMessage() message["From"], message["To"], message["Subject"] = expediteur, destinataire, objet if repondre_a: message["Reply-To"] = repondre_a message["Date"] = formatdate(localtime=True) message["X-Unsent"] = "1" # Outlook ouvre l'aperçu comme un brouillon message.set_content(corps_texte) message.add_alternative(corps_html, subtype="html") partie_html = message.get_payload()[1] for nom, donnees, type_mime in fichiers: principal, secondaire = type_mime.split("/") partie_html.add_related(donnees, maintype=principal, subtype=secondaire, cid=f"<{nom}>", filename=nom, disposition="inline") return message def envoyer_mailgun(cle, domaine, region, champs, fichiers): base = "https://api.eu.mailgun.net" if region == "eu" else "https://api.mailgun.net" limite = uuid.uuid4().hex corps = bytearray() for nom, valeur in champs: corps += f'--{limite}\r\nContent-Disposition: form-data; name="{nom}"\r\n\r\n'.encode() corps += valeur.encode() + b"\r\n" for nom, donnees, type_mime in fichiers: corps += (f'--{limite}\r\nContent-Disposition: form-data; name="inline"; filename="{nom}"\r\n' f"Content-Type: {type_mime}\r\n\r\n").encode() + donnees + b"\r\n" corps += f"--{limite}--\r\n".encode() requete = urllib.request.Request( f"{base}/v3/{domaine}/messages", data=bytes(corps), method="POST", headers={"Authorization": "Basic " + base64.b64encode(f"api:{cle}".encode()).decode(), "Content-Type": f"multipart/form-data; boundary={limite}"}) try: with urllib.request.urlopen(requete, timeout=60) as reponse: return json.load(reponse).get("id", "") except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace").strip() aide = {401: "La clé MAILGUN_API_KEY est refusée.", 403: "La clé n'a pas le droit d'envoyer depuis ce domaine.", 404: f"Domaine {domaine} inconnu dans la région {region} : vérifier MAILGUN_DOMAIN, ou passer à " f"MAILGUN_REGION={'us' if region == 'eu' else 'eu'} si le domaine a été créé dans l'autre région." }.get(e.code, "") raise EchecMailgun(f"Mailgun a répondu {e.code} : {detail} {aide}".strip()) except urllib.error.URLError as e: raise EchecMailgun(f"Mailgun injoignable : {e.reason}") def main(): p = argparse.ArgumentParser(description="Envoie un email adalan par Mailgun, images intégrées au message.") p.add_argument("html", type=Path, help="fichier HTML de l'email (copie remplie de assets/modeles/email.html)") p.add_argument("--objet", required=True, help="objet de l'email ; peut contenir des {{variables}}") p.add_argument("--de", required=True, help='expéditeur, par exemple "adalan "') qui = p.add_mutually_exclusive_group(required=True) qui.add_argument("--destinataires", type=Path, help="liste CSV : une colonne email et une colonne par variable") qui.add_argument("--a", action="append", metavar="EMAIL", help="destinataire, répétable, pour un essai") p.add_argument("--var", action="append", default=[], metavar="NOM=VALEUR", help="valeur commune à tous, quand la liste n'a pas la colonne ou la laisse vide") p.add_argument("--repondre-a", metavar="EMAIL", help="adresse qui reçoit les réponses (Reply-To)") p.add_argument("--tag", help="étiquette Mailgun, pour retrouver l'envoi dans les statistiques") p.add_argument("--images", type=Path, default=IMAGES, help="dossier des images cid: (par défaut assets/email)") p.add_argument("--apercu", type=Path, default=Path("apercu.eml"), help="aperçu .eml du premier destinataire") p.add_argument("--envoyer", action="store_true", help="envoie vraiment ; sans cette option, rien ne part") p.add_argument("--test", action="store_true", help="avec --envoyer : Mailgun accepte sans distribuer") p.add_argument("--forcer", action="store_true", help="envoie même s'il reste des [crochets] dans le texte") args = p.parse_args() if not args.html.is_file(): erreur(f"email introuvable : {args.html}") source = RE_COMMENTAIRE.sub("", args.html.read_text(encoding="utf-8")) visible = re.sub(r"|", "", source) fichiers = images(visible, args.images) globales = {"lien_desinscription": DESINSCRIPTION} for paire in args.var: nom, egal, valeur = paire.partition("=") if not egal or not RE_VAR.fullmatch("{{" + nom.strip() + "}}"): erreur(f"--var {paire} : écrire nom=valeur, le nom en lettres, chiffres et _.") globales[nom.strip()] = valeur destinataires = lire_liste(args.destinataires) if args.destinataires else [{"email": a} for a in args.a] for d in destinataires: if not RE_EMAIL.match(d["email"]): erreur(f"adresse invalide : {d['email']}") noms = set(RE_VAR.findall(visible)) | set(RE_VAR.findall(args.objet)) manquants, valeurs_par_destinataire = {}, [] for d in destinataires: valeurs = {**globales, **{k: v for k, v in d.items() if v}} for nom in noms - set(valeurs): manquants.setdefault(nom, []).append(d["email"]) valeurs_par_destinataire.append(valeurs) if manquants: lignes = [f"{{{{{nom}}}}} pour {len(qui)} destinataire(s), dont {qui[0]}" for nom, qui in sorted(manquants.items())] erreur("variables sans valeur (colonne de la liste ou --var nom=valeur) :\n " + "\n ".join(lignes)) def composer(valeurs): objet = remplir(args.objet, valeurs, echapper=False) corps = remplir(source, valeurs, echapper=True) corps = re.sub(r"[\s\S]*?", f"{html.escape(objet)}", corps, count=1) return objet, corps, version_texte(corps) objet, corps_html, corps_texte = composer(valeurs_par_destinataire[0]) apercu = message_eml(args.de, destinataires[0]["email"], objet, corps_html, corps_texte, fichiers, args.repondre_a) args.apercu.write_bytes(bytes(apercu)) taille = len(corps_html.encode()) + sum(len(f[1]) for f in fichiers) print(f"{len(destinataires)} destinataire(s), {len(fichiers)} image(s) intégrée(s) : " f"{', '.join(f[0] for f in fichiers) or 'aucune'} ; environ {taille // 1024} Ko par message.") print(f"aperçu du premier message : {args.apercu}") trous = sorted(set(RE_TROU.findall(visible + args.objet))) if trous: print("à compléter, encore entre crochets :\n " + "\n ".join(trous)) if args.envoyer and not args.forcer: erreur("envoi annulé : compléter les crochets ou ajouter --forcer.") if not args.envoyer: print("rien n'est parti. Relancer avec --envoyer, ou --envoyer --test pour un essai sans distribution.") return cle, domaine = os.environ.get("MAILGUN_API_KEY"), os.environ.get("MAILGUN_DOMAIN") region = os.environ.get("MAILGUN_REGION", "us").lower() if not cle or not domaine: erreur("définir MAILGUN_API_KEY et MAILGUN_DOMAIN (et MAILGUN_REGION=eu pour un domaine européen).") if region not in ("us", "eu"): erreur(f"MAILGUN_REGION vaut {region} : us ou eu attendu.") envoyes = 0 for d, valeurs in zip(destinataires, valeurs_par_destinataire): objet, corps_html, corps_texte = composer(valeurs) champs = [("from", args.de), ("to", d["email"]), ("subject", objet), ("html", corps_html), ("text", corps_texte)] if args.repondre_a: champs.append(("h:Reply-To", args.repondre_a)) if args.tag: champs.append(("o:tag", args.tag)) if args.test: champs.append(("o:testmode", "yes")) try: identifiant = envoyer_mailgun(cle, domaine, region, champs, fichiers) except EchecMailgun as e: reste = len(destinataires) - envoyes erreur(f"{e}\n{envoyes} envoyé(s), {reste} non envoyé(s) à partir de {d['email']}. " "Retirer de la liste les adresses déjà servies avant de relancer.") envoyes += 1 print(f"envoyé {envoyes}/{len(destinataires)} {d['email']} {identifiant}") print("essai : Mailgun a accepté les messages sans les distribuer." if args.test else "terminé.") if __name__ == "__main__": main()