beancount mint importer

mint.py

#!/usr/bin/env python3

from beancount.core.number import D
from beancount.ingest import importer
from beancount.core.amount import Amount
from beancount.core import flags
from beancount.core import data
from beancount.core.position import Cost

from dateutil.parser import parse

import csv
import os
import re

class MintImporter(importer.ImporterProtocol):
    def __init__(self, account_mapping, category_mapping, category_flag_map):
        self.account_mapping = account_mapping
        self.category_mapping = category_mapping
        self.category_flag_map = category_flag_map
        self.unknown_categories = set()
        self.unknown_accounts = set()

    def identify(self, f):
        return re.match('mint_transactions.*\.csv', os.path.basename(f.name))

    def extract(self, f):
        entries = []

        with open(f.name) as f:
            for index, row in enumerate(csv.reader(f)):
                date, desc, original_desc, amount, transaction_type, category, account, _, notes = row
                if date == "Date":
                    continue
                trans_date = parse(date).date()

                meta = data.new_metadata(f.name, index)

                txn = data.Transaction(
                    meta=meta,
                    date=trans_date,
                    flag=self.category_flag_map.get(category, flags.FLAG_OKAY),
                    payee=original_desc,
                    narration=notes,
                    tags=set(),
                    links=set(),
                    postings=[],
                )

                if account not in self.account_mapping:
                    self.unknown_accounts.add(account)
                if category not in self.category_mapping:
                    self.unknown_categories.add(category)
                try:
                    account = self.account_mapping[account]
                    to_account = self.category_mapping[category]
                except KeyError:
                    continue

                if transaction_type == "debit":
                    counter_amount = amount
                    amount = "-" + amount
                else:
                    counter_amount = "-" + amount

                txn.postings.append(
                    data.Posting(account, Amount(D(amount), 'USD'), None, None, None, None))
                txn.postings.append(
                    data.Posting(to_account, Amount(D(counter_amount), 'USD'), None, None, None, None))

                entries.append(txn)

        if len(self.unknown_categories) + len(self.unknown_accounts) > 0:
            raise LookupError('Unknown categories: {}, accounts: {}'.format(self.unknown_categories, self.unknown_accounts))
        return entries