sender.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341
  1. from email.message import EmailMessage
  2. from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Union
  3. import jinja2
  4. from redmail.email.attachment import Attachments
  5. from redmail.email.body import HTMLBody, TextBody
  6. from redmail.models import EmailAddress, Error
  7. from .envs import get_span, is_last_group_row
  8. import smtplib
  9. from pathlib import Path
  10. from platform import node
  11. from getpass import getuser
  12. import datetime
  13. import os
  14. if TYPE_CHECKING:
  15. # These are never imported but just for linters
  16. import pandas as pd
  17. from PIL.Image import Image
  18. import matplotlib.pyplot as plt
  19. class EmailSender:
  20. """Email sender
  21. Parameters
  22. ----------
  23. host : str
  24. SMTP host address.
  25. port : int
  26. Port to the SMTP server.
  27. user : str, callable
  28. User name to authenticate on the server.
  29. password : str, callable
  30. User password to authenticate on the server.
  31. Examples
  32. --------
  33. .. code-block:: python
  34. mymail = EmailSender(server="smtp.mymail.com", port=123)
  35. mymail.set_credentials(
  36. user=lambda: read_yaml("C:/config/email.yaml")["mymail"]["user"],
  37. password=lambda: read_yaml("C:/config/email.yaml")["mymail"]["password"]
  38. )
  39. mymail.send(
  40. subject="Important email",
  41. html="<h1>Important</h1><img src={{ nice_pic }}>",
  42. body_images={'nice_pic': 'path/to/pic.jpg'},
  43. )
  44. """
  45. default_html_theme = "modest.html"
  46. default_text_theme = "pandas.txt"
  47. templates_html = jinja2.Environment(loader=jinja2.FileSystemLoader(str(Path(__file__).parent / "templates/html")))
  48. templates_html_table = jinja2.Environment(loader=jinja2.FileSystemLoader(str(Path(__file__).parent / "templates/html/table")))
  49. templates_text = jinja2.Environment(loader=jinja2.FileSystemLoader(str(Path(__file__).parent / "templates/text")))
  50. templates_text_table = jinja2.Environment(loader=jinja2.FileSystemLoader(str(Path(__file__).parent / "templates/text/table")))
  51. # Set globals
  52. templates_html_table.globals["get_span"] = get_span
  53. templates_text_table.globals["get_span"] = get_span
  54. templates_html_table.globals["is_last_group_row"] = is_last_group_row
  55. templates_text_table.globals["is_last_group_row"] = is_last_group_row
  56. attachment_encoding = 'UTF-8'
  57. _cls_smtp_server = smtplib.SMTP
  58. def __init__(self, host:str, port:int, user_name:str=None, password:str=None):
  59. self.host = host
  60. self.port = port
  61. self.user_name = user_name
  62. self.password = password
  63. # Defaults
  64. self.sender = None
  65. self.receivers = None
  66. self.cc = None
  67. self.bcc = None
  68. self.subject = None
  69. self.text = None
  70. self.html = None
  71. self.html_template = None
  72. self.text_template = None
  73. def send(self, **kwargs) -> EmailMessage:
  74. """Send an email message.
  75. Parameters
  76. ----------
  77. subject : str
  78. Subject of the email.
  79. receivers : list, optional
  80. Receivers of the email.
  81. sender : str, optional
  82. Sender of the email.
  83. cc : list, optional
  84. Cc or Carbon Copy of the email.
  85. Extra recipients of the email.
  86. bcc : list, optional
  87. Blind Carbon Copy of the email.
  88. Extra recipients of the email that
  89. don't see who else got the email.
  90. html : str, optional
  91. HTML body of the email. May contain
  92. Jinja templated variables of the
  93. tables, images and other variables.
  94. text_body : str, optional
  95. Text body of the email.
  96. body_images : dict of bytes, path-likes and figures, optional
  97. HTML images to embed with the html. The key should be
  98. as Jinja variables in the html and the values represent
  99. images (path to an image, bytes of an image or image object).
  100. body_tables : Dict[str, pd.DataFrame], optional
  101. HTML tables to embed with the html. The key should be
  102. as Jinja variables in the html and the values are Pandas
  103. DataFrames.
  104. html_params : dict, optional
  105. Extra parameters passed to html_table as Jinja parameters.
  106. Examples
  107. --------
  108. >>> sender = EmailSender(host="myserver", port=1234)
  109. >>> sender.send(
  110. sender="me@gmail.com",
  111. receiver="you@gmail.com",
  112. subject="Some news",
  113. html='<h1>Hi,</h1> Nice to meet you. Look at this: <img src="{{ my_image }}">',
  114. body_images={"my_image": Path("C:/path/to/img.png")}
  115. )
  116. >>> sender.send(
  117. sender="me@gmail.com",
  118. receiver="you@gmail.com",
  119. subject="Some news",
  120. html='<h1>Hi {{ name }},</h1> Nice to meet you. Look at this table: <img src="{{ my_table }}">',
  121. body_images={"my_image": Path("C:/path/to/img.png")},
  122. html_params={"name": "Jack"},
  123. )
  124. Returns
  125. -------
  126. EmailMessage
  127. Email message.
  128. """
  129. msg = self.get_message(**kwargs)
  130. self.send_message(msg)
  131. return msg
  132. def get_message(self,
  133. subject:Optional[str]=None,
  134. sender:Optional[str]=None,
  135. receivers:Union[List[str], str, None]=None,
  136. cc:Union[List[str], str, None]=None,
  137. bcc:Union[List[str], str, None]=None,
  138. html:Optional[str]=None,
  139. text:Optional[str]=None,
  140. html_template:Optional[str]=None,
  141. text_template:Optional[str]=None,
  142. body_images:Optional[Dict[str, Union[str, bytes, 'plt.Figure', 'Image']]]=None,
  143. body_tables:Optional[Dict[str, 'pd.DataFrame']]=None,
  144. body_params:Optional[Dict[str, Any]]=None,
  145. attachments:Optional[Dict[str, Union[str, os.PathLike, 'pd.DataFrame', bytes]]]=None) -> EmailMessage:
  146. """Get the email message."""
  147. subject = subject or self.subject
  148. sender = self.get_sender(sender)
  149. receivers = self.get_receivers(receivers)
  150. cc = self.get_cc(cc)
  151. bcc = self.get_bcc(bcc)
  152. html = html or self.html
  153. text = text or self.text
  154. html_template = html_template or self.html_template
  155. text_template = text_template or self.text_template
  156. if subject is None:
  157. raise ValueError("Email must have a subject")
  158. msg = self._create_body(
  159. subject=subject,
  160. sender=sender,
  161. receivers=receivers,
  162. cc=cc,
  163. bcc=bcc,
  164. )
  165. if text is not None or text_template is not None:
  166. body = TextBody(
  167. template=self.get_text_template(text_template),
  168. table_template=self.get_text_table_template(),
  169. )
  170. body.attach(
  171. msg,
  172. text,
  173. tables=body_tables,
  174. jinja_params=self.get_text_params(extra=body_params, sender=sender),
  175. )
  176. if html is not None or html_template is not None:
  177. body = HTMLBody(
  178. template=self.get_html_template(html_template),
  179. table_template=self.get_html_table_template(),
  180. )
  181. body.attach(
  182. msg,
  183. html=html,
  184. images=body_images,
  185. tables=body_tables,
  186. jinja_params=self.get_html_params(extra=body_params, sender=sender)
  187. )
  188. if attachments:
  189. att = Attachments(attachments, encoding=self.attachment_encoding)
  190. att.attach(msg)
  191. return msg
  192. def get_receivers(self, receivers:Union[list, str, None]) -> Union[List[str], None]:
  193. """Get receivers of the email"""
  194. return receivers or self.receivers
  195. def get_cc(self, cc:Union[list, str, None]) -> Union[List[str], None]:
  196. """Get carbon copy (cc) of the email"""
  197. return cc or self.cc
  198. def get_bcc(self, bcc:Union[list, str, None]) -> Union[List[str], None]:
  199. """Get blind carbon copy (bcc) of the email"""
  200. return bcc or self.bcc
  201. def get_sender(self, sender:Union[str, None]) -> str:
  202. """Get sender of the email"""
  203. return sender or self.sender or self.user_name
  204. def _create_body(self, subject, sender, receivers=None, cc=None, bcc=None) -> EmailMessage:
  205. msg = EmailMessage()
  206. msg["from"] = sender
  207. msg["subject"] = subject
  208. # To whoom the email goes
  209. if receivers:
  210. msg["to"] = receivers
  211. if cc:
  212. msg['cc'] = cc
  213. if bcc:
  214. msg['bcc'] = bcc
  215. return msg
  216. def send_message(self, msg:EmailMessage):
  217. "Send the created message"
  218. user = self.user_name
  219. password = self.password
  220. server = self._cls_smtp_server(self.host, self.port)
  221. server.starttls()
  222. if user is not None or password is not None:
  223. server.login(user, password)
  224. server.send_message(msg)
  225. server.quit()
  226. def get_params(self, sender:str) -> Dict[str, Any]:
  227. "Get Jinja parametes passed to template"
  228. # TODO: Add receivers to params
  229. return {
  230. "node": node(),
  231. "user": getuser(),
  232. "now": datetime.datetime.now(),
  233. "sender": EmailAddress(sender),
  234. }
  235. def get_html_params(self, extra:Optional[dict]=None, **kwargs) -> Dict[str, Any]:
  236. params = self.get_params(**kwargs)
  237. params.update({
  238. "error": Error(content_type='html-inline')
  239. })
  240. if extra:
  241. params.update(extra)
  242. return params
  243. def get_text_params(self, extra:Optional[dict]=None, **kwargs) -> Dict[str, Any]:
  244. params = self.get_params(**kwargs)
  245. params.update({
  246. "error": Error(content_type='text')
  247. })
  248. if extra:
  249. params.update(extra)
  250. return params
  251. def get_html_table_template(self, layout:Optional[str]=None) -> jinja2.Template:
  252. layout = self.default_html_theme if layout is None else layout
  253. if layout is None:
  254. return None
  255. return self.templates_html_table.get_template(layout)
  256. def get_html_template(self, layout:Optional[str]=None) -> jinja2.Template:
  257. if layout is None:
  258. return None
  259. return self.templates_html.get_template(layout)
  260. def get_text_table_template(self, layout:Optional[str]=None) -> jinja2.Template:
  261. layout = self.default_text_theme if layout is None else layout
  262. if layout is None:
  263. return None
  264. return self.templates_text_table.get_template(layout)
  265. def get_text_template(self, layout:Optional[str]=None) -> jinja2.Template:
  266. if layout is None:
  267. return None
  268. return self.templates_text.get_template(layout)
  269. def set_template_paths(self,
  270. html:Union[str, os.PathLike, None]=None,
  271. text:Union[str, os.PathLike, None]=None,
  272. html_table:Union[str, os.PathLike, None]=None,
  273. text_table:Union[str, os.PathLike, None]=None):
  274. """Create Jinja envs for body templates using given paths
  275. This is a shortcut for manually setting them like:
  276. .. clode-block:: python
  277. sender.templates_html = jinja2.Environment(loader=jinja2.FileSystemLoader(...))
  278. sender.templates_text = jinja2.Environment(loader=jinja2.FileSystemLoader(...))
  279. ...
  280. """
  281. if html is not None:
  282. self.templates_html = jinja2.Environment(loader=jinja2.FileSystemLoader(html))
  283. if text is not None:
  284. self.templates_text = jinja2.Environment(loader=jinja2.FileSystemLoader(text))
  285. if html_table is not None:
  286. self.templates_html_table = jinja2.Environment(loader=jinja2.FileSystemLoader(html_table))
  287. if text_table is not None:
  288. self.templates_text_table = jinja2.Environment(loader=jinja2.FileSystemLoader(text_table))