Attachments and rate limiting

Author: MatejVitek

scripts/send_email.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-# USAGE: python send_email.py RECIPIENT_EMAIL [RECEIPIENT_EMAIL ...] -f BODY_FILE [OPTIONS]
+# USAGE: python send_email.py RECIPIENT_EMAIL [RECIPIENT_EMAIL ...] -f BODY_FILE [OPTIONS]
 # Keep in mind that many SMTP providers require app-specific passwords, 2FA, or other security measures.
 # For Gmail specifically, you need to:
 # - enable 2FA;
@@ -11,12 +11,13 @@
 import os
 import re
 
+# Install matej libraries (tested with 0.12.3)
 from matej.argparse import ArgParser, StrArg
 from matej.web.email import send_email
 
 
 class EmailArg(StrArg):
-	EMAIL_RE = re.compile(r'[\w\.-]+@[\w\.-]+\.\w+')
+	EMAIL_RE = re.compile(r'[\w\.-]+@[\w\.-]+\.[\w-]+')
 
 	def _type(self, s):
 		s = super()._type(s)
@@ -32,6 +33,7 @@ def parse_cli_args():
 	ap = ArgParser(description="Send an email via SMTP.")
 	ap.add_arg(EmailArg('emails', help="Addresses of the recipients", nargs='*'))
 	ap.add_path_arg('-f', '--file', help="Path to file containing the email body", required=True)
+	ap.add_path_arg('-a', '--attachment', '--attachments', help="Path(s) to file(s) to attach", nargs='+')
 	ap.add_str_arg('-s', '--subject', help="Email subject")
 	ap.add_str_arg('-u', '--user', help="SMTP login username")
 	ap.add_str_arg('-p', '--password', help="SMTP password (for increased security, set it in the environment variable SMTP_PASSWORD or leave empty to get prompted securely)")
@@ -42,9 +44,10 @@ def parse_cli_args():
 	ap.add_str_arg('-n', '--from-name', help="Sender display name")
 	ap.add_str_arg('-F', '--from-full', help="Override 'From' field with custom value")
 	ap.add_arg(EmailArg('-r', '--reply-to', help="Reply-To email address"))
-	ap.add_arg(EmailArg('-c', '--cc', help="CC emails (will be added on each email if sending individually)", nargs='*'))
-	ap.add_arg(EmailArg('-b', '--bcc', help="BCC emails (will be added on each email if sending individually)", nargs='*'))
+	ap.add_arg(EmailArg('-c', '--cc', help="CC emails (will be added on each email if sending individually)", nargs='+'))
+	ap.add_arg(EmailArg('-b', '--bcc', help="BCC emails (will be added on each email if sending individually)", nargs='+'))
 	ap.add_bool_arg('-i', '--send-individually', help="Send emails to each recipient separately", default=True)
+	ap.add_number_arg('-l', '--limit', '--rate-limit', help="Rate limit (in emails per second)", nargs=1, default=None, min=0)
 	return ap.parse_args()
 
 
@@ -80,6 +83,10 @@ def detect_html(content):
 def main():
 	args = parse_cli_args()
 
+	# Validate recipients
+	if not args.emails and not args.cc and not args.bcc:
+		raise ValueError("At least one recipient email address is required (To, CC, or BCC).")
+
 	# SMTP username prompt (with repeat on empty)
 	while not args.user:
 		args.user = input("SMTP username (email): ").strip()
@@ -106,6 +113,7 @@ def main():
 		plain_text = raw
 		html = None
 
+	# Send the email
 	send_email(
 		smtp_server=args.server,
 		smtp_port=args.port,
@@ -122,7 +130,9 @@ def main():
 		subject=args.subject,
 		plain_text=plain_text,
 		html_content=html,
+		attachments=args.attachment,
 		send_individually=args.send_individually,
+		rate_limit=args.limit,
 	)
 
 

src/matej/web/email.py

@@ -1,16 +1,21 @@
 import asyncio as aio
 from email.message import EmailMessage
 from email.utils import make_msgid, formataddr
+import mimetypes
+from pathlib import Path
 import smtplib
+import time
 import warnings
 
+from matej.collections import ensure_iterable
+
 try:
 	import aiosmtplib
 except ImportError:
 	aiosmtplib = None
 
 
-def _build(subject, plain_text, from_name, from_email, from_full=None, to_addrs=None, cc_addrs=None, reply_to=None, html_content=None):
+def _build(subject, plain_text, from_name, from_email, from_full=None, to_addrs=None, cc_addrs=None, reply_to=None, html_content=None, attachments=None):
 	msg = EmailMessage()
 	msg['Subject'] = subject
 	msg['From'] = from_full if from_full else formataddr((from_name, from_email))
@@ -21,10 +26,28 @@ def _build(subject, plain_text, from_name, from_email, from_full=None, to_addrs=
 		msg['Reply-To'] = reply_to
 	msg['Message-ID'] = make_msgid()
 	msg['X-Priority'] = '3'
-	msg['X-Mailer'] = 'Python EmailSender'
+	msg['X-Mailer'] = 'PyMailer'
+	if from_email:
+		if 'gmail' in from_email.lower():
+			msg['XMailer'] = 'Gmail'    # Maybe less chance of spam filters
+		elif 'outlook' in from_email.lower():
+			msg['XMailer'] = 'Outlook'  # if we use these headers?
 	msg.set_content(plain_text)
 	if html_content:
 		msg.add_alternative(html_content, subtype='html')
+
+	if attachments:
+		for attachment in ensure_iterable(attachments):
+			attachment = Path(attachment)
+			if not attachment.is_file():
+				raise ValueError(f"Attachment not found or is not a file: {attachment}")
+			content_type, _ = mimetypes.guess_type(attachment)
+			if content_type is None:
+				content_type = 'application/octet-stream'
+			main_type, sub_type = content_type.split('/', 1)
+			with attachment.open('rb') as f:
+				msg.add_attachment(f.read(), maintype=main_type, subtype=sub_type, filename=attachment.name)
+
 	return msg
 
 
@@ -57,7 +80,7 @@ def _infer_email_address(user, server):
 	return f'{user}@{domain}'
 
 
-def send_email(smtp_server, smtp_user, smtp_password, to_emails=None, cc_emails=None, bcc_emails=None, reply_to=None, subject='', plain_text='', html_content=None, smtp_port=465, smtp_protocol=None, from_name='', from_email='', from_full=None, send_individually=False, asynchronous=True):
+def send_email(smtp_server, smtp_user, smtp_password, to_emails=None, cc_emails=None, bcc_emails=None, reply_to=None, subject='', plain_text='', html_content=None, attachments=None, smtp_port=465, smtp_protocol=None, from_name='', from_email='', from_full=None, send_individually=False, asynchronous=True, rate_limit=None):
 	"""
 	Send an email with support for To, CC, BCC, and optional individual sending.
 
@@ -69,11 +92,11 @@ def send_email(smtp_server, smtp_user, smtp_password, to_emails=None, cc_emails=
 		Username for SMTP authentication (usually the sender's email).
 	smtp_password : str
 		Password or app-specific password for SMTP authentication.
-	to_emails : List[str], optional
+	to_emails : Collection[str], optional
 		List of recipient email addresses.
-	cc_emails : List[str], optional
+	cc_emails : Collection[str], optional
 		List of CC (carbon copy) email addresses.
-	bcc_emails : List[str], optional
+	bcc_emails : Collection[str], optional
 		List of BCC (blind carbon copy) email addresses.
 	reply_to : str, optional
 		Email address for the Reply-To header.
@@ -83,6 +106,8 @@ def send_email(smtp_server, smtp_user, smtp_password, to_emails=None, cc_emails=
 		Plain-text version of the email body.
 	html_content : str, optional
 		HTML version of the email body. If provided, the email will be sent as multipart/alternative.
+	attachments : Collection[str], optional
+		List of file paths to attach to the email.
 	smtp_port : int, default=465
 		Port number for SMTP (typically 465 for SSL, 587 for STARTTLS).
 	smtp_protocol : str, optional
@@ -94,16 +119,16 @@ def send_email(smtp_server, smtp_user, smtp_password, to_emails=None, cc_emails=
 	from_full : str, optional
 		Override the "From" field with a custom value.
 	send_individually : bool, default=False
-		If True, send one email per recipient (To + CC + BCC each get a separate message).
+		If `True`, send one email per recipient (To + CC + BCC each get a separate message).
 	asynchronous : bool, default=True
-		Use aiosmtplib for asnychronous sending.
+		Send emails asynchronously.
+	rate_limit : int, optional
+		Number of emails to send per second. If `None` or `0`, no rate limiting is applied.
 	"""
 	to_emails = to_emails or []
 	cc_emails = cc_emails or []
 	bcc_emails = bcc_emails or []
 
-	all_recipients = list(set(to_emails + cc_emails + bcc_emails))
-
 	# Determine SMTP protocol if not provided
 	if smtp_protocol is None:
 		smtp_protocol = 'ssl' if smtp_port == 465 else 'starttls' if smtp_port == 587 else ''
@@ -129,19 +154,30 @@ def send_email(smtp_server, smtp_user, smtp_password, to_emails=None, cc_emails=
 			# Cc list: all CC recipients
 			cc_list = cc_emails if cc_emails else []
 			# Build email message with one TO, full CC, no BCC in headers
-			msg = _build(subject, plain_text, from_name, from_email, from_full, to_list, cc_list, reply_to, html_content)
+			msg = _build(subject, plain_text, from_name, from_email, from_full, to_list, cc_list, reply_to, html_content, attachments)
 			# Recipients for SMTP envelope include TO, CC, and BCC
 			smtp_recipients = list(set(to_list + cc_list + bcc_emails))
 			msgs.append((msg, smtp_recipients))
 	else:
-		msg = _build(subject, plain_text, from_name, from_email, from_full, to_emails, cc_emails, reply_to, html_content)
-		msgs = [(msg, all_recipients)]
+		msg = _build(subject, plain_text, from_name, from_email, from_full, to_emails, cc_emails, reply_to, html_content, attachments)
+		msgs = [(msg, list(set(to_emails + cc_emails + bcc_emails)))]
 
 	if asynchronous:
 		async def _send_all():
-			tasks = [_send_async(recipients, msg, smtp_server, smtp_port, smtp_protocol, smtp_user, smtp_password) for msg, recipients in msgs]
+			semaphore = aio.Semaphore(1)
+
+			async def limited_send(i, msg, recipients):
+				async with semaphore:
+					if rate_limit and i > 0:
+						await aio.sleep(1 / rate_limit)
+					await _send_async(recipients, msg, smtp_server, smtp_port, smtp_protocol, smtp_user, smtp_password)
+
+			tasks = [aio.create_task(limited_send(i, msg, recipients)) for i, (msg, recipients) in enumerate(msgs)]
 			await aio.gather(*tasks)
+
 		return aio.run(_send_all())
 
-	for msg, recipients in msgs:
+	for i, (msg, recipients) in enumerate(msgs):
+		if rate_limit and i > 0:
+			time.sleep(1 / rate_limit)
 		_send(recipients, msg, smtp_server, smtp_port, smtp_protocol, smtp_user, smtp_password)