sparebanken-sor-ynab/src/ynab_api.py

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)