qifimport.py

def convert_qif(filename, sql_buffer, force_export_all): 
    
    output = []
    output_div = []
    item_count = 1 
    
    file = open(filename, 'r', encoding='utf8', errors='ignore')

    item = {}
    for line in file:
        line = line.replace('\n', ' ').replace('\r', '')
        if line[0].startswith('^'):
            item_count += 1

            item['Flag'] = None

            if item['typ'] == 'Ausgabe': # preping special type Ausgabe / pure QST
                item['Steuern'] = item['Wert']
                item['Wert'] = None
                item['Gebühren'] = None   
                item['Flag'] = "A" # for Ausgabe 
                item['typ'] = "Dividende"

            if item['typ'] == "Dividende":
                output_div.append(copy.deepcopy(item))
            elif "WP-Split" in item['typ']:
                if force_export_all:
                    print("Aktiensplit found:")
                    print(item)
            elif "Payout" in value: #throw away record 
                pass                 
            else:
                output.append(copy.deepcopy(item))

            item = {}
            continue

        #Datum
        elif 'D' in line[0]:
            date = parse(line[1:].rstrip() , dayfirst = False)
            item['Datum'] = date.strftime('%d.%m.%Y')

        #Typ
        elif 'N' in line[0]:
            value = line[1:].rstrip()

            if "DivIn" in value or "DivAus" in value:
                value = 'Dividende'
            elif "WP-Einlage" in value or "Kauf" in value:
                value = 'Einlieferung'
            elif "WP-Entnahme" in value or "Verkauf" in value:
                value = 'Auslieferung'
            elif "Ausgabe" in value: #very few correction booking of QST
                value = 'Ausgabe'

            item['typ'] = value

        #Anzahl
        elif 'Q' in line[0]:
            item['Stück'] = ptok(line[1:].rstrip())

        #Name
        elif 'Y' in line[0]:
            item['Wertpapiername'] = prep_str(ptok(line[1:].rstrip()))

        #Gebühren
        elif 'O' in line[0]:
            fees_sum = 0
            tax_sum = 0
            fee = line[1:].rstrip().split('|')
            
            fees_sum+=float(fee[0])
            tax_sum+=float(fee[1])
            fees_sum+=float(fee[2])
            fees_sum+=float(fee[3])
            fees_sum+=float(fee[4])
            tax_sum+=float(fee[5])
            tax_sum+=float(fee[6])
            fees_sum+=float(fee[7])
            tax_sum+=float(fee[8])
            fees_sum+=float(fee[9])
            tax_sum+=float(fee[10])
            tax_sum+=float(fee[11])

            item['Steuern'] = ptok(str(tax_sum))
            item['Gebühren'] = ptok(str(fees_sum))

        #Wert
        elif 'U' == line[0]:
            item['Wert'] = ptok(line[1:].rstrip())
            #print(item['Wert'])

        elif 'I' in line[0]:
            item['Kurs'] = line[1:].rstrip()
            """ 
            Catch edge case where QIF includes a i.e. 1.130 which could be 1.13 but is actually 1130 
            Hint is the last zero - that's a current bug in FM 2023
            """
            
            if item['Kurs'][-1] == "0" and item['Kurs'].count('.') == 1:
                if len(item['Kurs'].split('.')[1]) == 3:
                    #print(f"Detected last number zero {item['Kurs']}")
                    item['Kurs'] = item['Kurs'].replace('.','',1)

            if item['Kurs'].count('.') >= 2:  # kurs higher than 1000 EUR
                print(f"Warning Kurs > 1000 EUR {item['Wertpapiername']} {item['ISIN']} Kurs {item['Kurs']}")
                item['Kurs'] = item['Kurs'].replace('.','',1)
                #print(f"Corrected Value {item['Kurs']}")

        elif '@' in line[0]:
            item['ISIN'] = line[1:].rstrip()

        elif '~' in line[0]:
            item['WKN'] = line[1:].rstrip()

    account_name = os.path.split(filename)[1].split('.')[0].lower()
    # prepare sql buffer for normal transactions

    print(f"{account_name.upper()} Read {len(output)} Buy/Sell entries")
    for item in output[:]:
        try:
            string_to_hash = "".join([item for item in [item['ISIN'],item['Datum'],item['Stück'],item['typ'],item['Wert']] if item])
        except:
            print("Error with Hash")
            print(item)
        try:
            data=check_duplicate(string_to_hash, account_name)
            check_for_difference_between_qif_and_db(string_to_hash, account_name, item)
        except Exception as e:
            print("Error:")
            print(e)
            print(item)

        if '|' in item['Stück']:
            print("Split found")
            split = item['Stück'].split('|')
            item['Stück'] = split[0]
            item['Kurs'] = None
            split_value = float(split[1].replace(',','.'))

        if data['count'] == 0:
            transaction_date = datetime.datetime.strptime(item['Datum'], "%d.%m.%Y")
            split_value = 0 
            try:
                kurs = Decimal(item['Kurs'].replace(',','.'))
                gebuehren = Decimal(item['Gebühren'].replace(',','.'))
            except: 
                print(item)

            if gebuehren <= 0: 
                gebuehren = None

            steuern = Decimal(item['Steuern'].replace(',','.'))
            if steuern <= 0: 
                steuern = None

            try:
                transaction = PurchaseTransaction(
                            isin=item['ISIN'],
                            type=item['typ'],
                            transaction_date=transaction_date, 
                            transaction_date_year=transaction_date.year,
                            transaction_date_month=transaction_date.month,
                            account = account_name,
                            price=kurs,
                            hash=data['hash'],
                            amount = Decimal(item['Wert'].replace(',','.')),
                            volume = Decimal(item['Stück'].replace(',','.')),
                            dividend_value = None,
                            fees = gebuehren,
                            taxes = steuern
                        )
                sql_buffer.append(transaction)
            except Exception as e:
                print(f"Problem with item {item}")
                print(e)
        else:
            if not force_export_all:
                output.remove(item)
    
    # prepare sql buffer for dividends
    print(f"{account_name.upper()} Read {len(output_div)} Dividend entries")
    for item in output_div[:]:
        try:
            string_to_hash = "".join([item for item in [item['ISIN'],item['Datum'],item['Stück'],item['typ'],item['Wert'],item['Wert']] if item])
            data=check_duplicate(string_to_hash, account_name)
            check_for_difference_between_qif_and_db(string_to_hash, account_name, item)
        except Exception as e:
            print("Error:")
            print(e)
            print(item)

        if item['Wert']:
            dividend_amount = Decimal(item['Wert'].replace(',','.'))
        
        if data['count'] == 0:
            transaction_date = datetime.datetime.strptime(item['Datum'], "%d.%m.%Y")

            split_value = 0 
            kurs = None

            if item['Gebühren']:
                gebuehren = float(item['Gebühren'].replace(',','.'))
                if gebuehren <= 0: 
                    gebuehren = None

            if item['Steuern']:
                steuern = float(item['Steuern'].replace(',','.'))
                if steuern <= 0: 
                    steuern = None

            if item['Stück']:
                volume = float(item['Stück'].replace(',','.'))
                if volume <= 0: 
                    volume = None
    
            try:
                transaction:PurchaseTransaction
                transaction = PurchaseTransaction(
                            isin=item['ISIN'],
                            type=item['typ'],
                            transaction_date=transaction_date, 
                            transaction_date_year=transaction_date.year,
                            transaction_date_month=transaction_date.month,
                            account = account_name,
                            hash=data['hash'],
                            amount = dividend_amount,
                            volume = volume,
                            dividend_value = dividend_amount,
                            fees = None, # no fees for dividends
                            taxes = steuern,
                            flags = item['Flag'], 
                        )

                if 'Flag' in item.keys():
                    if item['Flag']:
                        if 'A' in item['Flag']: #ausgabe
                            if steuern: 
                                transaction.dividend_value_gross = 0
                                transaction.dividend_value = steuern*-1
                            else:
                                transaction.dividend_value_gross = 0
                                transaction.dividend_value = 0                               
                            transaction.amount = 0
                
                    item.pop('Flag', None) # remove flags before export

                sql_buffer.append(transaction)
            except Exception as e:
                print(f"Problem with item {item}")
                print(f"{kurs} / {dividend_amount} / {steuern} / {item['Wert']} / {item['Stück']}")
                print(e)
                raise(e)
        else:
            if not force_export_all:
                output_div.remove(item)

    #nothing to write 
    if len(output) == 0:
        print(f'{account_name.upper()} Buy/Sell Transactions: Nothing to do or only duplicates')

    if len(output_div) == 0:
        print(f'{account_name.upper()} Dividends: Nothing to do or only duplicates')
         
    output_path = configuration.config.pp_output
    if len(output) > 0:
        keys = output[0].keys()
        
        with open(output_path + "/" + account_name + '_transactions.csv', 'w', newline='', encoding='cp1252')  as output_file:
            dict_writer = csv.DictWriter(output_file, keys, delimiter=';')
            dict_writer.writeheader()
            dict_writer.writerows(output)

    if len(output_div) > 0:
        keys = output_div[0].keys()
        with open(output_path + "/" + account_name + '_dividends.csv', 'w', newline='', encoding='cp1252')  as output_file:
            dict_writer = csv.DictWriter(output_file, keys, delimiter=';')
            dict_writer.writeheader()
            dict_writer.writerows(output_div)

    file.close()