""" Email Processor for Genesis System This module provides functionality to read and respond to emails. It uses the imaplib and email libraries to connect to an IMAP server, read emails from a specified mailbox, and send automated responses. """ import imaplib import email from email.mime.text import MIMEText import smtplib import logging import os from typing import List, Tuple, Optional # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) class EmailProcessor: """ A class for processing emails. """ def __init__( self, imap_server: str, imap_port: int, smtp_server: str, smtp_port: int, email_address: str, password: str, mailbox: str = "INBOX", ) -> None: """ Initializes the EmailProcessor. Args: imap_server: The IMAP server address. imap_port: The IMAP server port. smtp_server: The SMTP server address. smtp_port: The SMTP server port. email_address: The email address to use for login. password: The password for the email address. mailbox: The mailbox to read emails from (default: INBOX). """ self.imap_server = imap_server self.imap_port = imap_port self.smtp_server = smtp_server self.smtp_port = smtp_port self.email_address = email_address self.password = password self.mailbox = mailbox def connect_imap(self) -> imaplib.IMAP4_SSL: """ Connects to the IMAP server. Returns: The IMAP connection object. Raises: imaplib.IMAP4.error: If the connection fails. """ try: imap = imaplib.IMAP4_SSL(self.imap_server, self.imap_port) imap.login(self.email_address, self.password) imap.select(self.mailbox) logging.info(f"Connected to IMAP server {self.imap_server}") return imap except imaplib.IMAP4.error as e: logging.error(f"IMAP connection failed: {e}") raise def connect_smtp(self) -> smtplib.SMTP: """ Connects to the SMTP server. Returns: The SMTP connection object. Raises: smtplib.SMTPException: If the connection fails. """ try: smtp = smtplib.SMTP(self.smtp_server, self.smtp_port) smtp.starttls() # Secure the connection smtp.login(self.email_address, self.password) logging.info(f"Connected to SMTP server {self.smtp_server}") return smtp except smtplib.SMTPException as e: logging.error(f"SMTP connection failed: {e}") raise def read_emails(self) -> List[Tuple[str, email.message.Message]]: """ Reads emails from the specified mailbox. Returns: A list of tuples, where each tuple contains the email UID and the email message object. """ try: imap = self.connect_imap() _, message_ids = imap.search(None, "ALL") emails = [] for message_id in message_ids[0].split(): _, data = imap.fetch(message_id, "(RFC822)") raw_email = data[0][1] try: msg = email.message_from_bytes(raw_email) # Handle byte strings correctly emails.append((message_id.decode(), msg)) #Decode the uid except Exception as e: logging.error(f"Error parsing email: {e}") imap.close() imap.logout() logging.info(f"Read {len(emails)} emails from mailbox {self.mailbox}") return emails except Exception as e: logging.error(f"Error reading emails: {e}") return [] def send_email(self, to_address: str, subject: str, body: str) -> None: """ Sends an email. Args: to_address: The recipient's email address. subject: The email subject. body: The email body. """ try: smtp = self.connect_smtp() msg = MIMEText(body) msg["Subject"] = subject msg["From"] = self.email_address msg["To"] = to_address smtp.sendmail(self.email_address, to_address, msg.as_string()) smtp.quit() logging.info(f"Email sent to {to_address} with subject: {subject}") except Exception as e: logging.error(f"Error sending email to {to_address}: {e}") def process_email(self, uid: str, message: email.message.Message) -> None: """ Processes a single email. This is where the core logic lives. Args: uid: The unique identifier of the email. message: The email message object. """ from_address = message.get("From") subject = message.get("Subject") if not from_address: logging.warning(f"Email with UID {uid} has no 'From' address. Skipping.") return if not subject: logging.warning(f"Email with UID {uid} has no 'Subject'. Skipping.") return logging.info(f"Processing email from {from_address} with subject: {subject}") # Example logic: If the subject contains "Genesis Request", send a response if "Genesis Request" in subject: response_body = f"Thank you for your Genesis Request! We are processing it. (UID: {uid})" response_subject = f"Re: {subject}" self.send_email(from_address, response_subject, response_body) logging.info(f"Sent automated response to {from_address} for UID {uid}") else: logging.info(f"No action taken for email from {from_address} with subject: {subject}") def run(self) -> None: """ Runs the email processing loop. Fetches emails and processes them. """ emails = self.read_emails() for uid, email_message in emails: self.process_email(uid, email_message) def main(): """ Main function to run the email processor. """ # Load environment variables (or use hardcoded defaults for testing ONLY) imap_server = os.getenv("IMAP_SERVER") or "your_imap_server.com" # Replace with actual server imap_port = int(os.getenv("IMAP_PORT") or 993) smtp_server = os.getenv("SMTP_SERVER") or "your_smtp_server.com" # Replace with actual server smtp_port = int(os.getenv("SMTP_PORT") or 587) email_address = os.getenv("EMAIL_ADDRESS") or "your_email@example.com" # Replace with actual email password = os.getenv("EMAIL_PASSWORD") or "your_password" # NEVER commit real credentials if not all([imap_server, imap_port, smtp_server, smtp_port, email_address, password]): logging.error("Missing required environment variables. Please set IMAP_SERVER, IMAP_PORT, SMTP_SERVER, SMTP_PORT, EMAIL_ADDRESS, and EMAIL_PASSWORD.") return processor = EmailProcessor( imap_server, imap_port, smtp_server, smtp_port, email_address, password ) processor.run() if __name__ == "__main__": main()