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()