mbank-to-mt940/revolut.py

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_FORMAT = '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 - fee,
            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,
            # include timestamp of transaction to make sure that SnelStart
            # does not detect similar transactions as the same one
            description=FEE_DESCRIPTION_FORMAT.format(int(completed_datetime.timestamp())),
            datetime=completed_datetime + FEE_DATETIME_DELTA,
            before_balance=balance - fee,
            after_balance=balance)