skrooge-common 2.11.0-1build2 / usr / share / skrooge / skrooge-sabb.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#**************************************************************************
#*   Copyright (C) 2017 by Bernhard Scheirle bernhard@scheirle.de
#*   This program is free software: you can redistribute it and/or modify
#*   it under the terms of the GNU General Public License as published by
#*   the Free Software Foundation, either version 3 of the License, or
#*   (at your option) any later version.
#*   
#*   This program is distributed in the hope that it will be useful,
#*   but WITHOUT ANY WARRANTY; without even the implied warranty of
#*   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#*   GNU General Public License for more details.
#*   
#*   You should have received a copy of the GNU General Public License
#*   along with this program.  If not, see <http://www.gnu.org/licenses/>
#**************************************************************************

"""
Skrooge AqBanking Bridge (SABB)
-------------------------------

Authors: 
 * Bernhard Scheirle <bernhard@scheirle.de>
""" 

import argparse
import contextlib
import csv
import datetime
import io
import os
import re
import shutil
import subprocess
import sys
import tempfile
from distutils.version import LooseVersion

__VERSION__ = "1.0.0"

class Account(object):
    def __init__(self):
        self.bank_number    = ""
        self.account_number = ""
        self.iban           = ""

    def toString(self):
        return 'Account(' + self.bank_number + ', ' + self.account_number + ', ' + self.iban + ')'

    def isValid(self):
        return self.bank_number != "" and self.account_number != "" and self.iban != ""


class Accounts(object):
    def __init__(self):
        self._account_map = {}

    def account_map_key(self, bank_number, account_number):
        return str(bank_number) + '.' + str(account_number)

    def _build_iban_map(self):
        self._account_map = {}

        regex_account_number = re.compile('accountNumber=\"(.*)\"')
        regex_bank_number    = re.compile('bankCode=\"(.*)\"')
        regex_iban           = re.compile('iban=\"(.*)\"')
        #regex_bic            = re.compile('bic=\"(.*)\"')

        accounts_folder = os.path.expanduser('~/.aqbanking/settings/accounts/')
        for file_name in os.listdir(accounts_folder):
            if not file_name.endswith('.conf'):
                continue
            file_path = os.path.join(accounts_folder, file_name)
            with open(file_path, 'r') as account_file:
                content = account_file.read()
                account = Account()
                match = regex_account_number.search(content)
                if match:
                    account.account_number = match.group(1)
                match = regex_bank_number.search(content)
                if match:
                    account.bank_number = match.group(1)
                match = regex_iban.search(content)
                if match:
                    account.iban = match.group(1)
                #  match = regex_bic.search(content)
                #  if match:
                #    account.bic = match.group(1)
                #  print(account.toString())
                if account.isValid():
                    key = self.account_map_key(account.bank_number, account.account_number)
                    self._account_map[key] = account

    def get_account(self, bank_number, account_number):
        if not self._account_map:
            self._build_iban_map()

        key = self.account_map_key(bank_number, account_number)
        if key in self._account_map:
            return self._account_map[key]
        else:
            return Account()

    def get_accounts(self):
        if not self._account_map:
            self._build_iban_map()
        return self._account_map.values()


class AqDialect(csv.Dialect):
    delimiter = ';'
    quotechar = '"'
    quoting = csv.QUOTE_NONE
    lineterminator = '\n'


class AqDialectTransfairs(csv.Dialect):
    delimiter = ';'
    quotechar = '"'
    quoting = csv.QUOTE_ALL
    lineterminator = '\n'


class SkroogeDialect(csv.Dialect):
    delimiter = ';'
    quotechar = '"'
    escapechar = '\\'
    quoting = csv.QUOTE_ALL
    lineterminator = '\n'


@contextlib.contextmanager
def TemporaryContextFile():
    with tempfile.TemporaryDirectory(prefix="SSAB.") as dir_path:
        context_file_path = os.path.join(dir_path, "context")
        open(context_file_path, 'a').close()
        yield context_file_path


class SABB(object):
    # Tools
    AqBanking = 'aqbanking-cli'
    AqHBCI    = 'aqhbci-tool4'

    ReturnValue_NormalExit = 0
    ReturnValue_InvalidVersion = 1
    
    def __init__(self):
        self.accounts = Accounts()

    def build_command(self, executable, args):
        com = [executable]
        if executable is self.AqBanking:
            com.append('--charset=utf-8')
        com.extend(args)
        return com

    def get_csv_reader(self, process_result, fieldnames, dialect=AqDialect):
        input = process_result.stdout.decode("utf-8").replace('\t', ';')
        reader = csv.DictReader(input.splitlines(), fieldnames=fieldnames, dialect=dialect)
        return reader

    def get_csv_writer(self, fieldnames, generateHeader=True):
        output = io.StringIO("")
        writer = csv.DictWriter(output, fieldnames=fieldnames, dialect=SkroogeDialect)
        if generateHeader:
            writer.writeheader()
        return output, writer

    def format_iban(self, iban):
        result = ""
        for i, c in enumerate(iban):
            if i % 4 == 0:
                result = result + " "
            result = result + c
        return result.strip()

    def get_iban(self, bank_number, account_number):
        account = self.accounts.get_account(bank_number, account_number)
        if account.isValid():
            return self.format_iban(account.iban.upper())
        else:
            return ""

    def check_version(self):
        process_result = subprocess.run(
            self.build_command(self.AqBanking, ['versions']),
            stdout=subprocess.PIPE
        )
        process_result.check_returncode()
        lines = process_result.stdout.decode("utf-8").splitlines()
        valid_version = False
        for line in lines:
            line = line.strip()
            if line.startswith("AqBanking-CLI:"):
                line = line[14:].strip()
                if LooseVersion(line) >= LooseVersion("5.6.10"):
                    valid_version = True
                    break
        if not valid_version:
            print("Please install a newer version of aqbanking. At least version 5.6.10 is requiered.")
        return valid_version

    def get_accounts(self):
        if not self.check_version():
            return self.ReturnValue_InvalidVersion
        process_result = subprocess.run(
            self.build_command(self.AqBanking, ['listaccs']),
            stdout=subprocess.PIPE
        )
        process_result.check_returncode()
        fieldnames_input  = ['ignore','bank_number','account_number','bank_name','account_name']
        fieldnames_output = ['id']
        reader = self.get_csv_reader(process_result, fieldnames_input)
        output, writer = self.get_csv_writer(fieldnames_output)
        accounts = []
        for record in reader:
            row = {}
            account_id = self.get_iban(record['bank_number'], record['account_number'])
            if account_id == "" or account_id in accounts:
                # Filter out duplicates and empty ids
                continue
            row['id'] = account_id
            accounts.append(account_id)
            writer.writerow(row)
        print(output.getvalue().strip())
        return self.ReturnValue_NormalExit

    def write_balance_file(self, output_folder_path, context_file_path):
        process_result = subprocess.run(
            self.build_command(self.AqBanking,
                                ['listbal',
                                '-c', context_file_path]),
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL
        )
        fieldnames_input  = ['ignore', 'bank_number', 'account_number', 'bank_name', 'account_name',
                                'booked_date', 'booked_time', 'booked_value', 'booked_currency',
                                'noted_date',  'noted_time',  'noted_value',  'noted_currency']
        fieldnames_output = ['id', 'date', 'balance', 'unit']
        reader = self.get_csv_reader(process_result, fieldnames_input)
        output, writer = self.get_csv_writer(fieldnames_output)
        for record in reader:
            row = {}
            row['id'] = self.get_iban(record['bank_number'], record['account_number'])
            try:
                row['balance'] = record['booked_value']
                row['unit']    = record['booked_currency']
                row['date']    = datetime.datetime.strptime(record['booked_date'], "%d.%m.%Y").strftime("%Y-%m-%d")
            except:
                try:
                    row['balance'] = record['noted_value']
                    row['unit']    = record['noted_currency']
                    row['date']    = datetime.datetime.strptime(record['noted_date'], "%d.%m.%Y").strftime("%Y-%m-%d")
                except:
                    continue
            writer.writerow(row)
        output_file_path = os.path.join(output_folder_path, "balance.csv")
        with open(output_file_path, 'w') as f:
            f.write(output.getvalue())

    def convert_transactions(self, aqbanking_output, generateHeader=True):
        reader = csv.DictReader(aqbanking_output.splitlines(), dialect=AqDialectTransfairs)
        fieldnames_output = ['date', 'mode', 'comment', 'payee', 'amount', 'unit']
        output, writer = self.get_csv_writer(fieldnames_output, generateHeader)
        for record in reader:
            row = {}
            row['date'] = record['date']
            row['mode'] = record['purpose']
            comment = ''
            for i in range(1, 12):
                comment = comment + record['purpose' + str(i)] + " "
            row['comment'] = comment.strip()
            row['payee']   = (record['remoteName'] + " " + record['remoteName1']).strip()
            row['amount']  = record['value_value']
            row['unit']    = record['value_currency']
            writer.writerow(row)
        return output.getvalue()

    def download(self, output_folder_path, balance, openTerminal):
        if not self.check_version():
            return self.ReturnValue_InvalidVersion
        with TemporaryContextFile() as context_file_path:
            args = ['request',
                    '--ignoreUnsupported',
                    '--transactions',
                    '-c',
                    context_file_path
                    ]
            if balance:
                args.append('--balance')
            
            command = self.build_command(self.AqBanking, args)
            if openTerminal:
                shell_command = ['x-terminal-emulator', '-e']
                shell_command.extend(command)
                command = shell_command
            process = subprocess.Popen(command)
            process.wait()
            

            output_folder_path = os.path.abspath(output_folder_path)
            if not os.path.exists(output_folder_path):
                os.makedirs(output_folder_path)
            files = {}
            for account in self.accounts.get_accounts():
                process_result = subprocess.run(
                    self.build_command(self.AqBanking,
                                       ['listtrans',
                                        '--bank=' + account.bank_number,
                                        '--account=' + account.account_number,
                                        '-c',
                                        context_file_path
                                        ]),
                    stdout=subprocess.PIPE,
                    stderr=subprocess.DEVNULL
                )
                transactions = process_result.stdout.decode("utf-8")
                output_file_path = os.path.join(output_folder_path, self.format_iban(account.iban.upper()) + ".csv")
                if output_file_path in files:
                    files[output_file_path] = files[output_file_path] + '\n' + self.convert_transactions(transactions, False)
                else:
                    files[output_file_path] = self.convert_transactions(transactions, True)

            for path, content in files.items():
                with open(path, 'w') as f:
                    f.write(content)

            if balance:
                self.write_balance_file(output_folder_path, context_file_path)
        return self.ReturnValue_NormalExit


def main():
    parser = argparse.ArgumentParser(prog='sabb', description='Skrooge AqBanking Bridge (SABB)')

    # Global arguments
    parser.add_argument('--version', '-v', action='version', version='%(prog)s ' + __VERSION__, help='Shows version information.')
    subparsers = parser.add_subparsers(title='Commands', dest='command')

    # Command: listaccounts
    parser_listaccounts = subparsers.add_parser('listaccounts', help='Returns a list of accounts that can be queried with AqBanking.')

    # Command: bulkdownload
    parser_download = subparsers.add_parser('bulkdownload',         help='Downloads all transactions into the given output folder')
    parser_download.add_argument('--output_folder', required=True,  help='The folder to store the csv files.')
    parser_download.add_argument('--balance',       required=False, action='store_true',
        help='Additionally also download the current balance of all accounts and stores it in a "balance.csv" file in the output folder.')

    args = parser.parse_args()

    if (args.command == "listaccounts"):
        return SABB().get_accounts()
    elif (args.command == "bulkdownload"):
        return SABB().download(args.output_folder, args.balance, True)
    else:
        parser.print_help()
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())