Importador Banco Inter - Extratos e Faturas CSV (Beancount Ingest)

importer.py

import csv
from datetime import datetime, date, timedelta
from decimal import Decimal
from os import path

from beancount.core import amount, data
from beancount.core.amount import Amount
from beancount.ingest.importer import ImporterProtocol


def brl_to_decimal(brl: str) -> Decimal:
    brl = brl.replace('.', '').replace(',', '.').replace('R$', '').replace(' ', '')
    return Decimal(brl)


class ExtractImporter(ImporterProtocol):
    """
    ExtratImporter for Pagarme CSV Transactions
                        "BRL",
                        "Assets:Caixa:Inter",
                        "Income:PythonPro...Ditribuicacao",
                        "Income:PythonPro...ProLabore",
                        "Expenses:Casa",
    """

    def __init__(self, currency,
                 asset_inter,
                 income_pythonpro_distribuicao,
                 income_pythonpro_prolabore,
                 expenses_casa,

                 ):

        self.expenses_casa = expenses_casa
        self.income_pythonpro_prolabore = income_pythonpro_prolabore
        self.income_pythonpro_distribuicao = income_pythonpro_distribuicao
        self.currency = currency
        self.asset_inter = asset_inter

    def name(self):
        raise NotImplementedError()

    def identify(self, file):
        raise NotImplementedError()

    def extract(self, file, existing_entries=None):
        # Open the CSV file and create directives.
        entries = []
        index = 0
        meta = data.new_metadata(file.name, index)
        balance = Decimal(0)
        entry_date = date.today()
        pro_labore_value = 1100
        with open(file.name) as file_reader:
            # ignore first 5 lines
            last_line = None
            for _ in range(5):
                last_line = next(file_reader)
            is_old_extract = last_line.strip() != ''
            if is_old_extract:
                next(file_reader)
                next(file_reader)
            for index, row in enumerate(csv.DictReader(file_reader, delimiter=';')):
                meta = data.new_metadata(file.name, index)
                entry_date = datetime.strptime(row.get('DATA LANÇAMENTO',row.get('Data Lançamento')), '%d/%m/%Y').date()

                entry_value = brl_to_decimal(row.get('VALOR', row.get('Valor')))
                balance = brl_to_decimal(row.get('SALDO', row.get('Saldo')))
                description = row.get('HISTÓRICO', row.get('Descrição'))
                if 'PYTHON PRO TREINAMENTO' in description:
                    destination = self.income_pythonpro_distribuicao
                    if entry_value == Decimal(-pro_labore_value):
                        destination = self.income_pythonpro_prolabore
                    txn = data.Transaction(

                        meta, entry_date, self.FLAG, None, description,

                        data.EMPTY_SET,
                        data.EMPTY_SET, [
                            data.Posting(self.asset_inter, self.to_amount(entry_value), None, None, None, None),
                            data.Posting(destination, self.to_amount(-entry_value), None,
                                         None, None, None),
                        ])
                else:
                    txn = data.Transaction(
                        meta, entry_date, self.FLAG, None, description,
                        data.EMPTY_SET,
                        data.EMPTY_SET, [
                            data.Posting(self.asset_inter, self.to_amount(entry_value), None, None, None, None),
                            data.Posting('Expenses:Unknown', self.to_amount(-entry_value), None, None, None, None),
                        ])
                entries.append(txn)

            # Insert a final balance check.
            if entries:
                last_date = entry_date + timedelta(days=1)
                entries.append(
                    data.Balance(meta, last_date,
                                 self.asset_inter,
                                 amount.Amount(balance, self.currency),
                                 None, None))

            return entries

    def file_account(self, file):
        return self.asset_inter

    def file_name(self, file):
        raise NotImplementedError()

    def file_date(self, file):
        # Extract the statement from file.
        with open(file.name) as f:
            # discard first two lines
            next(f)
            next(f)
            # data line Ex:Período ;20/07/2020;02/08/2020; ou ;18/09/2021 a 23/09/2021
            date_line = next(f)
            if 'a' in date_line:
                date_str = date_line.split()[-1]
            else:
                date_str = date_line.split(';')[-2]

        return datetime.strptime(date_str, '%d/%m/%Y').date()

    def to_amount(self, total_income: Decimal):
        return Amount(total_income, self.currency)


class ExtractImporterRenzo(ExtractImporter):
    def name(self):
        return 'InterRenzo'

    def identify(self, file):
        # Match if the filename is as downloaded
        return path.basename(file.name) == 'ExtratoRenzo.csv'

    def file_name(self, file):
        dt = self.file_date(file)
        return f'Inter_Renzo_extrato_{dt.year}_{dt.month:02d}_{dt.day:02d}.csv'

class ExtractImporterAmanda(ExtractImporter):
    def name(self):
        return 'InterAmanda'

    def identify(self, file):
        # Match if the filename is as downloaded
        return path.basename(file.name) == 'ExtratoAmanda.csv'

    def file_name(self, file):
        dt = self.file_date(file)
        return f'Inter_Amanda_extrato_{dt.year}_{dt.month:02d}_{dt.day:02d}.csv'


class ExtractImporterPriscila(ExtractImporter):
    def name(self):
        return 'InterPriscila'

    def identify(self, file):
        # Match if the filename is as downloaded
        return path.basename(file.name) == 'ExtratoPriscila.csv'

    def file_name(self, file):
        dt = self.file_date(file)
        return f'Inter_Priscila_extrato_{dt.year}_{dt.month:02d}_{dt.day:02d}.csv'


class FaturaImporter(ImporterProtocol):
    """
    FaturaImporter for Inter Credit Card CSV Transactions
                        "BRL",
                        "Liabilities:Cartoes:Inter:Priscila ",
                        "Expenses:Unknown",
    """

    def __init__(self, currency,
                 liability_credit_card,
                 expenses_account='Expenses:Unknown',
                 ):

        self.expenses_account = expenses_account
        self.currency = currency
        self.liability_credit_card = liability_credit_card

    def name(self):
        raise NotImplementedError()

    def identify(self, file):
        raise NotImplementedError()

    def extract(self, file, existing_entries=None):
        # Open the CSV file and create directives.
        entries = []
        index = 0
        meta = data.new_metadata(file.name, index)
        entry_date = date.today()
        with open(file.name, encoding='utf-8-sig') as file_reader:
            print(f'#### Reading: {file.name}')
            # ignore first lines
            # for _ in range(4):
            #     next(file_reader)
            # fifth_line = next(file_reader)
            # if fifth_line.startswith('Vencimento'):
            #     # Ignore next two line
            #     next(file_reader)
            #     next(file_reader)
            for index, row in enumerate(csv.DictReader(file_reader, delimiter=',')):
                print(f'#### LINE {index}: {row}')
                meta = data.new_metadata(file.name, index)
                # "Data","Lançamento","Categoria","Tipo","Valor"
                # raise Exception(str(row.keys()))
                desc = row['Lançamento']
                if desc== 'PAGAMENTO ON LINE':
                    # pagamentos serão lançados no extrato
                    continue
                entry_date = datetime.strptime(row['Data'], '%d/%m/%Y').date()
                entry_value = brl_to_decimal(row['Valor'])
                store = row['Categoria']

                transaction_type = row['Tipo'].replace('"', '')
                if transaction_type in {'Compra à vista', ''}:
                    description = f'{store} - {transaction_type} - {desc}'
                    txn = data.Transaction(
                        meta, entry_date, self.FLAG, None, description,
                        data.EMPTY_SET,
                        data.EMPTY_SET, [
                            data.Posting(self.liability_credit_card, self.to_amount(-entry_value), None, None, None,
                                         None),
                            data.Posting(self.expenses_account, self.to_amount(entry_value), None, None, None, None),
                        ])
                elif transaction_type.startswith('Parcela 1/'):
                    total_installments = int(transaction_type.split('/')[1].strip())
                    description = f'{store} - {total_installments} parcelas de {entry_value}'
                    entry_value *= total_installments
                    txn = data.Transaction(
                        meta, entry_date, self.FLAG, None, description,
                        data.EMPTY_SET,
                        data.EMPTY_SET, [
                            data.Posting(self.liability_credit_card, self.to_amount(-entry_value), None, None, None,
                                         None),
                            data.Posting(self.expenses_account, self.to_amount(entry_value), None, None, None, None),
                        ])
                else:
                    print(f"### No match for transaction type: {transaction_type}")
                    continue
                entries.append(txn)
            return entries

    def file_account(self, file):
        return self.liability_credit_card

    def file_name(self, file):
        raise NotImplementedError()

    def file_date(self, file):
        # Extract the statement from file.
        with open(file.name) as f:
            # discard first 1 line
            next(f)

            date_line=next(f)
            date_str = date_line.split(',')[0].replace('"', '').strip()

        return datetime.strptime(date_str, '%d/%m/%Y').date()

    def to_amount(self, total_income: Decimal):
        return Amount(total_income, self.currency)


class FaturaImporterPriscila(FaturaImporter):
    def name(self):
        return 'FaturaInterPriscila'

    def identify(self, file):
        # Match if the filename is as downloaded
        return path.basename(file.name) == 'FaturaPriscila.csv'

    def file_name(self, file):
        dt = self.file_date(file)
        return f'Inter_Priscila_fatura_{dt.year}_{dt.month:02d}_{dt.day:02d}.csv'
    
    
class FaturaImporterRenzo(FaturaImporter):
    def name(self):
        return 'FaturaInterRenzo'

    def identify(self, file):
        # Match if the filename is as downloaded
        return path.basename(file.name) == 'FaturaRenzo.csv'

    def file_name(self, file):
        dt = self.file_date(file)
        return f'Inter_Renzo_fatura_{dt.year}_{dt.month:02d}_{dt.day:02d}.csv'