Custom YNAV4 to Firefly III import

import-ynab4-into-firefly-iii.py

import csv
import re
from operator import itemgetter, methodcaller

from api import FireflyAPI

token = 'TOKEN'
account_map = {}
investment_change_opposite = 'OPPOSING ACCOUNT'
investment_change_payees = []


def parse_transaction(t):
    def map_account(acc):
        return account_map.get(acc, acc)

    transaction = {
        'type': None,
        'date': '{y}-{m}-{d}'.format(**re.match(r'(?P<d>\d+)\.(?P<m>\d+)\.(?P<y>\d+)', t['Date']).groupdict()),
        'amount': max(map(lambda x: abs(float(x[:-1].replace(',', '.'))), [t['Outflow'], t['Inflow']])),
        'description': t['Payee'],
        'notes': t['Memo'],
        'tags': ['YNAB import'],
        'reconciled': t['Cleared'] == 'R'
    }
    if t['Payee'].startswith('Transfer : '):
        transaction['type'] = 'transfer'
        transaction['source_name'] = map_account(t['Account'])
        transaction['destination_name'] = map_account(t['Payee'][11:])
        transaction['description'] = t['Memo'] if t['Memo'] != "" else "Transfer"
    elif t['Payee'] in investment_change_payees:
        transaction['type'] = 'transfer'
        transaction['description'] = 'Kursgewinn/verlust'
        if t['Outflow'] != '0,00€':
            transaction['source_name'] = map_account(t['Account'])
            transaction['destination_name'] = investment_change_opposite
        elif t['Inflow'] != '0,00€':
            transaction['destination_name'] = map_account(t['Account'])
            transaction['source_name'] = investment_change_opposite
        else:
            transaction['type'] = None
    elif t['Outflow'] != '0,00€':
        transaction['type'] = 'withdrawal'
        transaction['source_name'] = map_account(t['Account'])
        transaction['destination_name'] = t['Payee']
        transaction['category_name'] = t['Category']
    elif t['Inflow'] != '0,00€':
        transaction['type'] = 'deposit'
        transaction['destination_name'] = map_account(t['Account'])
        transaction['source_name'] = t['Payee']
    return transaction


if __name__ == '__main__':
    # Load Data
    with open('input.csv', encoding='utf-8-sig') as csvfile:
        transactions = list(csv.DictReader(csvfile, delimiter='\t'))
    accounts = list(set(map(itemgetter('Account'), transactions)))

    for t in transactions:
        # Rename Payees that have same name as account so that the automapper doesnt get confused
        if t['Payee'].lower() in map(methodcaller('lower'), accounts):
            t['Payee'] += " Bank"

        # Remove Payee from split Payee/Transfer Transactions
        if '/ Transfer :' in t['Payee']:
            t['Payee'] = 'Transfer :' + t['Payee'].split('/ Transfer :')[1]

    # Remove one side of every Transfer (Only keep those with Outflow)
    transactions = list(filter(lambda t: not t['Payee'].startswith('Transfer : ') or t['Outflow'] != '0,00€',
                               transactions))

    # Parse Transactions into dict for API
    transactions = list(map(parse_transaction, transactions))

    # filter out invalid transactions (e.g. In=Out=0,00€)
    transactions = list(filter(lambda x: x['type'] is not None, transactions))

    # map accounts
    for t in transactions:
        t['source_name'] = account_map.get(t['source_name'], t['source_name'])
        if 'destination_name' in t:
            t['destination_name'] = account_map.get(t['destination_name'], t['destination_name'])

    # check if asset accounts exist
    api = FireflyAPI('FIREFLYURL', token)
    accounts = api.request('/api/v1/accounts?type=asset')
    accounts = list(map(itemgetter('name'), map(itemgetter('attributes'), accounts)))

    for t in transactions:
        if t['type'] in ['withdrawal', 'transfer'] and t['source_name'] not in accounts:
            raise Exception('Account {} does not exist'.format(t['source_name']))
        if t['type'] in ['deposit', 'transfer'] and t['destination_name'] not in accounts:
            raise Exception('Account {} does not exist'.format(t['destination_name']))

    l = len(transactions)
    for i, t in enumerate(transactions):
        data = {
            'transactions': [t]
        }
        print('{}% {}/{}'.format((i + 1) * 100 // l, i + 1, l))
        api.request('/api/v1/transactions', method='POST', data=data)