finance-watcher / scripts / import_from_sheet.py
import_from_sheet.py
Raw
from typing import List, Any, Optional
from os import path
import sys
import csv
from datetime import datetime
from finance_watcher_dataclasses.import_dataclasses import AccountRow
from finance_watcher_dataclasses.enums import AccountTypeEnum as AccountTypeEnum
from database.finance_database import FinanceDatabase

EXIT_ERROR = -1
EXIT_SUCCESS = 0
DATETIME_FORMAT = "%Y-%m-%d"
NUM_CHARS_TO_REMOVE = 3 ## includes the () and the space

## check if only one argument, then check if file exists
def check_arguements(arguments: List[str]) -> bool:
    if len(arguments) != 2:
        print("Please enter a file to import from. (Should be in a .csv)")
        return False
    if not path.isfile(arguments[1]):
        print("The file does not exist. Please enter the correct path.")
        return False
    return True

def get_account_row(
    row: List[Any], 
    dates: List[str],
    finance_db: FinanceDatabase
) -> Optional[AccountRow]:
    account_string: str = row[0]

    account_type = AccountTypeEnum.get_account_type_from_str(account_string)
    if account_type is None:
        return None
    
    account_type_from_db = finance_db.ensure_account_type(
        account_type=account_type
    )
    if account_type_from_db is None:
        return None

    account_name = account_string[len(account_type.value)+NUM_CHARS_TO_REMOVE:]

    date_to_amount = {}
    for index in range(1,len(row)):
        date = datetime.strptime(dates[index],DATETIME_FORMAT)
        amount_str: str = row[index]
        try:
            amount_without_commas = amount_str.replace(',','')
            amount = float(amount_without_commas)
            date_to_amount[date] = amount
        except:
            continue

    account_found = finance_db.ensure_account(
        name=account_name,
        account_type_id=account_type_from_db.id,
        dashboard_user_id=1
    )
    if account_found is None:
        return None

    return AccountRow(
        account_id=account_found.id,
        date_to_amount=date_to_amount
    )

## go through line by line and format properly and insert into the database

def import_from_spreadsheet():
    arguments = sys.argv
    if not check_arguements(arguments):
        exit(EXIT_ERROR)

    file = open(arguments[1])
    csv_reader = csv.reader(file)
    dates_row = next(csv_reader) # first row is dates
    finance_db = FinanceDatabase()
    account_rows: List[AccountRow] = []
    for row in csv_reader:
        if row[0] == '':
            continue
        if len(row) != len(dates_row):
            print(f"Read error for income stream `{row[0]}`. Skipping...")
            continue
        account_row = get_account_row(
            row, dates_row, finance_db=finance_db
        )
        if account_row is None:
            continue
        account_rows.append(account_row)

    ## insert into the database
    finance_db.insert_account_totals(
        account_rows
    )
    print("Done.")

import_from_spreadsheet()