khushalps
eef89c9ebe
git-subtree-dir: utils_v2 git-subtree-split: a6614afbe332e89b495f068705f04f085f931adf
269 lines
9.5 KiB
Python
269 lines
9.5 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Tuesday, 16th Jul, 2024
|
|
|
|
OBJECTIVE:
|
|
|
|
To be able to send out mails from code.
|
|
|
|
REFERENCES:
|
|
|
|
N/A
|
|
|
|
DOWNLOADS:
|
|
|
|
N/A
|
|
|
|
"""
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** IMPORT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# To make sibling directories accessible for imports:
|
|
import sys
|
|
sys.path.append(".")
|
|
sys.path.append("..")
|
|
|
|
# For working with mails:
|
|
import aiosmtplib
|
|
from email.mime.multipart import MIMEMultipart
|
|
from email.mime.image import MIMEImage
|
|
from email.mime.text import MIMEText
|
|
from email.mime.base import MIMEBase
|
|
from email import encoders
|
|
|
|
# My utils:
|
|
from utils import rate_limit_utils
|
|
|
|
# Common:
|
|
from shared.statuses import StatusCodes
|
|
|
|
# For random strings:
|
|
import string
|
|
import random
|
|
|
|
# For system-level activities:
|
|
import os
|
|
|
|
# For working with files in RAM:
|
|
import io
|
|
|
|
# For debugging:
|
|
from icecream import IceCreamDebugger
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MACROS / ONE-TIME INIT ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** VARIABLES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
# --- Nothing Yet
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** CLASSES ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
class MailMessage:
|
|
|
|
def __init__(self, to_email, subject):
|
|
|
|
"""
|
|
Create an instance of the message that you would like to send.
|
|
:param to_email: The EMail ID of th recipient.
|
|
:param subject: The subject of the mail.
|
|
"""
|
|
|
|
self.message = MIMEMultipart()
|
|
self.message["To"] = to_email
|
|
self.message["Subject"] = subject
|
|
|
|
def add_text(self, text):
|
|
|
|
"""
|
|
Add plain-text to the mail body.
|
|
:param text: The text to add to the mail body.
|
|
:return: None.
|
|
"""
|
|
|
|
self.message.attach(MIMEText(text, "plain"))
|
|
|
|
def add_html(self, html_text):
|
|
|
|
"""
|
|
Add HTML text to the mail body.
|
|
:param html_text: The HTML text to add to the mail body.
|
|
:return: None.
|
|
"""
|
|
|
|
self.message.attach(MIMEText(html_text, "html"))
|
|
|
|
def add_inline_image(self, image_file, content_id = None):
|
|
|
|
"""
|
|
Add an inline image to the body of the mail.
|
|
NOTE: This is NOT the same as sending an image as an attachment.
|
|
:param image_file: The image data to attach to the mail body.
|
|
:param content_id: Inline images are inserted via HTML bocks. This field identifies the image resource. If not
|
|
specified, I will generate a random string. You may write a custom value here if you know what you are
|
|
doing. For most use cases, please ignore this field.
|
|
:return: None.
|
|
"""
|
|
|
|
# Read the image as bytes:
|
|
image_bytes = None
|
|
if type(image_file) is str:
|
|
with open(image_file, "rb") as opened_image_file:
|
|
image_bytes = opened_image_file.read()
|
|
if type(image_file) is io.BytesIO:
|
|
image_file.seek(0)
|
|
image_bytes = image_file.getvalue()
|
|
|
|
# Declare the part to be attached to the multipart message:
|
|
if image_bytes is not None:
|
|
|
|
# Create the HTML block if the image pointer is blank:
|
|
if content_id is None:
|
|
content_id = ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(8))
|
|
self.add_html(f"""
|
|
<html>
|
|
<body>
|
|
<p><img src="cid:{content_id}"></p>
|
|
</body>
|
|
</html>
|
|
""")
|
|
|
|
# Then add the image:
|
|
image_part = MIMEImage(image_bytes)
|
|
image_part.add_header("Content-ID", f"<{content_id}>")
|
|
self.message.attach(image_part)
|
|
|
|
def add_attachment(self, attachment_file, file_name = None):
|
|
|
|
# Declare the part to be attached to the multipart message:
|
|
part = MIMEBase("application", "octet-stream")
|
|
|
|
# If the attachment is a file stored in the local disk:
|
|
if type(attachment_file) is str:
|
|
file_name = file_name or os.path.split(attachment_file)[-1]
|
|
with open(attachment_file, "rb") as attachment:
|
|
part.set_payload(attachment.read())
|
|
|
|
# If the file is held in RAM:
|
|
if type(attachment_file) is io.BytesIO():
|
|
attachment_file.seek(0)
|
|
part.set_payload(attachment_file.read())
|
|
|
|
# Encode and attach the file:
|
|
encoders.encode_base64(part)
|
|
part.add_header(
|
|
"Content-Disposition",
|
|
f"attachment; filename= {file_name}",
|
|
)
|
|
self.message.attach(part)
|
|
|
|
def get_message(self):
|
|
return self.message
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
import asyncio
|
|
from utils import json_utils
|
|
from utils_v2.mail.async_mail import AsyncMailClient
|
|
|
|
async def test():
|
|
|
|
rate_lim = rate_limit_utils.TokenBucket(
|
|
rate_limit = 1,
|
|
seconds = 60.0,
|
|
)
|
|
|
|
mail_client = AsyncMailClient(
|
|
email = "sender@gmail.com",
|
|
password = "secret_password",
|
|
server = "smtp.gmail.com",
|
|
rate_limiters = rate_lim
|
|
)
|
|
|
|
my_mail = MailMessage(
|
|
to_email = "recipient@gmail.com",
|
|
subject = "Bhopli is the best!"
|
|
)
|
|
my_mail.add_html(
|
|
"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<title>Sample HTML String</title>
|
|
<style>
|
|
.heading {
|
|
color: #ff9025;
|
|
}
|
|
.sub-heading {
|
|
color: #000000;
|
|
}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1 class="heading">Hello, Bhopli!</h1>
|
|
<h2 class="sub-heading">Bhopli is the best, most well-behaved cat in the known universe.</h2>
|
|
</body>
|
|
</html>
|
|
"""
|
|
)
|
|
my_mail.add_text("This is how you should pet her 👇")
|
|
my_mail.add_inline_image(r"/path/to/image/cat_petting.png")
|
|
my_mail.add_attachment(r"/path/to/file/sample_label.pdf")
|
|
|
|
await mail_client.login()
|
|
result = await mail_client.send(my_mail)
|
|
print("MAIL RESULT:", json_utils.to_json_string(result))
|
|
await mail_client.logout()
|
|
|
|
asyncio.run(test())
|