"""
Send a message by email to one or more recipients (by SMTP or sendmail).
"""
# Copyright (c) 2009-2024, UChicago Argonne, LLC. See LICENSE file.
import os
import sys
SMTP_TIMEOUT = 10
[docs]
class MailerError(Exception):
pass
[docs]
def sendMail_sendmail(
subject, message, recipients, sendmail_cfg, sender=None, logger=None
):
"""
Send an email message using sendmail (linux only).
:param str subject: short text for email subject
:param str message: full text of email body
:param [str] recipients: list of email addresses to receive the message
:param dict sendmail_cfg: such as returned from :mod:`PvMail.ini_config.Config.get`
:user: required - (*str*) username to for sendmail (or similar) program
:param str sender: "From" address, if *None* use *smtp_cfg['user']* value
:param obj logger: optional message logging method
EXAMPLE::
>>> import PvMail.ini_config
>>> sendmail_cfg = PvMail.ini_config.Config().get()
>>> recipients = ['joe@gmail.com', 'sally@example.org']
>>> subject = 'sendmail test message'
>>> message = PvMail.ini_config.__doc__
>>> sendMail_sendmail(subject, message, recipients, sendmail_cfg)
"""
if sys.platform not in ("linux2"):
raise MailerError(f"Cannot use this method on {sys.platform=!r}")
sender = sender or sendmail_cfg["user"]
if isinstance(recipients, str):
recipients = [
recipients,
]
def _sendmail_handler(email_program, from_addr, recipients, subject, message):
to_addr = str(" ".join(recipients))
mail_command = "%s -F %s -t %s" % (email_program, from_addr, to_addr)
mail_message = [
mail_command,
]
for who in recipients:
mail_message.append("To: " + who)
mail_message.append("Subject: " + subject)
mail_message.append(message)
cmd = """cat << +++ | %s\n+++""" % "\n".join(mail_message)
return mail_command, cmd
def _mail_handler(email_program, from_addr, recipients, subject, message):
# to_addr = str(" ".join(recipients))
# mail_command = "%s %s" % (email_program, to_addr)
# mail_message = "%s\n%s" % (subject, message)
# # TODO: needs to do THIS
# """
# cat /tmp/message.txt | mail jemian@anl.gov
# """
# # cmd = '''echo %s | %s''' % (mail_message, mail_command)
# # return mail_command, cmd # lgtm [py/unreachable-statement]
raise MailerError("code needs improvement here")
cmd = None
for email_program, handler in (
("/usr/lib/sendmail", _sendmail_handler),
("/usr/bin/sendmail", _sendmail_handler),
("/usr/bin/mail", _mail_handler),
):
if os.path.exists(email_program):
results = handler(email_program, sender, recipients, subject, message)
mail_command, cmd = results
break
if cmd is None:
raise MailerError("Cannot find mail transport agent for sendmail.")
if logger is not None:
logger("sending email to: " + str(recipients))
logger("email program: " + email_program)
logger("mail command: " + mail_command)
logger("email From: " + sender)
try:
if logger is not None:
logger("email command:\n" + cmd)
if os.path.exists(email_program):
os.popen(cmd) # send the message
else:
if logger is not None:
logger("email program (%s) does not exist" % email_program)
except Exception as reason:
# err_msg = traceback.format_exc()
final_msg = f"{cmd=!r}\ntraceback: {reason}"
logger(final_msg)
raise MailerError(final_msg)
if logger is not None:
logger("sendmail sent")
[docs]
def sendMail_SMTP(subject, message, recipients, smtp_cfg, sender=None, logger=None):
"""
send email message through SMTP server
:param str subject: short text for email subject
:param str message: full text of email body
:param [str] recipients: list of email addresses to receive the message
:param dict smtp_cfg: such as returned from :mod:`PvMail.ini_config.Config.get`
:server: required - (*str*) SMTP server
:user: required - (*str*) username to login to SMTP server
:port: optional - (*str*) SMTP port
:password: optional - (*str*) password for username
:connection_security: optional - (*str*) **STARTTLS** (the only choice, if specified)
:param str sender: "From" address, if *None* use *smtp_cfg['user']* value
EXAMPLE::
>>> import PvMail.ini_config
>>> smtp_cfg = PvMail.ini_config.Config().get()
>>> recipients = ['joe@gmail.com', 'sally@example.org']
>>> subject = 'SMTP test message'
>>> message = PvMail.ini_config.__doc__
>>> sendMail_SMTP(subject, message, recipients, smtp_cfg)
"""
import email
import smtplib
host = smtp_cfg.get("server", None)
if host is None:
raise MailerError("must define an SMTP host to be used")
port = smtp_cfg.get("port", None)
username = smtp_cfg.get("user", None)
if username is None:
raise MailerError("must define a username for the SMTP server")
if sender is None:
sender = username
password = smtp_cfg.get("password", None)
connection_security = smtp_cfg.get("connection_security", None)
if connection_security not in (None, "STARTTLS"):
msg = "connection_security must be: STARTTLS or not defined, found: "
raise MailerError(msg + connection_security)
msg = email.message.Message()
if isinstance(recipients, str):
msg["To"] = recipients
else:
for who in recipients:
msg["To"] = who
msg["From"] = sender
msg["Subject"] = subject
msg.set_payload(message)
if logger is not None:
logger("sending email to: " + str(recipients))
logger("SMTP server: " + host)
logger("SMTP user: " + username)
logger("email From: " + sender)
smtpserver = smtplib.SMTP(timeout=SMTP_TIMEOUT)
# smtpserver.set_debuglevel(1)
if port is None:
smtpserver.connect(host)
else:
smtpserver.connect(host, int(port))
if logger is not None:
logger("SMTP connected")
smtpserver.ehlo()
if smtp_cfg.get("connection_security", None) == "STARTTLS":
smtpserver.starttls()
smtpserver.ehlo()
if logger is not None:
logger("SMTP STARTTLS")
if password is not None:
smtpserver.login(username, password)
if logger is not None:
logger("SMTP authenticated")
smtpserver.sendmail(username, recipients, str(msg))
smtpserver.close()
if logger is not None:
logger("SMTP complete")
[docs]
def send_message(subject, message, recipients, config):
"""
send an email message
:param str subject: short text for email subject
:param str message: full text of email body
:param [str] recipients: list of email addresses to receive the message
:param dict config: such as returned from :mod:`PvMail.ini_config.Config`
"""
email_agent_dict = dict(sendmail=sendMail_sendmail, SMTP=sendMail_SMTP)
agent = email_agent_dict[config.mail_transfer_agent]
agent(subject, message, recipients, config.get())
[docs]
def main():
"""
User on-demand test of the mailer module and configuration.
"""
import argparse
from . import DOCS_URL
from . import __version__
from . import ini_config
doc = f"Test the email sender from PvMail {__version__}"
parser = argparse.ArgumentParser(description=doc)
msg = "email address(es), whitespace-separated if more than one"
parser.add_argument("recipient", action="store", nargs="+", help=msg, default="")
results = parser.parse_args()
print(doc)
cfg = ini_config.Config()
print("Using configuration from: " + cfg.ini_file)
print("Using user name: " + cfg.get()["user"])
recipients = results.recipient
print("Sending email(s) to: " + str(" ".join(recipients)))
print("mail transfer agent: " + cfg.mail_transfer_agent)
subject = "PvMail mailer test message: " + cfg.mail_transfer_agent
message = f"{doc}\nFor more help, see: {DOCS_URL}"
send_message(subject, message, recipients, cfg)
