bachelorarbeit_importer/cleanup_script.py

#!/usr/bin/env python3
import argparse
import csv
import sys
import json
import logging
import requests
import hashlib
import io
import datetime
from rich import progress
from rich.logging import RichHandler
from rich.console import Console
from rich.traceback import install
install(show_locals=True, locals_max_length=150, locals_max_string=300)

class Company:
    def __init__(self, data, report, out):
        self.data = data
        self.out = out
        self.bvdid = data["BvD ID Nummer"]
        self.name = data["Unternehmensname"]
        self.gv2020 = None
        self.gv2021 = None
        self.gv2022 = None
        self.gv2023 = None
        self.gv2024 = None
        self.gn2020 = None
        self.gn2021 = None
        self.gn2022 = None
        self.gn2023 = None
        self.gn2024 = None
        try:
            if "Gewinn/(Verlust) vor Steuern EUR 2020" in data.keys() and data["Gewinn/(Verlust) vor Steuern EUR 2020"]!= '' and not self.gv2020:
                self.gv2020 = int(data["Gewinn/(Verlust) vor Steuern EUR 2020"])
            elif "Gewinn/Verlust vor Steuern EUR 2020" in data.keys() and data["Gewinn/Verlust vor Steuern EUR 2020"] != '' and not self.gv2020:
                self.gv2020 = int(data["Gewinn/Verlust vor Steuern EUR 2020"])
            else:
                self.gv2020 = None
        except ValueError:
            self.gv2020 = None
        try:
            if "Gewinn/(Verlust) vor Steuern EUR 2021" in data.keys() and data["Gewinn/(Verlust) vor Steuern EUR 2021"] != '' and not self.gv2021:
                self.gv2021 = int(data["Gewinn/(Verlust) vor Steuern EUR 2021"])
            elif "Gewinn/Verlust vor Steuern EUR 2021" in data.keys() and data["Gewinn/Verlust vor Steuern EUR 2021"] != '' and not self.gv2021:
                self.gv2021 = int(data["Gewinn/Verlust vor Steuern EUR 2021"])
            else:
                self.gv2021 = None
        except ValueError:
            self.gv2021 = None
        try:
            if "Gewinn/(Verlust) vor Steuern EUR 2022" in data.keys() and data["Gewinn/(Verlust) vor Steuern EUR 2022"] != '' and not self.gv2022:
                self.gv2022 = int(data["Gewinn/(Verlust) vor Steuern EUR 2022"])
            elif "Gewinn/Verlust vor Steuern EUR 2022" in data.keys() and data["Gewinn/Verlust vor Steuern EUR 2022"] != '' and not self.gv2022:
                self.gv2022 = int(data["Gewinn/Verlust vor Steuern EUR 2022"])
            else:
                self.gv2022 = None
        except ValueError:
            self.gv2022 = None
        try:
            if "Gewinn/(Verlust) vor Steuern EUR 2023" in data.keys() and data["Gewinn/(Verlust) vor Steuern EUR 2023"] != '' and not self.gv2023:
                self.gv2023 = int(data["Gewinn/(Verlust) vor Steuern EUR 2023"])
            elif "Gewinn/Verlust vor Steuern EUR 2023" in data.keys() and data["Gewinn/Verlust vor Steuern EUR 2023"] != '' and not self.gv2023:
                self.gv2023 = int(data["Gewinn/Verlust vor Steuern EUR 2023"])
            else:
                self.gv2023 = None
        except ValueError:
            self.gv2023 = None
        try:
            if "Gewinn/(Verlust) vor Steuern EUR 2024" in data.keys() and data["Gewinn/(Verlust) vor Steuern EUR 2024"] != '' and not self.gv2024:
                self.gv2024 = int(data["Gewinn/(Verlust) vor Steuern EUR 2024"])
            elif "Gewinn/Verlust vor Steuern EUR 2024" in data.keys() and data["Gewinn/Verlust vor Steuern EUR 2024"] != '' and not self.gv2024:
                self.gv2024 = int(data["Gewinn/Verlust vor Steuern EUR 2024"])
            else:
                self.gv2024 = None
        except ValueError:
            self.gv2024 = None
        try:
            if "Gewinn/(Verlust) nach Steuern EUR 2020" in data.keys() and data["Gewinn/(Verlust) nach Steuern EUR 2020"] != '' and not self.gn2020:
                self.gn2020 = int(data["Gewinn/(Verlust) nach Steuern EUR 2020"])
            elif "Gewinn/Verlust nach Steuern EUR 2020" in data.keys() and data["Gewinn/Verlust nach Steuern EUR 2020"] != '' and not self.gn2020:
                self.gn2020 = int(data["Gewinn/Verlust nach Steuern EUR 2020"])
            else:
                self.gn2020 = None
        except ValueError:
            self.gn2020 = None
        try:
            if "Gewinn/(Verlust) nach Steuern EUR 2021" in data.keys() and data["Gewinn/(Verlust) nach Steuern EUR 2021"] != '' and not self.gn2021:
                self.gn2021 = int(data["Gewinn/(Verlust) nach Steuern EUR 2021"])
            elif "Gewinn/Verlust nach Steuern EUR 2021" in data.keys() and data["Gewinn/Verlust nach Steuern EUR 2021"] != '' and not self.gn2021:
                self.gn2021 = int(data["Gewinn/Verlust nach Steuern EUR 2021"])
            else:
                self.gn2021 = None
        except ValueError:
            self.gn2021 = None
        try:
            if "Gewinn/(Verlust) nach Steuern EUR 2022" in data.keys() and data["Gewinn/(Verlust) nach Steuern EUR 2022"] != '' and not self.gn2022:
                self.gn2022 = int(data["Gewinn/(Verlust) nach Steuern EUR 2022"])
            elif "Gewinn/Verlust nach Steuern EUR 2022" in data.keys() and data["Gewinn/Verlust nach Steuern EUR 2022"] != '' and not self.gn2022:
                self.gn2022 = int(data["Gewinn/Verlust nach Steuern EUR 2022"])
            else:
                self.gn2022 = None
        except ValueError:
            self.gn2022 = None
        try:
            if "Gewinn/(Verlust) nach Steuern EUR 2023" in data.keys() and data["Gewinn/(Verlust) nach Steuern EUR 2023"] != '' and not self.gn2023:
                self.gn2023 = int(data["Gewinn/(Verlust) nach Steuern EUR 2023"])
            elif "Gewinn/Verlust nach Steuern EUR 2023" in data.keys() and data["Gewinn/Verlust nach Steuern EUR 2023"] != '' and not self.gn2023:
                self.gn2023 = int(data["Gewinn/Verlust nach Steuern EUR 2023"])
            else:
                self.gn2023 = None
        except ValueError:
            self.gn2023 = None
        try:
            if "Gewinn/(Verlust) nach Steuern EUR 2024" in data.keys() and data["Gewinn/(Verlust) nach Steuern EUR 2024"] != '' and not self.gn2024:
                self.gn2024 = int(data["Gewinn/(Verlust) nach Steuern EUR 2024"])
            elif "Gewinn/Verlust nach Steuern EUR 2024" in data.keys() and data["Gewinn/Verlust nach Steuern EUR 2024"] != '' and not self.gn2024:
                self.gn2024 = int(data["Gewinn/Verlust nach Steuern EUR 2024"])
            else:
                self.gn2024 = None
        except ValueError:
            self.gn2024 = None
        try:
            self.st2020 = int(data["Steuern EUR 2020"])
        except ValueError:
            self.st2020 = None
        try:
            self.st2021 = int(data["Steuern EUR 2021"])
        except ValueError:
            self.st2021 = None
        try:
            self.st2022 = int(data["Steuern EUR 2022"])
        except ValueError:
            self.st2022 = None
        try:
            self.st2023 = int(data["Steuern EUR 2023"])
        except ValueError:
            self.st2023 = None
        try:
            self.st2024 = int(data["Steuern EUR 2024"])
        except ValueError:
            self.st2024 = None
        try:
            self.ek2020 = int(data["Eigenkapital EUR 2020"])
        except ValueError:
            self.ek2020 = None
        try:
            self.ek2021 = int(data["Eigenkapital EUR 2021"])
        except ValueError:
            self.ek2021 = None
        try:
            self.ek2022 = int(data["Eigenkapital EUR 2022"])
        except ValueError:
            self.ek2022 = None
        try:
            self.ek2023 = int(data["Eigenkapital EUR 2023"])
        except ValueError:
            self.ek2023 = None
        try:
            self.ek2024 = int(data["Eigenkapital EUR 2024"])
        except ValueError:
            self.ek2024 = None
        self.report = report

    def calculate_tax(self):
        if not self.st2020 and self.gv2020 != None and self.gn2020 != None:
            self.st2020 = self.gv2020 - self.gn2020
        if not self.st2021 and self.gv2021 != None and self.gn2021 != None:
            self.st2021 = self.gv2021 - self.gn2021
        if not self.st2022 and self.gv2022 != None and self.gn2022 != None:
            self.st2022 = self.gv2022 - self.gn2022
        if not self.st2023 and self.gv2023 != None and self.gn2023 != None:
            self.st2023 = self.gv2023 - self.gn2023
        if not self.st2024 and self.gv2024 != None and self.gn2024 != None:
            self.st2024 = self.gv2024 - self.gn2024

    def validate(self):
        #fallback, in case tax wasn't already calculated
        self.calculate_tax()
        if True:
            if self.st2020 and self.ek2020:
                self.report.valid_data += 1
            else:
                self.report.invalid_data +=1
            if self.st2021 and self.ek2021:
                self.report.valid_data += 1
            else:
                self.report.invalid_data +=1
            if self.st2022 and self.ek2022:
                self.report.valid_data += 1
            else:
                self.report.invalid_data +=1
            if self.st2023 and self.ek2023:
                self.report.valid_data += 1
            else:
                self.report.invalid_data +=1
            if self.st2024 and self.ek2024:
                self.report.valid_data += 1
            else:
                self.report.invalid_data +=1


class dataimport:
    def __init__(self, filename, logfile, output, seek=0):
        self.seek = seek
        self.progress = progress.Progress(
            *progress.Progress.get_default_columns(),
            progress.MofNCompleteColumn(),
            progress.TimeElapsedColumn(),
            expand=True
        )
        self.filename = filename
        FORMAT = "%(message)s"
        self.logfile = open(logfile, 'a')
        self.output = output
        if self.logfile != "NONE":
            self.logconsole = Console(file=self.logfile)
            logging.basicConfig(
                level="NOTSET", format=FORMAT, datefmt="[%X]", handlers=[
                    RichHandler(rich_tracebacks=True, console=self.progress.console,
                                show_path=False, show_time=False, level="NOTSET"),
                    RichHandler(rich_tracebacks=True, console=self.logconsole,
                                show_path=False, level="WARNING")])
        else:
            logging.basicConfig(
                level="NOTSET", format=FORMAT, datefmt="[%X]", handlers=[
                    RichHandler(rich_tracebacks=True, console=self.progress.console,
                                show_path=False, show_time=False, level="NOTSET")])

        self.log = logging.getLogger("import")
        self.total_rows = self.get_total(self.filename)
        self.errors = 0
        self.data = {}
        self.duplicate_database_id = None
        self.task = self.progress.add_task(f"Importing {self.filename.split('/')[-1]}", total=self.get_total(self.filename))
        self.progress.update(self.task, advance=self.seek)
        self.valid_data = 0
        self.invalid_data = 0
        self.importer()

    def importer(self):
        with self.progress:
            with open(self.filename, mode='r', encoding='utf-8-sig', newline='') as csv_file:
                with open(self.output, mode='a+', encoding='utf-8-sig', newline='') as output_csv:
                    csv_reader = csv.DictReader(csv_file, delimiter=',')
                    out_names = []
                    output_writer = csv.DictWriter(output_csv, fieldnames=out_names)
                    self.log.warning(self.get_total(self.output))
                    if self.get_total(self.output) <= 0:
                        self.log.warning(f"WRITING HEADER FOR FILE {self.output}!")
                        output_writer.writeheader()
                    rownum = 0
                    for row in csv_reader:
                        if rownum < self.seek:
                            rownum += 1
                            continue
                        for key in csv_reader.fieldnames:
                            self.data[key] = row[key]
                        self.comp_import(self.data, output_writer)
                        self.data = {}
                        rownum += 1
                        self.progress.update(self.task, advance=1)
                self.progress.console.rule()
                self.log.info(f"Rows: {self.total_rows}")
                self.log.info(f"Valid: {self.valid_data}")
                self.log.info(f"Invalid: {self.invalid_data}")
                if self.errors == 0:
                    self.log.info(f"Errors: {self.errors}")
                    self.progress.console.rule()
                elif self.errors > 0:
                    self.log.error(f"Errors: {self.errors}")
                    self.progress.console.rule()
                else:
                    self.log.critical("ERROR CALCULATION EXCEPTION")

    def get_total(self, file):
        return sum(1 for _ in open(file, mode='r')) - 1
    
    def comp_import(self, data, out):
        current = Company(data, report=self, out=out)
        current.validate()



parser = argparse.ArgumentParser(description='Import data from ORBIS', epilog='Copyright Denkena Consulting')
parser.add_argument('filename', nargs="+")
parser.add_argument('-l', '--logfile', default="log_importer", nargs="?")
parser.add_argument('-o', '--output', default="export_cleaned.csv", nargs="?")
parser.add_argument('-s', '--seek', type=int, default=0)
args = parser.parse_args()
if len(args.filename) > 1 and args.seek > 0:
    parser.error("Seek combined with multiple files is a bad idea!")
for filename in args.filename:
    dataimport(filename, args.logfile, args.output, args.seek)