Commit f5ab200c authored by seth's avatar seth 😄


parent 09712259
Beancount importer for CSV-CAMT files
__copyright__ = "Copyright (C) 2016 Martin Blais"
__license__ = "GNU GPLv2"
import csv
import codecs
import datetime
import re
import pprint
import logging
from os import path
from dateutil.parser import parse
from beancount.core.number import D
from beancount.core.number import ZERO
from beancount.core.number import MISSING
from beancount.core import data
from beancount.core import account
from beancount.core import amount
from beancount.core import position
from beancount.core import inventory
from beancount.ingest import importer
from beancount.ingest import regression
class Importer(importer.ImporterProtocol):
"""An importer for CAMT-CSV files."""
def __init__(self,
self.iban = iban
self.currency = currency
self.account = account
def identify(self, file):
# Match if the filename is as downloaded and the header has the unique
# fields combination we're looking for.
return (re.match(r"\d{8}-\d{7,8}-umsatz.CSV", path.basename(
and re.match(".*Auftragskonto.*", file.head()))
def file_name(self, file):
return 'sparkasse.{}'.format(path.basename(
def file_account(self, _):
return self.account
def file_date(self, file):
# Extract the statement date from the filename.
date_str = re.match(r"20\d{6}", path.basename(
return datetime.datetime.strptime(date_str, f'%Y%m%d').date()
def extract(self, file):
# Open the CSV file and create directives.
entries = []
index = 0
print("File: ",, "\n")
csv_file =, 'r', "iso-8859-1")
for index, row in enumerate(csv.DictReader(csv_file, delimiter=';')):
print("\nEntry no", index)
meta = data.new_metadata(, index)
print("Metadata: ", meta)
iban = row['Auftragskonto']
print("Auftragskonto: ", iban)
date = datetime.datetime.strptime(row['Buchungstag'], f"%d.%m.%y").date()
print("Buchungstag: ", date)
rtype = row['Buchungstext']
print("Buchungstext: ", rtype)
desc = row['Verwendungszweck']
print("Verwendungszweck: ", desc)
payee = row['Beguenstigter/Zahlungspflichtiger']
print("Begünstigter: ", payee)
units_de = row['Betrag']
units_en = units_de.translate(units_de.maketrans(',.', '.,'))
units = amount.Amount(D(units_en), self.currency)
print("Betrag: ", units)
txn = data.Transaction(
meta, date, self.FLAG, payee, desc, data.EMPTY_SET, data.EMPTY_SET, [
data.Posting(self.account, units, None, None, None, None),
data.Posting("Expenses:Todo", -units, None, None, None, None)
return entries
Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment