commit c11d39b7c432d3ad70afbf1a29fec1331e3eb019 Author: Gerwin van der Lugt Date: Mon Feb 17 09:54:00 2020 +0100 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b97131e --- /dev/null +++ b/.gitignore @@ -0,0 +1,131 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +pip-wheel-metadata/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +.python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +.idea/ diff --git a/data.py b/data.py new file mode 100644 index 0000000..edafb21 --- /dev/null +++ b/data.py @@ -0,0 +1,12 @@ +from collections import namedtuple + +Transaction = namedtuple( + 'Transaction', [ + 'amount', + 'name', + 'iban', + 'description', + 'datetime', + 'after_balance', + 'before_balance' + ]) \ No newline at end of file diff --git a/main.py b/main.py new file mode 100644 index 0000000..0119c24 --- /dev/null +++ b/main.py @@ -0,0 +1,37 @@ +import argparse + +from revolut import RevolutCsvReader +from mt940 import Mt940Writer + + +def main(): + parser = argparse.ArgumentParser( + prog='oddity-revolut-to-mt940', + description='Convert Revolut CSV-files to MT940 format.') + + parser.add_argument('--in', + dest='input_file', + help='path to Revolut csv-file', + required=True) + + parser.add_argument('--account-iban', + dest='account_iban', + help='Revolut account IBAN', + required=True) + + parser.add_argument('--out', + dest='output_file', + help='path to MT940 output path', + required=True) + + args = parser.parse_args() + + reader = RevolutCsvReader(args.input_file) + + with Mt940Writer(args.output_file, args.account_iban) as writer: + for transaction in reader.get_all_transactions(): + writer.write_transaction(transaction) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/mt940.py b/mt940.py new file mode 100644 index 0000000..530403d --- /dev/null +++ b/mt940.py @@ -0,0 +1,193 @@ +from data import Transaction + +BANK_NAME = 'Revolut' +BANK_BIC = 'REVOGB21' + +DEFAULT_SEQUENCE_NO = 1 + + +class Mt940Writer: + + def __init__(self, filename, account_iban): + self.file = open(filename, 'w') + self.account_iban = account_iban + + self._write_header() + self._written_starting_balance = False + self._written_ending_balance = False + + self._balance = None + self._date = None + + + def __enter__(self): + return self + + + def __exit__(self, exc_type, exc_val, exc_tb): + self.release() + + + def write_transaction(self, transaction : Transaction): + if not self._written_starting_balance: + self._write_starting_balance(transaction.datetime, + transaction.before_balance) + + self.file.writelines([ + Mt940.make_61( + transaction.datetime, + transaction.amount), + Mt940.make_86( + transaction.iban, + transaction.name, + transaction.description) + ]) + + self._balance = transaction.after_balance + self._date = transaction.datetime + + + def release(self): + if not self.file.closed \ + and self._written_starting_balance \ + and not self._written_ending_balance: + self._write_ending_balance() + + if not self.file.closed: + self.file.close() + + + def _write_header(self): + self.file.write( + Mt940.make_header(BANK_BIC)) + self.file.writelines([ + Mt940.make_20(BANK_NAME), + Mt940.make_25(self.account_iban, CURRENCY), + Mt940.make_28(DEFAULT_SEQUENCE_NO) + ]) + + + def _write_starting_balance(self, date, balance): + self.file.write( + Mt940.make_60f(date, balance, CURRENCY)) + self._written_starting_balance = True + + + def _write_ending_balance(self): + self.file.write( + Mt940.make_62f(self._date, self._balance, CURRENCY)) + self._written_ending_balance = True + + + +CURRENCY = 'EUR' + +# format identifier +TAG_940 = '940' + +# header +FORMAT_HEADER = \ + '{bic}\n' + \ + TAG_940 + '\n' + \ + '{bic}\n' + +# transaction ref +FORMAT_20 = ':20:{bank}\n' + +# account id +FORMAT_25 = ':25:{iban} {currency}\n' + +# sequence no +FORMAT_28 = ':28:{seqno}\n' + +# opening balance +FORMAT_60F = ':60F:{sign}{date}{currency}{amount}\n' + +# closing balance +FORMAT_62F = ':62F:{sign}{date}{currency}{amount}\n' + +# transaction +FORMAT_61 = ':61:{date}{date2}{amount}{magic}\n' + +# transaction 2 +FORMAT_86 = ':86:/IBAN/{iban}/NAME/{name}/REMI/{description}\n' + +MAGIC = 'NTRFNONREF' + + +class Mt940: + + @staticmethod + def make_header(bic): + return FORMAT_HEADER.format( + bic=bic) + + @staticmethod + def make_20(bank): + return FORMAT_20.format( + bank=bank) + + @staticmethod + def make_25(iban, currency): + return FORMAT_25.format( + iban=iban, + currency=currency) + + @staticmethod + def make_28(seqno): + return FORMAT_28.format( + seqno=Mt940.pad_5(seqno)) + + @staticmethod + def make_60f(datetime, balance, currency): + return FORMAT_60F.format( + sign=Mt940.amount_sign(balance), + date=Mt940.date(datetime), + currency=currency, + amount= Mt940.amount_val(balance)) + + @staticmethod + def make_62f(datetime, balance, currency): + return FORMAT_62F.format( + sign=Mt940.amount_sign(balance), + date=Mt940.date(datetime), + currency=currency, + amount= Mt940.amount_val(balance)) + + @staticmethod + def make_61(datetime, amount): + return FORMAT_61.format( + date=Mt940.date(datetime), + date2=Mt940.date(datetime, with_year=False), + amount=Mt940.amount(amount), + magic=MAGIC) + + @staticmethod + def make_86(iban, name, description): + return FORMAT_86.format( + iban=iban, + name=name, + description=description) + + @staticmethod + def pad_5(val): + return str(val).zfill(5) + + @staticmethod + def amount_sign(val): + return 'C' if val > 0 else 'D' + + @staticmethod + def amount_val(val): + return '{0:.2f}'.format(abs(val)).replace('.', ',') + + @staticmethod + def amount(val): + return Mt940.amount_sign(val) + Mt940.amount_val(val) + + @staticmethod + def date(val, with_year=True): + if with_year: + return val.strftime('%y%m%d') + else: + return val.strftime('%m%d') \ No newline at end of file diff --git a/revolut.py b/revolut.py new file mode 100644 index 0000000..67986b5 --- /dev/null +++ b/revolut.py @@ -0,0 +1,125 @@ +import os +import string +import csv +import math + +from datetime import datetime, timedelta + +from data import Transaction + +EXCPECT_HEADERS = [ + 'Date started', 'Time started', 'Date completed', 'Time completed', + 'State', 'Type', 'Description', 'Reference', 'Payer', 'Card name', + 'Card number', 'Orig currency', 'Orig amount', 'Payment currency', + 'Amount', 'Fee', 'Balance', 'Account', 'Beneficiary account number', + 'Beneficiary sort code or routing number', 'Beneficiary IBAN', + 'Beneficiary BIC' +] + +NAME_REMOVE_PREFIXES = [ + 'Payment from ', + 'To ' +] + +DATE_FORMAT = '%Y-%m-%d' +TIME_FORMAT = '%H:%M:%S' +DATETIME_FORMAT = DATE_FORMAT + TIME_FORMAT + +FEE_NAME = 'Revolut' +FEE_IBAN = '' +FEE_DESCRIPTION = 'Bank transaction fee' +FEE_DATETIME_DELTA = timedelta(seconds=1) + + +class RevolutCsvReader: + + def __init__(self, filename): + if not os.path.isfile(filename): + raise ValueError('File does not exist: {}'.format(filename)) + + self.filename = filename + + self.file = open(self.filename, 'r') + self.reader = csv.reader(self.file) + + self._validate() + + + def __del__(self): + if not self.file.closed: + self.file.close() + + + def _validate(self): + def _santize_header(header): + header = ''.join([c for c in header + if c in string.printable]) + header = header.strip() + return header + + headers = [_santize_header(h) for h in next(self.reader)] + if headers != EXCPECT_HEADERS: + raise ValueError('Headers do not match expected Revolut CSV format.') + + + def get_all_transactions(self): + transactions = [] + for row in self.reader: + transactions = self._parse_transaction(row) + transactions + + return transactions + + + def _parse_transaction(self, row): + + def _santize_name(name_): + for remove_prefix in NAME_REMOVE_PREFIXES: + if name_.startswith(remove_prefix): + name_ = name_[len(remove_prefix):] + + return name_ + + def _parse_datetime(date_str, time_str): + return datetime.strptime(date_str + time_str, DATETIME_FORMAT) + + + _0, _1, completed_date_str, completed_time_str, _4, _5, name, description, _8, _9, _10, \ + _11, _12, _13, amount_str, fee_str, balance_str, _17, _18, _19, iban, _21 \ + = row + + completed_datetime = _parse_datetime(completed_date_str, completed_time_str) + amount, fee, balance = \ + float(amount_str), float(fee_str), float(balance_str) + + transaction_without_fee = Transaction( + amount=amount, + name=_santize_name(name), + iban=iban, + description=description, + datetime=completed_datetime, + before_balance=balance - amount, + after_balance=balance - fee) + + batch = [transaction_without_fee] + + if not math.isclose(fee, 0.00): + fee_transaction = self._make_fee_transaction( + completed_datetime, + balance, + fee) + + batch.append(fee_transaction) + + return batch + + + def _make_fee_transaction(self, completed_datetime, balance, fee): + return Transaction( + amount=fee, + name=FEE_NAME, + iban=FEE_IBAN, + description=FEE_DESCRIPTION, + datetime=completed_datetime + FEE_DATETIME_DELTA, + before_balance=balance - fee, + after_balance=balance) +