ofx-processor/ofx_processor/processors/revolut.py

import csv

import click
import dateparser

from ofx_processor.utils.processor import Processor, Line


def _amount_str_to_float(amount):
    if amount:
        return float(amount.replace(",", "."))
    return ""


class RevolutLine(Line):
    def __init__(self, data: dict = None):
        super(RevolutLine, self).__init__(data)

    def _process_inflow(self):
        return _amount_str_to_float(self.data.get("Paid In (EUR)"))

    def _process_outflow(self):
        return _amount_str_to_float(self.data.get("Paid Out (EUR)"))

    def get_date(self):
        return dateparser.parse(self.data.get("Completed Date")).strftime("%Y-%m-%d")

    def get_amount(self):
        outflow = self._process_outflow()
        inflow = self._process_inflow()
        amount = -outflow if outflow else inflow
        amount = int(amount * 1000)
        return amount

    def get_memo(self):
        return " - ".join(
            filter(
                None,
                map(
                    str.strip,
                    [self.data.get("Category", ""), self.data.get("Exchange Rate", "")],
                ),
            )
        )

    def get_payee(self):
        return self.data.get("Reference")


class RevolutProcessor(Processor):
    line_class = RevolutLine
    account_name = "revolut"

    def parse_file(self):
        with open(self.filename) as f:
            reader = csv.DictReader(f, delimiter=";")
            return [line for line in reader]

    @staticmethod
    @click.command("revolut", help="Process Revolut bank statement (CSV)")
    @click.argument("csv_filename")
    def main(csv_filename):
        RevolutProcessor(csv_filename).push_to_ynab()