91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
import logging
|
|
import os
|
|
import csv
|
|
import subprocess
|
|
|
|
from datetime import date, datetime
|
|
from collections import namedtuple
|
|
|
|
from tx_categories import TX_TO_CATEGORY, Category
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
# poetry run python process_csv.py
|
|
# DESIRED_MONTH = date.today().month
|
|
DESIRED_MONTH = 2
|
|
CSV_TX = namedtuple('CSV_TX', ['TX_DATE', 'POST_DATE', 'DESCRIPTION', 'CATEGORY', 'TYPE', 'AMOUNT', 'MEMO'])
|
|
|
|
class Transaction:
|
|
def __init__(self, date, payee, type, amount):
|
|
# self.date = date.replace(tzinfo=pytz.utc).astimezone(pytz.timezone('America/Chicago'))
|
|
self.date = date
|
|
self.payee = payee
|
|
self.type = type
|
|
self.amount = amount
|
|
self.category = self.get_category()
|
|
|
|
def get_category(self):
|
|
category = None
|
|
for agent, cat in TX_TO_CATEGORY.items():
|
|
if agent.lower() in self.payee.lower():
|
|
category = cat
|
|
|
|
return category or Category.OTHER
|
|
|
|
def load_chase_csv():
|
|
working_dir = os.path.abspath(os.getcwd())
|
|
filenames = os.listdir(working_dir)
|
|
tx_filename = None
|
|
for filename in filenames:
|
|
if filename.lower().startswith("chase"):
|
|
tx_filename = os.path.join(working_dir, filename)
|
|
break
|
|
|
|
if not tx_filename:
|
|
exit("Could not find tx file")
|
|
|
|
with open(tx_filename) as csvfile:
|
|
reader = csv.reader(csvfile, delimiter=',', quotechar='|')
|
|
output = list(reader)
|
|
|
|
all_transactions = []
|
|
for raw_tx in output[1:]:
|
|
all_transactions.append(CSV_TX(*raw_tx))
|
|
|
|
return all_transactions
|
|
|
|
def get_transactions(all_transactions):
|
|
formatted_transactions = []
|
|
for tx in all_transactions:
|
|
tx_post_date = datetime.strptime(tx.POST_DATE, '%m/%d/%Y')
|
|
if tx_post_date.month != DESIRED_MONTH or tx_post_date.year != date.today().year:
|
|
continue
|
|
cc_trans = Transaction(
|
|
date=tx_post_date,
|
|
payee=tx.DESCRIPTION,
|
|
type=tx.TYPE,
|
|
amount=tx.AMOUNT
|
|
)
|
|
|
|
formatted_transactions.append(cc_trans)
|
|
|
|
sorted_transactions = sorted(formatted_transactions, key=lambda x: x.date, reverse=False)
|
|
return sorted_transactions
|
|
|
|
def print_transactions(transactions):
|
|
for tx in transactions:
|
|
print(f"{tx.date} - {tx.payee} - {tx.amount} - {tx.category}")
|
|
|
|
def convert_tx_to_csv(transactions):
|
|
output = ""
|
|
for tx in transactions:
|
|
output += f"{tx.date.strftime('%m/%d/%Y')},{tx.payee},{tx.amount},{tx.category.value}\n"
|
|
return output
|
|
|
|
if __name__ == "__main__":
|
|
all_transactions = load_chase_csv()
|
|
transactions = get_transactions(all_transactions)
|
|
print_transactions(transactions)
|
|
output = convert_tx_to_csv(transactions)
|
|
subprocess.run("pbcopy", universal_newlines=True, input=output)
|
|
print("TXs copied to clipboard!")
|