How do I:
I've checked similar posts, but I can't manage to actually connect it to my mailbox and make it all work without having to do anything manually.
I'm using iOS'/iCloud's IMAP (imap.mail.me.com) and I'm using mail.login() right now, but it doesn't seem to actually get any content from unread emails.
The code I have right now:
import imaplib
import email
from email.header import decode_header
from OpenSSL import crypto
import time, datetime
# Email receiving configuration
imap_server = "imap.mail.me.com" # <-- iCloud's mail IMAP
imap_user = "example@example.com" # <- Email
imap_password = "example_password" # <-- Email App-Specific password
# Paths to your certificate and private key
cert_path = 'path\to\my\certificate.pem' #<-- Path to cert.pem
key_path = 'path\to\my\privatekey.pem' # <-- Path to privkey.pem
# Function to decrypt S/MIME email
def decrypt_smime(encrypted_message):
try:
# Load the certificate and private key
with open(cert_path, 'rb') as f:
cert = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
with open(key_path, 'rb') as f:
key = crypto.load_privatekey(crypto.FILETYPE_PEM, f.read())
# Create a Crypto object and decrypt the email
smime = crypto.SMIME()
smime.load_cert(cert)
smime.load_key(key)
decrypted_message = smime.decrypt(encrypted_message)
return decrypted_message.decode('utf-8')
except Exception as e:
print(f"Error decrypting S/MIME email: {e}")
return None
# Function to process email payload
def process_email_payload(msg):
if msg.is_multipart():
for part in msg.walk():
content_type = part.get_content_type()
if content_type == "application/pkcs7-mime":
payload = part.get_payload(decode=True)
payload = decrypt_smime(payload)
if payload:
return payload.strip()
elif content_type in ["text/plain", "text/html"]:
payload = part.get_payload(decode=True).strip()
return payload.decode('utf-8')
else:
payload = msg.get_payload(decode=True)
if msg.get_content_type() == "application/pkcs7-mime":
payload = decrypt_smime(payload)
return payload.decode('utf-8')
return None
# Function to check for new emails
def check_email():
try:
mail = imaplib.IMAP4_SSL(imap_server)
mail.login(imap_user, imap_password)
mail.select("inbox")
result, data = mail.search(None, "UNSEEN")
# Debugging information
print(f"Search result {datetime.datetime.now():%H.%M.%S}: {result}") # <-- Also prints current time so I can differentiate the different outputs
print(f"Search data: {data}")
if result != "OK":
print(f"Error searching Inbox: {result}")
return
if data[0] is None:
print("No new emails.")
return
email_ids = data[0].split()
if not email_ids:
print("No new emails.")
return
print(f"Email IDs: {email_ids}") # Debug email IDs
for email_id in email_ids:
result, msg_data = mail.fetch(email_id, "(RFC822)")
print(f"Fetch result: {result}") # Debug fetch result
print(f"Fetch msg_data: {msg_data}") # Debug msg_data
if result != "OK":
print(f"Error fetching email ID {email_id}: {result}")
continue
for response_part in msg_data:
if isinstance(response_part, tuple):
raw_email = response_part[1]
print(f"Raw email: {raw_email}") # Debug raw email data
try:
msg = email.message_from_bytes(raw_email)
subject, encoding = decode_header(msg["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding if encoding else 'utf-8')
sender = email.utils.parseaddr(msg["From"])[1]
print(f"Subject: {subject}")
print(f"From: {sender}")
payload = process_email_payload(msg)
if payload:
print(f"Payload: {payload}") # Debug payload content
else:
print(f"Could not decode payload for email ID {email_id}")
continue
if "test" in payload:
print("Test prompt received")
except Exception as e:
print(f"Error decoding email ID {email_id}: {e}")
continue
mail.close()
mail.logout()
except Exception as e:
print(f"Failed to check email: {e}")
# Actually try to check and decode emails (loop)
while True:
check_email()
time.sleep(10)
best code as of now, programmed in Visual Studio Code on windows, not Linux
It does actively check for emails, but it doesn't fetch anything:
Fetch result: OK
Fetch msg_data: [b'0000 ()']
terminal output
and doesn't print subject, etc., because it isn't isinstance(response_part, tuple) I assume and msg_data is literally empty (basically).
And I've tried dozens of other possible solutions, but this seems to be the best for now.
So how do I actually actively check for unread emails and decode their body?
Please try your best with describing the solution, since I'm really interested in learning more information about this and I would like to improve my over-all coding skills. I'm also interested in what I did wrong.
I managed to make it work, the main problem was the way the mail was fetched. When changing this to:
result, msg_data = mail.fetch(email_id, "(BODY.PEEK[])")
It seems to work perfectly fine.
Fully updated check_mail():
def check_email():
try:
mail = imaplib.IMAP4_SSL(imap_server)
mail.login(imap_user, imap_password)
mail.select("inbox")
result, data = mail.search(None, "UNSEEN")
# Debugging information
print(f"Search result {datetime.datetime.now():%H.%M.%S}: {result}")
print(f"Search data: {data}")
if result != "OK":
print(f"Error searching Inbox: {result}")
return
if not data or not data[0]:
print("No new emails.")
return
email_ids = data[0].split()
if not email_ids:
print("No new emails.")
return
# print(f"Email IDs: {email_ids}") # Debug email IDs
for email_id in email_ids:
# Try fetching different parts of the email
result, msg_data = mail.fetch(email_id, "(BODY.PEEK[])")
# print(f"Fetch result: {result}") # Debug fetch result
# print(f"Fetch msg_data: {msg_data}") # Debug msg_data
if result != "OK":
print(f"Error fetching email ID {email_id}: {result}")
continue
if not msg_data or msg_data == [b'']:
print(f"No data fetched for email ID {email_id}")
continue
# Checking each part of the message data
for response_part in msg_data:
# print(f"Response part: {response_part}") # Debug each response part
if isinstance(response_part, tuple):
raw_email = response_part[1]
# print(f"Raw email: {raw_email[:500]}...") # Debug raw email data (truncated)
try:
msg = email.message_from_bytes(raw_email)
subject, encoding = decode_header(msg["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding if encoding else 'utf-8')
sender = email.utils.parseaddr(msg["From"])[1]
print(f"Subject: {subject}")
print(f"From: {sender}")
payload = process_email_payload(msg)
if payload:
# print(f"Payload: {payload}") # Debug payload content
""
else:
print(f"Could not decode payload for email ID {email_id}")
continue
"""
You can put all kinds of if statements here, for example:
if "test" in payload:
print("Test Prompt received")
To check if there's "test" in the body of the email, or
if "test" in subject:
print("Test Subject received")
To check if there's test in the body. You can also check the sender and do stuff based off of the sender, for example:
if sender == "example@example.com" and "test" in payload:
print("The sender of the email was (...) and there was 'test' in the email's body")
"""
mail.store(email_id, '+FLAGS', '\\Seen')
mail.store(email_id, '+FLAGS', '\\Deleted')
# Automatically marks the sent email as "Seen/Read" and moves it to "Deleted/Bin".
except Exception as e:
print(f"Error decoding email ID {email_id}: {e}")
continue
mail.close()
mail.logout()
except Exception as e:
print(f"Failed to check email: {e}")
You can remove the comments for some prints to receive the raw body of the email, but this gives the cleanest output with the most important information.