Initial commit

.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/

data.py

@@ -0,0 +1,12 @@
+from collections import namedtuple
+
+Transaction = namedtuple(
+    'Transaction', [
+        'amount',
+        'name',
+        'iban',
+        'description',
+        'datetime',
+        'after_balance',
+        'before_balance'
+    ])

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()

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')

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)
+