amazon2monarch.py

import csv
import argparse
from datetime import datetime
import os
from rich import print
from ollama import generate
from tqdm import tqdm

key_website = 'Website'
key_order_date = 'Order Date'
key_total_cost = 'Total Owed'
key_name = 'Product Name'
key_payment_type = 'Payment Instrument Type'

whole_foods_value = 'panda01'
amazon_com_value = 'Amazon.com'

def convert(
        inputfile, 
        outputfile, 
        since_date,
        output_category, 
        output_merchant_prefix, 
        output_account,
        output_tags,
        output_notes_prefix
    ):
    print('\n')
    print (f'Input CSV file is: ', inputfile.name)
    print (f'Output CSV will be written at: ', outputfile.name)

    amazon_transactions = 0
    whole_foods_transactions = 0
    zero_cost_transactions = 0

    csv_reader = csv.DictReader(inputfile, delimiter=',')

    output_field_names = ['Date', 'Merchant', 'Category', 'Account', 'Original Statement', 'Notes', 'Amount', 'Tags']
    csv_writer = csv.DictWriter(outputfile, output_field_names)
    csv_writer.writeheader()

    rows_to_process = []
    
    for row in csv_reader:
        if row[key_website] == whole_foods_value:
            whole_foods_transactions += 1
        elif row[key_website] == amazon_com_value:
            transaction_date = strptime(row[key_order_date])
            if row[key_total_cost] == '0':
                zero_cost_transactions += 1
            elif since_date is None:
                rows_to_process.append(row)
                amazon_transactions += 1
            elif since_date is not None and since_date.date() <= transaction_date.date():
                rows_to_process.append(row)
                amazon_transactions += 1

    print('\n')
    if since_date is not None:
        print (f'{amazon_transactions} transactions after this date will be processed: ', since_date.date())
    else:
        print (f'A total of {amazon_transactions} transactions will be processed.')

    print('\n')
    input(f'Press Enter to continue, or Control+C to abort...')
    print('\n')

    for row in tqdm(rows_to_process):
        csv_writer.writerow({
            'Date': str(strptime(row[key_order_date]).date()),
            'Merchant': output_merchant_prefix + ai_summary_merchant(row[key_name]),
            'Category': output_category,
            'Account': output_account,
            'Original Statement': row[key_name],
            'Notes': output_notes_prefix + row[key_payment_type],
            'Amount': '-' + row[key_total_cost],
            'Tags': output_tags
        })

    print('\n')
    if since_date is not None:
        print(f'Processed {amazon_transactions} Amazon.com transactions since {str(since_date.date())}.')
    else:
        print(f'Processed {amazon_transactions} Amazon.com transactions.')
    print(f'Ignored {whole_foods_transactions} Whole Foods transactions.')
    print(f'Ignored {zero_cost_transactions} transactions with $0 order amounts.')

def strptime(str):
    if '.' not in str:
        return datetime.strptime(str, '%Y-%m-%dT%H:%M:%S%z')
    else:
        return datetime.strptime(str, '%Y-%m-%dT%H:%M:%S.%f%z')
    
def arg_date(str):
    try:
        return datetime.strptime(str, '%Y-%m-%d')
    except ValueError:
        msg = 'not a valid date: {0!r}'.format(str)
        raise argparse.ArgumentTypeError(msg)
    
def ai_summary_merchant(original_statement):
    prompt = "Your job is to summarize an Amazon.com transaction original statement information to shorter item name to be imported as a 'Merchant' into a budgeting app. Generated summary should be strictly not longer than 5 words. Optimize for the reader to learn most important thing about the product when reading the summary. Just output the summary, nothing else. Don't provide any suggestions or alternatives, be definitive and generate only one summary. The text for original statement is - "
    output = generate(model='granite3.3', prompt=prompt+original_statement)
    return output.response.strip('\"')

parser = argparse.ArgumentParser(
    prog='amazon2monarch',
    description='This script converts Amazon Order History (Retail.OrderHistory.1.csv file) to a CSV that can be imported into Monarch.',
)
parser.add_argument('-i', '--input_csv', 
                    required=True, 
                    type=argparse.FileType('r', encoding='utf-8-sig'),
                    help='Path of the input CSV. This is usually the Retail.OrderHistory.1.csv file from your Amazon Data Request.')
parser.add_argument('-o', '--output_csv', 
                    required=True, 
                    type=argparse.FileType('w'), 
                    help='Path of the output CSV. This is the file you will upload to Monarch.')
parser.add_argument('-d', '--since_date', 
                    required=False, 
                    type=arg_date,
                    help='Process transactions past since this date. If not provided, will process all transactions.')
parser.add_argument('-oc', '--output_category',
                    default='Uncategorized',
                    help='Set the category of all transactions in the output file. Defaults to \'Uncategorized\'.')
parser.add_argument('-om', '--output_merchant_prefix',
                    default='',
                    help='Prefix each transactions with a text for ease of reading. Defaults to empty.')
parser.add_argument('-oa', '--output_account',
                    default='Amazon Gift Card Balance',
                    help='Name of the account on Monarch for all the processed transactions. Defaults to \'Amazon Gift Card Balance\'.')
parser.add_argument('-ot', '--output_tags',
                    default='Amazon Gift Card',
                    help='Comma separated list of Tags on Monarch to be added to all transactions. Defaults to \'Amazon Gift Card\'.')
parser.add_argument('-on', '--output_notes_prefix',
                    default='Payment Method: ',
                    help='Prefix text to be added to transaction notes in the output files. Payment method is added by default, and the prefix \'Payment Method: \' is added.')

args = parser.parse_args()

try:
    convert(
        args.input_csv, 
        args.output_csv, 
        args.since_date, 
        args.output_category,
        args.output_merchant_prefix,
        args.output_account,
        args.output_tags,
        args.output_notes_prefix
    )
except Exception as e:
    print(f'\nUh Oh. We failed because of the following reason: ')
    print(e.with_traceback)
    args.input_csv.close()
    args.output_csv.close()
    os.remove(args.output_csv.name)
finally:
    args.input_csv.close()
    args.output_csv.close()
    epilog=f'Thank you for using this script. If it helped you, buy me a beer on [link=https://venmo.com/u/rahuljiresal]Venmo[/link] or [link=https://paypal.me/rahuljiresal]Paypal[/link]. Cheers! 🍻'
    print('\n')
    print(epilog)