A script for sending an email to a user with their previous day's purchases.

daily_purchases_email.py

""""A script for sending an email to a user with their previous day's purchases.
"""
import math
import os
from pprint import pprint
from typing import List
from datetime import datetime, timedelta
from textwrap import dedent
import argparse

from plaid import Client as PlaidClient

plaid_client = PlaidClient(
    client_id=os.getenv("PLAID_CLIENT_ID"),
    secret=os.getenv("PLAID_SECRET"),
    public_key=os.getenv("PLAID_PUBLIC_KEY"),
    environment=os.getenv("PLAID_ENV"),
)


# https://plaid.com/docs/api/#transactions
MAX_TRANSACTIONS_PER_PAGE = 500
OMIT_ACCOUNT_SUBTYPES = ["cd", "savings"]

# https://support.google.com/accounts/answer/6010255?hl=en
# Allow Less Secure App Access to a Gmail account
SMTP_USER = os.getenv("GMAIL_SMTP_USER")
SMTP_PASS = os.getenv("GMAIL_SMTP_PASS")
SMTP_SERVER = "smtp.gmail.com: 587"


def get_day_suffix(day: int) -> str:
    """Determine the suffix for a given day of the month (e.g. "th" for April 20th)
    h/t: https://stackoverflow.com/a/5891598"""
    if 4 <= day <= 20 or 24 <= day <= 30:
        suffix = "th"
    else:
        suffix = ["st", "nd", "rd"][day % 10 - 1]
    return suffix


def get_all_transactions(
    access_token: str, start_date: datetime, end_date: datetime
) -> List[dict]:
    """Query the Plaid API for all transactions within a given date
    h/t: https://www.twilio.com/blog/2017/06/check-daily-spending-sms-python-plaid-twilio.html"""
    # Plaid expects dates in the ISO 8601 format (YYYY-MM-DD).
    start_date = start_date.strftime("%Y-%m-%d")
    end_date = end_date.strftime("%Y-%m-%d")

    account_ids = [
        account["account_id"]
        for account in plaid_client.Accounts.get(access_token)["accounts"]
        if account["subtype"] not in OMIT_ACCOUNT_SUBTYPES
    ]

    num_available_transactions = plaid_client.Transactions.get(
        access_token, start_date, end_date, account_ids=account_ids
    )["total_transactions"]
    num_pages = math.ceil(num_available_transactions / MAX_TRANSACTIONS_PER_PAGE)
    transactions = []

    for page_num in range(num_pages):
        transactions += [
            transaction
            for transaction in plaid_client.Transactions.get(
                access_token,
                start_date,
                end_date,
                account_ids=account_ids,
                offset=page_num * MAX_TRANSACTIONS_PER_PAGE,
                count=MAX_TRANSACTIONS_PER_PAGE,
            )["transactions"]
        ]
    return transactions


def get_transactions_for_email(
    start_date: datetime, end_date: datetime, bank_access_token: str
) -> List[dict]:
    """filter transactions for specifics ones we want for the email"""
    transactions = get_all_transactions(bank_access_token, yesterday, today)

    transactions_for_email = [
        transaction
        for transaction in transactions
        # only filter for purchases, not deposits
        if transaction["amount"] > 0
    ]
    return transactions_for_email


def construct_email_message(transactions: List[dict]) -> str:
    """Construct a message using the list of transactions"""
    purchases = [t["amount"] for t in transactions]
    daily_sum = sum(purchases)
    daily_num = len(purchases)
    sum_message = "You spent ${} on {} purchases yesterday.".format(
        daily_sum, daily_num
    )

    # format each transaction
    daily_transactions = [
        "{} - ${} ({})".format(
            transaction["name"], transaction["amount"], transaction["date"]
        )
        for transaction in transactions
    ]
    plaintext_message = """{sum_message}

    {daily_transactions}
    """.format(
        sum_message=sum_message, daily_transactions="\n".join(daily_transactions)
    )
    return dedent(plaintext_message)


def construct_email_subject(day: datetime) -> str:
    """Construct a daily email subject"""
    import calendar

    yesterday = datetime.now() - timedelta(days=1)
    weekday = calendar.day_name[yesterday.weekday()]
    month_name = calendar.month_name[yesterday.month]
    day_of_month = "{}{}".format(yesterday.day, get_day_suffix(yesterday.day))

    subject = "Your Daily Spending Summary for {weekday}, {month_name} {day_of_month}".format(
        weekday=weekday, month_name=month_name, day_of_month=day_of_month
    )
    return subject


def send_email(subject: str, message: str, recipient: str) -> None:
    from email.mime.multipart import MIMEMultipart
    from email.mime.text import MIMEText
    import smtplib

    msg = MIMEMultipart("alternative")
    msg["From"] = GMAIL_SMTP_USER
    msg["To"] = recipient
    msg["Subject"] = subject
    msg.attach(MIMEText(message, "plain"))
    server = smtplib.SMTP(SMTP_SERVER)
    server.starttls()
    server.login(msg["From"], GMAIL_SMTP_PASS)
    server.sendmail(msg["From"], msg["To"], msg.as_string())
    server.quit()


def construct_and_send_email(
    start_date: datetime, end_date: datetime, recipient: str, bank_access_token: str
):
    transactions = get_transactions_for_email(start_date, end_date, bank_access_token)
    subject = construct_email_subject(end_date)
    message = construct_email_message(transactions)
    send_email(subject, message, recipient)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )
    parser.add_argument("email_recipient", help="Intended recipient of the email.")
    parser.add_argument(
        "bank_access_token", help="Plaid authentication to the user's bank."
    )
    args = parser.parse_args()

    today = datetime.now()
    yesterday = datetime.now() - timedelta(days=1)
    construct_and_send_email(
        yesterday, today, args.email_recipient, args.bank_access_token
    )