I wrote this script to replace a VBA script for mailbox management that was in place before. Essentially, I have an Excel sheet in which the rules are defined. These have to be loaded and applied to the mails currently in the inbox. The file's structure looks like this:

criteria (required): "begins with" or "contains"
filename (optional): "filename part"
address (optional): "example@email.com"
subject (optional): "subject part"
directory (optional): "C:\\my\\custom\\abs\\path\\"
outlook_folder_1 (required): "folder name"
outlook_folder_2 (optional): "subfolder name"

Emails should be moved to the specified Outlook folder. For each mail, first address and subject are checked for matches and the mail is moved to the specified folder. Then attachments are checked, mail should be moved, if not moved yet, and the attachment downloaded, if a directory is provided.

The script is currently working fine, but I'm wondering if it'd be possible to improve performance somehow. Also, I was thinking about replacing the criteria column from the file and use regex to search mails based on patterns. What are your thoughts?


"""Central mailbox management rules"""
import logging
import os
import zipfile

import pandas as pd
from pandas.tseries.offsets import BDay
import win32com.client

from misc import sanitize_col_name

# create a logger
logger = logging.getLogger(__name__)

# get today's date
today = pd.datetime.now().date()

# defining paths and constants
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
CENTRAL_MAILBOX = "Central mailbox"

# get the outlook instance and inbox folders
outlook = win32com.client.Dispatch("Outlook.Application").Session
account = outlook.Accounts(1)
personal_inbox = outlook.Folders(
user = outlook.CreateRecipient(CENTRAL_MAILBOX)
central_inbox = outlook.GetSharedDefaultFolder(user, 6)

def load_criteria():
    criteria_path = os.path.join(
        ROOT, "rsc", "attachment_criteria.xlsx")
    df = pd.read_excel(criteria_path)
    for i in range(len(df)):
        row = df.iloc[-1]
        if row.isna().all():
            df = df[:-1]
    df.columns = [sanitize_col_name(c) for c in df.columns]
    return df

criteriadf = load_criteria()

def check_attachment(row, att):
    if not isinstance(row.filename, str):
        return False
    if att.Type == 6 or att.FileName.split(".")[1] in ["jpg", "png"]:
        return False
    if row.criteria == "begins with" and row.filename == att.FileName[:len(row.filename)]:
        return True
    elif row.criteria == "contains" and row.filename in att.FileName:
        return True
        return False

def check_subject(row, mail):
    if not isinstance(row.subject, str):
        return False
    if row.criteria == "begins with" and row.subject == mail.Subject[:len(row.subject)]:
        return True
    elif row.criteria == "contains" and row.subject in mail.Subject:
        return True
        return False

def check_zip_and_extract(filepath):
    filetype = filepath.split(".")[1]
    if not filetype == "zip":
        return None
    logger.info("The attachment is a zip file. Extracting now..")
    directory = os.path.dirname(filepath)
    zip_ref = zipfile.ZipFile(filepath, "r")
    logger.info(f"Extracted zip file to {directory}")

def get_mails(days):
    if not isinstance(days, int):
        logger.info("Days has to be an int. Aborting..")
        return None
    if central_inbox.Items.Count == 0:
        logger.info("There are no messages in the central inbox. Aborting..")
        return None
    retrieve_from = today - BDay(days)
    logger.info(f"Retrieving mails from {retrieve_from.strftime('%Y-%m-%d')} onwards.")
    mails = central_inbox.Items.Restrict(
        "[ReceivedTime] > '{}'".format(retrieve_from.strftime("%d/%m/%Y %H:%M"))
    logger.info(f"Retrieved {len(mails)} mails.")
    return mails

def run(mails):
    moved = 0
    downloaded = 0
    for mail in mails:
        for i, row in criteriadf.iterrows():
            if ((mail.SenderEmailAddress == row.address and (
                    not isinstance(row.subject, str) or check_subject(row, mail)))
                    or (not isinstance(row.address, str) and check_subject(row, mail))):
                    f"The mail from {mail.SenderEmailAddress} with subject {mail.Subject} matches the criteria at index {row.name}")
                logger.info(f"Moved mail {mail.Subject} to folder {row.outlook_folder_1}")
                moved += 1
                logger.debug("Total moved: %s", moved)

            if mail.Attachments.Count == 0:
            for att in mail.Attachments:
                att_valid = check_attachment(row, att)
                if (not isinstance(row.address, str)
                        and not isinstance(row.subject, str)
                        and att_valid):
                    logger.info(f"The attachment {att.FileName} matches the criteria at index {row.name}\n"
                                f"Mail has not been moved based on address or subject criteria yet.\n"
                                f"Moving based on attachment..")
                    logger.info(f"Moved mail {mail.Subject} to folder {row.outlook_folder_1}")
                    moved += 1
                    logger.debug("Total moved: %s", moved)

                if att_valid and isinstance(row.directory, str):
                    filepath = os.path.join(row.directory, att.FileName)
                    logger.info(f"Saved attachment {att.FileName} under {filepath}")
                    downloaded += 1
                    logger.debug("Total downloaded: %s", downloaded)

    return moved, downloaded


"""Util functions"""
import json
import logging
import logging.config
import re

with open(r"C:\path\to\config.json", "r") as f:
    log_config = json.load(f)


def get_root_logger():
    root_logger = logging.getLogger("root")
    return root_logger

# sanitize column names
def sanitize_col_name(name):
    nosigns = re.compile("[^A-Za-z0-9_\s]")
    nname = re.sub(nosigns, "", name).strip().replace(" ", "_").lower()
    nname = "empty" if nname == "" else nname
    return nname


"""Main script to call the program from CLI"""
import argparse
import sys

import harvester as harvester
from misc import get_root_logger

# getting the root logger
logger = get_root_logger()

# parsing arguments
parser = argparse.ArgumentParser(
    description="Download Outlook attachments according to pre-defined rules")
parser.add_argument("days", type=int, nargs="?", default=1)
parser.add_argument("--verbose", "-v", action="count")

def main():
    args = vars(parser.parse_args())
    bdays_back = args["days"]
    mails = harvester.get_mails(bdays_back)
    moved, downloaded = harvester.run(mails)
    logger.info(f"Downloaded {downloaded} attachments and moved {moved} emails.\n"
                "Have a nice day!")

if __name__ == "__main__":
    logger.info("Initialising email mover")
    status = main()

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.