-
-
Save guillain/46d558a4bb87ee480744ac0c5dd367f1 to your computer and use it in GitHub Desktop.
Email from Python with Jinja2
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python | |
| # -*- coding: utf-8 -*- | |
| # Fork from jriguera/email_notify.py | |
| # Url: https://gist.github.com/jriguera/f3191528b7676bd60af5 | |
| # Thanks to Jriguera!!! | |
| # Python 3 and compatibility with Python 2 | |
| from __future__ import unicode_literals, print_function | |
| import os, sys,logging, re, smtplib | |
| from email.mime.text import MIMEText | |
| from email.mime.multipart import MIMEMultipart | |
| from jinja2 import Environment, FileSystemLoader | |
| class EmailNotification(object): | |
| EMAIL_REGEX = re.compile('([\w\-\.\']+@(\w[\w\-]+\.)+[\w\-]+)') | |
| HTML_REGEX = re.compile('(^<!DOCTYPE html.*?>)') | |
| def __init__(self, smtphost=None, smtpport=None, fromuser=None, fromemail=None, login=None, | |
| password=None, templatedir='templates', logger=None, ssl=None): | |
| self.logger = logger | |
| if not logger: | |
| logging.basicConfig() | |
| self.logger = logging.getLogger(__name__) | |
| self.smtp = smtphost | |
| if self.smtp in (None,''): | |
| self.logger.error('No SMTP server provided') | |
| return | |
| self.port = smtpport | |
| if self.port in (None,''): | |
| self.port = 25 | |
| self.reply = fromemail | |
| if self.reply in (None,''): | |
| self.logger.error('No Email source provided') | |
| return | |
| self.mfrom = "%s <%s>" % (fromuser, fromemail) | |
| if self.mfrom in (None,''): | |
| self.logger.error('No User namme as sender provided') | |
| return | |
| self.smtplogin = login | |
| self.smtppass = password | |
| self.ssl = ssl | |
| if self.ssl in (None,''): | |
| self.ssl = 0 | |
| if os.path.isdir(templatedir): | |
| self.templatedir = templatedir | |
| else: | |
| self.templatedir = os.path.join(os.path.abspath(os.path.dirname(__file__)), templatedir) | |
| self.env = Environment(loader=FileSystemLoader(self.templatedir)) | |
| def _mailrender(self, data, template): | |
| template = template + ".tmpl" | |
| self.logger.debug("- Rendering template '%s'" % (template)) | |
| text = self.env.get_template(template) | |
| msg = text.render(data) | |
| return msg | |
| def _smtpconnect(self): | |
| try: | |
| self.logger.info('- SMTP connection') | |
| if self.ssl: | |
| smtp = smtplib.SMTP_SSL(self.smtp, self.port) | |
| else: | |
| smtp = smtplib.SMTP(self.smtp, self.port) | |
| except Exception as e: | |
| self.logger.error("Cannot connect with '%s:%s': %s" % (self.smtp, self.port, e)) | |
| raise | |
| smtp.ehlo() | |
| if self.smtplogin: | |
| try: | |
| self.logger.info('- Login ongoing') | |
| smtp.login(self.smtplogin, self.smtppass) | |
| except smtplib.SMTPException as e: | |
| self.logger.error("Cannot auth with '%s' on %s: %s" % (self.smtplogin, self.smtp, e)) | |
| raise | |
| return smtp | |
| def _smtpsend(self, smtp, recipient, subject, content): | |
| if self.HTML_REGEX.match(content) is None: | |
| self.logger.debug("- Sending text mail to '%s'" % (recipient)) | |
| msg = MIMEText(content) | |
| else: | |
| self.logger.debug("- Sending html mail to '%s'" % (recipient)) | |
| msg = MIMEMultipart('alternative') | |
| msg.attach(MIMEText(content, 'html', 'utf-8')) | |
| msg['From'] = self.mfrom | |
| msg['To'] = recipient | |
| msg['Reply-to'] = self.reply | |
| msg['Subject'] = subject | |
| smtp.sendmail(self.mfrom, [recipient], msg.as_string()) | |
| def send_email(self, recipient, subject, msg): | |
| smtp = self._smtpconnect() | |
| try: | |
| self.logger.info('- Sending email') | |
| self._smtpsend(smtp, recipient, subject, msg) | |
| except smtplib.SMTPException as e: | |
| self.logger.error("Cannot send mail to '%s': %s" % (recipient, e)) | |
| raise | |
| finally: | |
| smtp.quit() | |
| def send_bulk(self, msgs): | |
| smtp = self._smtpconnect() | |
| processed = 0 | |
| for (recipient, subject, msg) in msgs: | |
| try: | |
| self.logger.info('- Sendng bulk email: recipient:' + recipient + ', subject: ' + subject + ', msg:' + msg) | |
| self._smtpsend(smtp, recipient, subject, msg) | |
| except smtplib.SMTPException as e: | |
| self.logger.error("Cannot send mail to '%s': %s" % (recipient, e)) | |
| else: | |
| processed += 1 | |
| smtp.quit() | |
| return processed | |
| def mailout(self, email, name, subject, data, template): | |
| if email is None: | |
| error = "Email is empty!" | |
| self.logger.error(error) | |
| raise ValueError(error) | |
| elif self.EMAIL_REGEX.match(email) is None: | |
| error = "Invalid email address!" | |
| self.logger.error(error) | |
| raise ValueError(error) | |
| msg = self._mailrender(data, template) | |
| self.send_email(email, subject, msg) | |
| def mailbulk(self, email_data, template): | |
| elist = [] | |
| for edata in email_data: | |
| try: | |
| email = edata["email"] | |
| name = edata["name"] | |
| subject = edata["subject"] | |
| data = edata["data"] | |
| except Exception as e: | |
| continue | |
| if email is None: | |
| error = "Email is empty!" | |
| self.logger.error(error) | |
| continue | |
| elif self.EMAIL_REGEX.match(email) is None: | |
| error = "Invalid email address!" | |
| self.logger.error(error) | |
| continue | |
| msg = self._mailrender(data, template) | |
| elist.append((email, subject, msg)) | |
| return self.send_bulk(elist) | |
| def main(argv): | |
| e = EmailNotification( | |
| smtphost="smtp.gmail.com", | |
| smtpport=465, | |
| fromuser="toto", | |
| fromemail="[email protected]", | |
| login='[email protected]', | |
| password='app password', | |
| logger=logging.basicConfig(level=logging.DEBUG), | |
| ssl=1) | |
| elist = [{ | |
| "email": "[email protected], [email protected]", | |
| "name": "Tata", | |
| "subject": "My subject", | |
| "data" : { | |
| "dear": "Tata", | |
| "msg": "How are you?" | |
| } | |
| }] | |
| e.mailbulk(elist, "email-html_notify") | |
| if __name__ == "__main__": | |
| main(sys.argv) | |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| <!DOCTYPE html> | |
| <html> | |
| <head> | |
| <title>OpenStack notification</title> | |
| <style type="text/css"> | |
| span.bold {font-weight: bold;} | |
| table.noborder {border: 0px; padding: 8px;} | |
| th {text-align: left;} | |
| </style> | |
| </head> | |
| <body> | |
| <p> | |
| Dear {{ dear }}, | |
| </p> | |
| {% block content %} | |
| {% endblock %} | |
| <p> | |
| Regards, <br /> | |
| <br /> | |
| The Platform Engineering OpenStack support team | |
| </p> | |
| <p> | |
| -- <br /> | |
| This email has been sent to the email address associated with your OpenStack login. | |
| They only relate to events that may affect your OpenStack resources. If you | |
| receive multiple copies of this email, it may be due to the fact that you | |
| are a member of multiple projects. | |
| </p> | |
| </body> | |
| </html> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| {% extends "email-html_base.tmpl" %} | |
| {% block content %} | |
| <p> | |
| {{ msg }} | |
| </p> | |
| <p> | |
| <b>AFFECTED INSTANCES:</b> | |
| </p> | |
| <table class='noborder'> | |
| <tr> | |
| <th>UUID</th><th>IP Address</th><th>Host</th> | |
| </tr> | |
| {% for instance in instances -%} | |
| <tr> | |
| <td>{{ instance.id }}</td><td>{{ instance.accessIPv4 }}</td><td>{{ instance.name }}</td> | |
| </tr> | |
| {% endfor %} | |
| </table> | |
| {% endblock %} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment