diff --git a/main.py b/main.py index 6010933..012d90a 100644 --- a/main.py +++ b/main.py @@ -1,39 +1,44 @@ import argparse -from revolut import RevolutCsvReader +from mbank import MbankCsvReader from mt940 import Mt940Writer def main(): parser = argparse.ArgumentParser( - prog='oddity-revolut-to-mt940', - description='Convert Revolut CSV-files to MT940 format.') + prog="oddity-mbank-to-mt940", + description="Convert mBank CSV-files to MT940 format.", + ) - parser.add_argument('--in', - dest='input_file', - help='path to Revolut csv-file', - required=True) + 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) + parser.add_argument( + "--out", dest="output_file", help="path to MT940 output path", required=True + ) args = parser.parse_args() - reader = RevolutCsvReader(args.input_file) + reader = MbankCsvReader(args.input_file) - with Mt940Writer(args.output_file, args.account_iban) as writer: + print(reader.range) + with Mt940Writer( + args.output_file, + reader.iban, + reader.range, + reader.starting_balance, + reader.date_start, + ) as writer: transactions = reader.get_all_transactions() for transaction in transactions: writer.write_transaction(transaction) - print('Wrote {} transactions to file: {}.'.format(len(transactions), args.output_file)) + print( + "Wrote {} transactions to file: {}.".format( + len(transactions), args.output_file + ) + ) if __name__ == "__main__": diff --git a/mbank.py b/mbank.py new file mode 100644 index 0000000..d896c06 --- /dev/null +++ b/mbank.py @@ -0,0 +1,155 @@ +import os +import string +import csv +import math + +from datetime import datetime, timedelta + +from data import Transaction + +EXPECT_HEADERS = [ + "#Data księgowania", + "#Data operacji", + "#Opis operacji", + "#Tytuł", + "#Nadawca/Odbiorca", + "#Numer konta", + "#Kwota", + "#Saldo po operacji", +] + +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) + + +def parse_float(s: string): + return float(s.replace(",", ".").replace(" ", "")) + + +class MbankCsvReader: + def __init__(self, file_path): + if not os.path.isfile(file_path): + raise ValueError("File does not exist: {}".format(file_path)) + + temp_file_path = file_path + ".tmp" + + # trimming metadata added my mbank export + with open(file_path, encoding="cp1250") as old, open( + temp_file_path, "w", encoding="utf-8" + ) as new: + lines = old.readlines() + (date_start, date_end, _3) = lines[14].split(";") + iban = "PL" + lines[20].split(";")[0].replace(" ", "") + starting_balance = parse_float(lines[35].split(";")[1].split(" ")[0]) + self.range = date_start + "-" + date_end + self.date_start = datetime.fromisoformat( + "-".join(list(reversed(date_start.split(".")))) + ) + self.date_end = datetime.fromisoformat( + "-".join(list(reversed(date_end.split(".")))) + ) + self.iban = iban + self.starting_balance = starting_balance + new.writelines(lines[37:-1]) + + self.filename = temp_file_path + + self.file = open(self.filename, "r", encoding="utf-8") + self.reader = csv.reader(self.file, delimiter=";") + + self._validate() + + def __del__(self): + if not self.file.closed: + self.file.close() + + def _validate(self): + headers = [h for h in next(self.reader) if h != ""] + print(headers) + print(EXPECT_HEADERS) + if headers != EXPECT_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): + return datetime.strptime(date_str + "7:00:00", DATETIME_FORMAT) + + print(row) + if len(row) < 2 or row[2] == "OTWARCIE RACHUNKU" or row[0] == "": + print("skipping") + return [] + + ( + _0, + completed_date_str, + _1, + description, + name, + iban, + amount_str, + balance_str, + _9, + ) = row + + completed_datetime = _parse_datetime(completed_date_str) + amount, balance = parse_float(amount_str), parse_float(balance_str) + + print(amount, balance) + + transaction_without_fee = Transaction( + amount=amount, + name=_santize_name(name), + iban=iban, + description=description, + datetime=completed_datetime, + before_balance=balance - amount, + after_balance=balance, + ) + + batch = [transaction_without_fee] + + # no support for transaction fees + # 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, + ) diff --git a/mt940.py b/mt940.py index 93bc0b5..62c11c3 100644 --- a/mt940.py +++ b/mt940.py @@ -1,16 +1,18 @@ from data import Transaction -BANK_NAME = 'Revolut' -BANK_BIC = 'REVOLT21' +BANK_NAME = "mBank" +BANK_BIC = "BREXPLPWMBK" DEFAULT_SEQUENCE_NO = 1 class Mt940Writer: - - def __init__(self, filename, account_iban): - self.file = open(filename, 'w') + def __init__(self, filename, account_iban, range, starting_balance, date_start): + self.file = open(filename, "w") self.account_iban = account_iban + self.range = range + self.starting_balance = starting_balance + self.date_start = date_start self._write_header() self._written_starting_balance = False @@ -19,124 +21,106 @@ class Mt940Writer: 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): + def write_transaction(self, transaction: Transaction): if not self._written_starting_balance: - self._write_starting_balance(transaction.datetime, - transaction.before_balance) + self._write_starting_balance(self.starting_balance) - self.file.writelines([ - Mt940.make_61( - transaction.datetime, - transaction.amount), - Mt940.make_86( - transaction.iban, - transaction.name, - transaction.description) - ]) + 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: + 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) - ]) + self.file.write(Mt940.make_header(BANK_BIC)) + self.file.writelines( + [ + Mt940.make_20(BANK_NAME, self.range), + Mt940.make_25(self.account_iban, CURRENCY), + Mt940.make_28(self.range), + ] + ) - - def _write_starting_balance(self, date, balance): - self.file.write( - Mt940.make_60f(date, balance, CURRENCY)) + def _write_starting_balance(self, balance): + self.file.write(Mt940.make_60f(self.date_start, balance, CURRENCY)) self._written_starting_balance = True - def _write_ending_balance(self): - self.file.write( - Mt940.make_62f(self._date, self._balance, CURRENCY)) + self.file.write(Mt940.make_62f(self._date, self._balance, CURRENCY)) self._written_ending_balance = True - -CURRENCY = 'EUR' +CURRENCY = "PLN" # format identifier -TAG_940 = '940' +TAG_940 = "940" # header -FORMAT_HEADER = \ - '{bic}\n' + \ - TAG_940 + '\n' + \ - '{bic}\n' +FORMAT_HEADER = "{bic}\n" + TAG_940 + "\n" + "{bic}\n" # transaction ref -FORMAT_20 = ':20:{bank}\n' +FORMAT_20 = ":20:{bank}{range}\n" # account id -FORMAT_25 = ':25:{iban} {currency}\n' +FORMAT_25 = ":25:{iban} {currency}\n" # sequence no -FORMAT_28 = ':28:{seqno}\n' +FORMAT_28 = ":28:{seqno}\n" # opening balance -FORMAT_60F = ':60F:{sign}{date}{currency}{amount}\n' +FORMAT_60F = ":60F:{sign}{date}{currency}{amount}\n" # closing balance -FORMAT_62F = ':62F:{sign}{date}{currency}{amount}\n' +FORMAT_62F = ":62F:{sign}{date}{currency}{amount}\n" # transaction -FORMAT_61 = ':61:{date}{date2}{amount}{magic}\n' +FORMAT_61 = ":61:{date}{date2}{amount}{magic}\n" # transaction 2 -FORMAT_86 = ':86:/IBAN/{iban}/NAME/{name}/REMI/{description}\n' +FORMAT_86 = ":86:/IBAN/{iban}/NAME/{name}/REMI/{description}\n" -MAGIC = 'NTRFNONREF' +MAGIC = "NTRFNONREF" class Mt940: - @staticmethod def make_header(bic): - return FORMAT_HEADER.format( - bic=bic) + return FORMAT_HEADER.format(bic=bic) @staticmethod - def make_20(bank): - return FORMAT_20.format( - bank=bank) + def make_20(bank, range): + return FORMAT_20.format(bank=bank, range=range) @staticmethod def make_25(iban, currency): - return FORMAT_25.format( - iban=iban, - currency=currency) + return FORMAT_25.format(iban=iban, currency=currency) @staticmethod def make_28(seqno): - return FORMAT_28.format( - seqno=Mt940.pad_5(seqno)) + return FORMAT_28.format(seqno=Mt940.pad_5(seqno)) @staticmethod def make_60f(datetime, balance, currency): @@ -144,7 +128,8 @@ class Mt940: sign=Mt940.amount_sign(balance), date=Mt940.date(datetime), currency=currency, - amount= Mt940.amount_val(balance)) + amount=Mt940.amount_val(balance), + ) @staticmethod def make_62f(datetime, balance, currency): @@ -152,7 +137,8 @@ class Mt940: sign=Mt940.amount_sign(balance), date=Mt940.date(datetime), currency=currency, - amount= Mt940.amount_val(balance)) + amount=Mt940.amount_val(balance), + ) @staticmethod def make_61(datetime, amount): @@ -160,14 +146,12 @@ class Mt940: date=Mt940.date(datetime), date2=Mt940.date(datetime, with_year=False), amount=Mt940.amount(amount), - magic=MAGIC) + magic=MAGIC, + ) @staticmethod def make_86(iban, name, description): - return FORMAT_86.format( - iban=iban, - name=name, - description=description) + return FORMAT_86.format(iban=iban, name=name, description=description) @staticmethod def pad_5(val): @@ -175,11 +159,11 @@ class Mt940: @staticmethod def amount_sign(val): - return 'C' if val > 0 else 'D' + return "C" if val > 0 else "D" @staticmethod def amount_val(val): - return '{0:.2f}'.format(abs(val)).replace('.', ',') + return "{0:.2f}".format(abs(val)).replace(".", ",") @staticmethod def amount(val): @@ -188,6 +172,6 @@ class Mt940: @staticmethod def date(val, with_year=True): if with_year: - return val.strftime('%y%m%d') + return val.strftime("%y%m%d") else: - return val.strftime('%m%d') + return val.strftime("%m%d") diff --git a/revolut.py b/revolut.py deleted file mode 100644 index 07dbde8..0000000 --- a/revolut.py +++ /dev/null @@ -1,127 +0,0 @@ -import os -import string -import csv -import math - -from datetime import datetime, timedelta - -from data import Transaction - -EXCPECT_HEADERS = [ - 'Date started (UTC)', 'Time started (UTC)', 'Date completed (UTC)', - 'Time completed (UTC)', '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) -