From c8795fc1e31a9ee0f336640956d43a6ba60b8f8a Mon Sep 17 00:00:00 2001 From: Sindre Stephansen Date: Mon, 13 Oct 2025 14:57:42 +0200 Subject: [PATCH] Move API code to separate file --- src/convert.py | 70 +-------------------------- src/ynab_api.py | 124 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 126 insertions(+), 68 deletions(-) create mode 100644 src/ynab_api.py diff --git a/src/convert.py b/src/convert.py index 7baea98..ff9eacc 100644 --- a/src/convert.py +++ b/src/convert.py @@ -1,9 +1,9 @@ import glob import logging import pandas as pd -import ynab from pathlib import Path +from ynab_api import write_to_api from bank_parsers.sparebank1 import parse_sparebank1 from bank_parsers.bank_norwegian import parse_bank_norwegian from bank_parsers.sparebanken_norge import parse_sparebanken_norge @@ -121,72 +121,6 @@ def write_output_file(bank_name, bank_config, ynab_data, output_directory, overw logging.info(f"Data saved to {output_file}") -def write_to_api(file_path, ynab_data, api_token): - configuration = ynab.Configuration( - access_token = api_token - ) - - with ynab.ApiClient(configuration) as api_client: - budgets_api = ynab.BudgetsApi(api_client) - budgets = budgets_api.get_budgets().data.budgets - - if len(budgets) == 1: - budget = budgets[0] - else: - budget_options = ", ".join([f"{i}: {budget.name}" for i, budget in enumerate(budgets)]) - print(f"Which budget should {file_path} be imported to?") - budget_index = input(f"{budget_options} ") - budget = budgets[int(budget_index)] - - accounts_api = ynab.AccountsApi(api_client) - accounts = accounts_api.get_accounts(budget.id).data.accounts - - if len(accounts) == 1: - account = accounts[0] - else: - account_options = ", ".join([f"{i}: {account.name}" for i, account in enumerate(accounts)]) - account_index = input(f"Which account? {account_options} ") - account = accounts[int(account_index)] - - transactions = [] - - for i, row in ynab_data.iterrows(): - import_id = f"IMPORT:{row['Date'].date()}:{row['Inflow']}:{row['Outflow']}:{i}" - amount = int((row['Inflow'] - row['Outflow']) * 100) - - try: - transaction = ynab.models.new_transaction.NewTransaction( - account_id = account.id, - var_date = row['Date'], - amount = amount, - payee_name = row['Payee'], - memo = row['Memo'], - cleared = 'cleared' if row['Cleared'] else 'uncleared', - import_id = import_id, - ) - except: - logging.error(f"Could not format transaction {row} for YNAB") - raise - - transactions.append(transaction) - - print(f"Adding {len(transactions)} transactions to account {account.name} in budget {budget.name}:") - print(f"{'Date':<12}{'Payee':<50}{'Memo':<50}{'Amount':10}{'Status':<9}{'Import ID'}") - for transaction in transactions: - print(f"{transaction.var_date} {transaction.payee_name:<50}{transaction.memo:<50}{transaction.amount/100:<10}{transaction.cleared:<9}{transaction.import_id}") - - confirmed = input("Is this correct? (y/N) ") - if confirmed.lower() != 'y': - print("Stopping processing") - exit(1) - - transaction_api = ynab.TransactionsApi(api_client) - transaction_data = ynab.PostTransactionsWrapper() - transaction_data.transactions = transactions - response = transaction_api.create_transaction(budget.id, transaction_data) - logging.debug(f"Create transaction response: {response}") - - def handle_on_success(file_path, on_success, archive_directory, overwrite): if on_success == 'delete': logging.info(f"Deleting {file_path}") @@ -259,7 +193,7 @@ def convert_bank_statements_to_ynab(input_paths, output_directory, archive_direc if output_type == 'file': write_output_file(bank_name, bank_config, ynab_data, output_directory, overwrite) elif output_type == 'api': - write_to_api(file_path, ynab_data, api_token) + write_to_api(file_path, bank_name, ynab_data, api_token) else: logging.error(f"Unknown output type '{output_type}'") exit(1) diff --git a/src/ynab_api.py b/src/ynab_api.py new file mode 100644 index 0000000..fddcf74 --- /dev/null +++ b/src/ynab_api.py @@ -0,0 +1,124 @@ +import ynab +import logging + +def select_budget(api_client): + budgets_api = ynab.BudgetsApi(api_client) + budgets = budgets_api.get_budgets().data.budgets + + if len(budgets) == 1: + budget = budgets[0] + else: + budget_options = ", ".join( + [ + f"{i}: {budget.name}" + for i, budget in enumerate(budgets) + if 'Archived' not in budget.name + ] + ) + budget_index = input(f"Which budget? ({budget_options}) ") + budget = budgets[int(budget_index)] + + return budget + + +def select_account(api_client, budget): + accounts_api = ynab.AccountsApi(api_client) + accounts = accounts_api.get_accounts(budget.id).data.accounts + + if len(accounts) == 1: + account = accounts[0] + else: + account_options = ", ".join( + [ + f"{i}: {account.name}" + for i, account in enumerate(accounts) + if not account.closed + ] + ) + account_index = input(f"Which account? ({account_options}) ") + account = accounts[int(account_index)] + + return account + + +def add_transactions(api_client, budget, account, data): + transaction_api = ynab.TransactionsApi(api_client) + + transactions = [] + skipped = 0 + amount_factor = 1000 + + start_date = data['Date'].min() + + logging.debug(f"Looking for existing transactions since {start_date}") + existing_transactions = transaction_api.get_transactions_by_account(budget.id, account.id, since_date=start_date).data.transactions + + for i, row in data.iterrows(): + try: + date = row['Date'].date() + import_id = f"IMPORT:{date}:{row['Inflow']}:{row['Outflow']}:{i}" + amount = int((row['Inflow'] - row['Outflow']) * amount_factor) + + for existing in existing_transactions: + skip = False + if existing.import_id == import_id: + skip = True + elif existing.var_date == date and existing.amount == amount and (existing.cleared == 'cleared') == row['Cleared']: + if (row['Payee'].lower() in existing.payee_name.lower()) or (row['Memo'] in existing.memo.lower()): + skip = True + else: + skip = input(f"On {date}, is {row['Payee']} {row['Memo']} the same as {existing.payee_name} {existing.memo} (y/N)").lower() == 'y' + + if skip: + skipped += 1 + break + else: + transaction = ynab.models.new_transaction.NewTransaction( + account_id = account.id, + var_date = date, + amount = amount, + payee_name = row['Payee'], + memo = row['Memo'], + cleared = 'cleared' if row['Cleared'] else 'uncleared', + import_id = import_id if row['Cleared'] else None, + ) + + transactions.append(transaction) + except: + logging.error(f"Could not format transaction {row} for YNAB") + raise + + if skipped > 0: + print(f"Skipped {skipped} existing transactions") + + if len(transactions) == 0: + print("No transactions to import") + return + + print(f"Adding {len(transactions)} transactions to account {account.name} in budget {budget.name}:") + print(f"{'Date':<12}{'Payee':<50}{'Memo':<50}{'Amount':10}{'Status':<11}{'Import ID'}") + for transaction in transactions: + print(f"{transaction.var_date} {transaction.payee_name:<50}{transaction.memo:<50}{transaction.amount/amount_factor:<10}{transaction.cleared:<11}{transaction.import_id}") + + confirmed = input(f"Import transactions to {budget.name} - {account.name}? (y/N) ") + if confirmed.lower() != 'y': + print("Stopping processing") + exit(1) + + transaction_data = ynab.PostTransactionsWrapper() + transaction_data.transactions = transactions + response = transaction_api.create_transaction(budget.id, transaction_data) + + logging.debug(f"Created transactions") + + +def write_to_api(file_path, bank_name, ynab_data, api_token): + configuration = ynab.Configuration( + access_token = api_token + ) + + with ynab.ApiClient(configuration) as api_client: + print(f"Pushing transactions from {bank_name} ({file_path}) to YNAB") + budget = select_budget(api_client) + account = select_account(api_client, budget) + add_transactions(api_client, budget, account, ynab_data)