This is my first article on getting started
A recent project in my work has a requirement to automatically send some messages to the specified mailbox, so how to implement the automatic mail sending function in Python? Here’s a brief introduction.
Python SMTP to send mail
Simple Mail Transfer Protocol (SMTP) is a Simple Mail Transfer Protocol. The Smplib library in Python encapsulates SMTP and supports SMTP. You can send plain text messages, HTML files, and messages with attachments.
First, we build a SendEmailManager class, which also follows the idea of object-oriented programming. The general structure is as follows:
class SendEmailManager(object) :
def __init__(self, **kwargs) :
# Initialize parameters.def _get_conf(self, key) :
# Get the configuration parameters.def _init_conf(self) :
Initialize configuration parameters.def _login_email(self) :
# Log in to the email server.def _make_mail_msg(self) :
# Build text mail objects.def do_send_mail(self) :
# Send email.Copy the codedef __init__(self, **kwargs)
Class initialization function, which can be used to set object properties and give initial values, can be arguments or fixed values, where the argument **kwargs is a variable dictionary of keyword arguments passed to the function argument, Here we are mainly on the SMTP server (here use QQ mailbox), send mail proxy mailbox, set in the mailbox client authorization password, variable parameters for some initialization. The specific code is as follows:
# SMTP server, here use QQ mailbox, other mailbox baidu
EMAIL_HOST = 'smtp.qq.com'
# Send email proxy mailbox
EMAIL_HOST_USER = '[email protected]'
The client authentication password is set in the mailbox. Note that this is not the mailbox password
EMAIL_HOST_PASSWORD = 'xxxxxxxxxxxxx'
Copy the codedef __init__(self, **kwargs) :
# Initialize parameters
self.email_host = EMAIL_HOST
self.email_host_user = EMAIL_HOST_USER
self.email_host_pass = EMAIL_HOST_PASSWORD
self.kwargs = kwargs
Copy the codedef _get_conf(self, key)
The key is used to read the value of the mutable parameter self.kwargs dictionary for other functions to use.
def _get_conf(self, key) :
# Get the configuration parameters
value = self.kwargs.get(key)
ifkey ! ="attach_file_list" and (value is None or value == ' ') :raise Exception("configuration parameter '%s' cannot be empty" % key)
return value
Copy the codedef _init_conf(self)
This function is responsible for initializing the configuration parameters returned by _get_conf so that subsequent functions can call them.
def _init_conf(self) :
Initialize configuration parameters
print(self._get_conf('receives'))
self.receives = self._get_conf('receives')
self.msg_subject = self._get_conf('msg_subject')
self.msg_content = self._get_conf('msg_content')
self.msg_from = self._get_conf('msg_from')
# attachment
self.attach_file_list = self._get_conf('attach_file_list')
Copy the codedef _login_email(self)
Log in to the email server. What I log in here is the QQ email server, and the port number is 465. For other email ports, please use Baidu, and the code is as follows:
def _login_email(self) :
# Log in to the email server
try:
server = smtplib.SMTP_SSL(self.email_host, port=465)
# set_debuglevel(1) prints out all the information about the SMTP server interaction
server.set_debuglevel(1)
# Log in to email
server.login(self.email_host_user, self.email_host_pass)
return server
except Exception as e:
print("mail login exception:", e)
raise e
Copy the codedef _make_mail_msg(self)
This function builds a mail instance object that handles the contents of a message. A normal email usually contains the sender information, email subject, email body, and some emails are attached with attachments. Please refer to the following code for specific Settings:
def _make_mail_msg(self) :
# Build mail objects
msg = MIMEMultipart()
msg.attach(MIMEText(self.msg_content, 'plain'.'utf-8'))
# Email Subject
msg['Subject'] = Header(self.msg_subject, "utf-8")
# Sender email information
msg['From'] = "<%s>" % self.msg_from
# msg['From'] = Header(self.msg_from + "<%s>" % self.email_host_user, "utf-8")
msg['To'] = ",".join(self.receives)
print("-", self.attach_file_list)
if self.attach_file_list:
for i, att in enumerate(self.attach_file_list):
# construct attachments to transfer files in the current directory
if not att:
break
att_i = MIMEText(open(att, 'rb').read(), 'base64'.'utf-8')
att_i["Content-Type"] = 'application/octet-stream'
Filename = filename = filename
att_i["Content-Disposition"] = 'attachment; filename="%s"' % att
msg.attach(att_i)
return msg
Copy the codedef do_send_mail(self)
To send an email, you string up the above several functions and go directly to the code:
def do_send_mail(self) :
# Send email
try:
self._init_conf()
server = self._login_email()
msg = self._make_mail_msg()
server.sendmail(self.email_host_user, self.receives, msg.as_string())
server.close()
print("Sent successfully!")
except Exception as e:
print("Mail sending exception", e)
Copy the codeConfigure parameters to test whether mails can be sent normally:
if __name__ == "__main__":
mail_conf = {
'msg_from': '****@foxmail.com'.# The email sender's address
'receives': ['****@qq.com',].This is a list, because there may be more than one recipient
'msg_subject': 'Python Mail test '.# The subject of the email
'msg_content': 'hello'.The content of the email
'attach_file_list': {"test.py": "test.py"."test.txt": "./test.txt"}, # is a list of attachment file paths
}
manager = SendEmailManager(**mail_conf)
manager.do_send_mail()
Copy the code
Automatic email sending
1. Code hardcoding
If = “if”; if = “if”; if = “if”;
def delayed_sending(manager) :
The # parameter manager is a SendEmailManager object
begin_time = int(time.time())
while True:
end_time = int(time.time())
if end_time - begin_time == 1*60*60:
threading.Thread(target=manager.do_send_mail()).start()
begin_time = end_time
print("Sent...")
Copy the codeAt the same time in order to prevent unknown reasons blocking the normal operation of the program, add multithreading to achieve asynchronous send mail, so that the program will be executed every hour to the specified mailbox to send an email, as long as the program does not stop the mail will not stop to send!!
Let’s say we need to send an email to a designated email address at 8 o ‘clock every day. Let’s do this:
def delayed_sending(manager) :
The # parameter manager is a SendEmailManager object
hour = 8
minute = 0
second = 0
while True:
now = datetime.datetime.now()
if now.hour == hour and now.minute == minute and now.second == second:
threading.Thread(target=manager.do_send_mail()).start()
print("Sent...")
Copy the codeAlthough the above two ways to achieve the function, but, the way to achieve is really low!
2. Realize the timing function through modules
In addition to hardcoding, we can do this using the Schedule module of the Python task timing runtime:
Run the schedule module using python tasks
def send_mail_by_schedule(manager) :
schedule.every(20).minutes.do(manager.do_send_mail) # Execute every 20 minutes
schedule.every().hour.do(manager.do_send_mail) # Execute every hour
schedule.every().day.at("In its").do(manager.do_send_mail) # Execute once a day at 10:00
schedule.every().monday.do(manager.do_send_mail) Execute once a week on Monday
schedule.every().friday.at("At").do(manager.do_send_mail) Execute once every Friday at 21:00
while True:
schedule.run_pending()
time.sleep(1)
Copy the codeIn a while True loop, schedule.run_pending() is used to keep the schedule running, and to query the list of tasks. In a task, you can set different times to run. This is similar to crontab.
However, if multiple scheduled tasks are executed, they are actually executed one by one from the top down. If the above task is complex and time-consuming, it will affect the running time of the following task. If you run five tasks every two minutes and each task takes one minute, that’s a total of five minutes, so that when the next two minutes come, the previous round of tasks will still be running and the next round of tasks will start again. The solution is simple: use multithreading/multi-process.
def run_threaded(manager) :
threading.Thread(target=manager.do_send_mail()).start()
Run the schedule module using python tasks
def send_mail_by_schedule(manager) :
schedule.every(1).minutes.do(run_threaded, manager) # Execute every 20 minutes
schedule.every().hour.do(run_threaded, manager) # Execute every hour
schedule.every().day.at("In its").do(run_threaded, manager) # Execute once a day at 10:00
schedule.every().monday.do(run_threaded, manager) Execute once a week on Monday
schedule.every().friday.at("At").do(run_threaded, manager) Execute once every Friday at 21:00
while True:
schedule.run_pending()
time.sleep(1)
Copy the codeIn this way, we can create a thread for each scheduled task and let the task run in the thread, thus achieving the effect of multiple tasks working in parallel. This way is not less time and effort than the first way, and does not require a lot of code at all, perfect!
conclusion
Tasks can also be executed automatically by using asynchronous task schedulers such as tasks like celery tasks that are not included in the tasks queue.
Once you’ve got the automated email out of the way, you can have the app send us the daily weather forecast, the daily news, chicken soup, etc. (or to your boyfriend or girlfriend) hahaha.
Finally, I would like to thank my girlfriend for her tolerance, understanding and support in work and life.
