Turn into mbank-compatible script

This commit is contained in:
Kuba Orlik 2023-09-03 15:31:36 +02:00
parent e1defce133
commit 9fa7da54dc
4 changed files with 239 additions and 222 deletions

43
main.py
View File

@ -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__":

155
mbank.py Normal file
View File

@ -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,
)

130
mt940.py
View File

@ -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):
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),
self.file.writelines(
[
Mt940.make_61(transaction.datetime, transaction.amount),
Mt940.make_86(
transaction.iban,
transaction.name,
transaction.description)
])
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),
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(DEFAULT_SEQUENCE_NO)
])
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")

View File

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