Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/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
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement