From a7e8ad49a792c8c16a44a11dc04dfca801149be5 Mon Sep 17 00:00:00 2001 From: "s.steuer" Date: Thu, 14 Aug 2025 17:51:21 +0200 Subject: [PATCH] =?UTF-8?q?Script=20nochmal=20komplett=20=C3=BCberarbeitet?= =?UTF-8?q?=20und=20von=20verschiebelogik.py=20in=20process=5Fbom.py=20umb?= =?UTF-8?q?enannt.=20Auch=20Die=20Logik=20wurde=20komplett=20=C3=BCberarbe?= =?UTF-8?q?itet.=20Die=20komplette=20Erstellung=20der=20finalen=20St=C3=BC?= =?UTF-8?q?ckliste=20l=C3=A4uft=20nun=20prozedural=20ausschlie=C3=9Flich?= =?UTF-8?q?=20=C3=BCber=20SQL-Befehle=20ab.=20Damit=20soll=20das=20Verhalt?= =?UTF-8?q?en=20von=20PL/SQL=20in=20SIVAS=20nachgestellt=20werden,=20um=20?= =?UTF-8?q?die=20Implementierung=20im=20SIVAS-Workflow=20zu=20erleichtern.?= =?UTF-8?q?=20Die=20Logik=20kommt=20nun=20komplett=20ohne=20Loops=20aus,?= =?UTF-8?q?=20d.h.=20die=20SQL-Abfragen=20sammeln=20nun=20direkt=20alle=20?= =?UTF-8?q?Positionsnummern,=20die=20jedem=20Logik-Fall=20zugeordnet=20sin?= =?UTF-8?q?d.=20Dies=20verbessert=20auch=20die=20Performance.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/process_bom.py | 768 +++++++++++++++++++++++++++++++++++++++++ lib/verschiebelogik.py | 695 ------------------------------------- 2 files changed, 768 insertions(+), 695 deletions(-) create mode 100644 lib/process_bom.py delete mode 100644 lib/verschiebelogik.py diff --git a/lib/process_bom.py b/lib/process_bom.py new file mode 100644 index 0000000..8c8888d --- /dev/null +++ b/lib/process_bom.py @@ -0,0 +1,768 @@ +import argparse +import csv +import os +import sqlite3 +import subprocess + +def main(): + parser = argparse.ArgumentParser(description='Programm erstellt finale SIVAS-Stückliste aus der RuleDesigner-Stücklisten-CSV unter Berücksichtigung der spezifizierten Verschiebelogik!', + prog='process_bom.py', + formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=50)) + + parser.register("action", "store_after_check", ArgparseDirectoryCheck) + parser.register("action", "store_after_code_check", ArgparseParentcodeCheck) + parser.add_argument("parent_code", action="store_after_code_check", type=str, help="Teilenummer der Baugruppe") + parser.add_argument("--csv_dir", action="store_after_check", default=os.environ.get("PROJECT_DATA_IN"), metavar="dir", help="Verzeichnis, aus dem die Stücklisten-CSV aus RuleDesigner eingelesen werden soll. Default: /data") + + subparsers = parser.add_subparsers(title="output type", required=True, help="Legt fest, in welcher Form die Stückliste ausgegeben werden soll. cmd = In Konsole, csv = als csv-Datei", dest="output_type") + sp_cmd = subparsers.add_parser("cmd", help="Ausgabe der fertig bearbeiteten Stückliste in der Konsole") + sp_csv = subparsers.add_parser("csv", help="Ausgabe der fertig bearbeiteten Stückliste als CSV-Datei") + sp_csv.register("action", "store_after_check", ArgparseDirectoryCheck) + + sp_csv.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben") + sp_cmd.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben") + sp_csv.add_argument("-od", "--out_dir", action="store_after_check", default=os.environ.get("PROJECT_DATA_OUT"), metavar="dir", help="Ausgabeverzeichnis fertigen Stücklisten-CSV") + sp_csv.add_argument("-o", "--open_csv", action="store_true", help="CSV-Datei nach dem schreiben öffnen") + args: dict = vars(parser.parse_args()) + + if os.environ.get("DB_PATH") is not None: + # nur 1 Datenbankinstanz über gesamte Laufzeit (ähnlich Singleton) + db = Database(os.environ.get("DB_PATH")) + else: + raise ValueError("Bitte bin/setenv.bat vor diesem Programm für erforderliche Umgebungsvariablen ausführen!") + + try: + old_bom = COMP(db) + new_bom = RD_COMP(db, args["parent_code"]) + csv_handler = CSVHandler() + sql_handler = SQLFileHandler() + bom_processing = BOMProcessing(db, args["parent_code"]) + + # SIVAS "nachbauen" + sql_source_filename = "sivas_replika.sql" + sivas_source_data = sql_handler.read_from_file(filename=sql_source_filename) + old_bom.rebuild_sivas(sivas_source_data) + + # CSV-Daten in Datenbank schreiben + csv_data = csv_handler.read_as_dict(filename=f"BOM_{args["parent_code"]}") + new_bom.create_tmp_table() + new_bom.insert_csv_data(csv_data) + + # Neue Stückliste erstellen / schreiben + bom_processing.write_new_bom() + + if args["output_type"] == "cmd": + if args["verbose"]: + output = repr(new_bom) + print(output) + else: + print(new_bom) + + elif args["output_type"] == "csv": + if args["verbose"]: + bom_data = new_bom.get_verbose_data() + else: + bom_data = new_bom.get_compact_data() + + filedir = args["out_dir"] + filename = f"BOM_{args["parent_code"]}_final.csv" + filepath = os.path.join(filedir, filename) + csv_handler.write_from_dict(data=bom_data, filedir=filedir, filename=filename) + + if args["open_csv"]: + subprocess.run(['cmd', '/c', 'start', '', filepath], shell=True) + + else: + NotImplementedError + finally: + db.close() + db.remove() # Damit bei jedem Lauf eine neue Datenbank angelegt wird -> mit :memory: traten Bugs auf + +class ArgparseDirectoryCheck(argparse.Action): + def __call__(self, parser, namespace, values: list[str], option_string = None): + if not os.path.isdir(values[0]): + raise NotADirectoryError + setattr(namespace, self.dest, values[0]) + +class ArgparseParentcodeCheck(argparse.Action): + def __call__(self, parser, namespace, values: str, option_string = None): + csv_filepath = os.path.join(os.environ.get("PROJECT_DATA_IN"), f"BOM_{values}.csv") + if not os.path.isfile(csv_filepath): + raise FileNotFoundError(csv_filepath) + setattr(namespace, self.dest, values) + +class FileHandler: + def check_filedir(self, filedir: str) -> str: + if filedir is None: + raise ValueError("Bitte bin/setenv.bat ausführen, oder Verzeichnis angegeben, in dem die Eingabedatei liegt!") + if not os.path.isdir(filedir): + raise NotADirectoryError + return filedir + + def check_filepath(self, filepath: str) -> str: + if not os.path.isfile(filepath): + raise FileNotFoundError("Angegebener Dateiname existiert nicht!") + return filepath + +class CSVHandler(FileHandler): + + def __init__(self, encoding="utf-8", delimiter: str = ";"): + self.delimiter = delimiter + self.encoding = encoding + + def read_as_dict(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA_IN"), ) -> list[dict[str]]: + + filename = f"{filename.removesuffix(".csv")}.csv" + filedir = self.check_filedir(filedir) + + filepath = os.path.join(filedir, filename) + if not filepath: + raise FileNotFoundError("Zur angegebenen Teilenummer existiert keine Stücklisten-CSV!") + + with open(filepath, "r", encoding=self.encoding, newline="") as csv_file: + reader = csv.DictReader(csv_file, delimiter=self.delimiter) + + required_col_names = ["ParentCode", "Positionsnummer", "ChildCode", "Menge", "Uebergabelaenge"] + if reader.fieldnames != required_col_names: + raise ValueError(f"Spaltennamen falsch. Korrekt: {required_col_names}") + + csv_data = [] + for line in reader: + line["Positionsnummer"] += "0" # Angleichung Positionsnummer an SIVAS-Format (Pos. * 10) + csv_data.append(line) + return csv_data + + def write_from_dict(self, filename: str, data: list[dict[str]], filedir: str = os.environ.get("PROJECT_DATA_OUT")) -> None: + + filedir = self.check_filedir(filedir) + filepath = os.path.join(filedir, f"{filename.removesuffix(".csv")}.csv") + + with open(filepath, "w", encoding=self.encoding, newline="") as csv_file: + writer = csv.DictWriter(csv_file, + fieldnames=data[0].keys(), + delimiter=self.delimiter) + writer.writeheader() + writer.writerows(data) + + self.check_filepath(filepath) + print(f"CSV geschrieben: {filepath}") + +class SQLFileHandler(FileHandler): + + def __init__(self, encoding: str = "utf-8"): + self.encoding = encoding + + def read_from_file(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA_IN")) -> str: + + filedir = self.check_filedir(filedir) + + filepath = os.path.join(filedir, filename) + filepath = self.check_filepath(filepath) + + with open(filepath, "r", encoding="utf-8", newline="") as sql_file: + sql = sql_file.read() + return sql + +class Database: + def __init__(self, db_path: str): + if not os.path.isdir(os.path.dirname(db_path)): + raise NotADirectoryError + + self.db_path = db_path + self.con = sqlite3.connect(self.db_path, autocommit=True) + self.con.row_factory = self.dict_factory + + @staticmethod + def dict_factory(cursor, row): + d = {} + for idx, col in enumerate(cursor.description): + d[col[0]] = row[idx] + return d + + def close(self): + self.con.close() + + def remove(self): + if os.path.isfile(self.db_path): + os.remove(self.db_path) + + +class RD_COMP: + + """ + Temporäre Tabelle, in die die Daten der Stücklisten-CSV aus RuleDesigner geschrieben werden + Tabelle wird für die Verarbeitung der Stückliste benötigt. Sobald die finale Stückliste steht und in die Tabelle 'COMP' + wird sie nicht mehr benötigt. + """ + + def __init__(self, db: Database, parent_code: str): + self.db = db + self.parent_code = parent_code + + def __repr__(self): + res = self.get_verbose_data() + output = self._sql_as_ascii_table(res) + return output + + def __str__(self): + res = self.get_compact_data() + output = self._sql_as_ascii_table(res) + return output + + def create_tmp_table(self) -> None: + """Schreibt neue Stücklistendaten aus CSV in temporäre Datenbanktabelle 'RD_COMP'.""" + + self.db.con.execute("""CREATE TABLE RD_COMP (ParentCode TEXT NOT NULL, + Positionsnummer INTEGER NOT NULL, + ChildCode TEXT NOT NULL, + Menge INTEGER NOT NULL, + Uebergabelaenge REAL NOT NULL, + PRIMARY KEY(Positionsnummer) + ) + """) + + def drop(self): + self.db.con.execute("DROP TABLE RD_COMP") + + def insert_csv_data(self, csv_data: list[dict[str]]): + self.db.con.executemany(""" + INSERT INTO RD_COMP (ParentCode, Positionsnummer, ChildCode, Menge, Uebergabelaenge) + VALUES (:ParentCode, :Positionsnummer, :ChildCode, :Menge, :Uebergabelaenge) + """, csv_data) + + def get_verbose_data(self) -> list[dict[str]]: + res = self.db.con.execute(""" + SELECT C.*, R.REGEL, B.BEDINGUNG + FROM COMP C + + LEFT JOIN VAR_TEIL_BEDINGUNG_2 B + ON C.ID = B.COMP_ID + + LEFT JOIN STKL_REGEL R + ON C.ID = R.COMP_ID + + WHERE C.PARENT = :parent_code + ORDER BY C.POSNR + """, {"parent_code": self.parent_code}).fetchall() + return res + + def get_compact_data(self) -> list[dict[str]]: + res = self.db.con.execute(""" + SELECT C.POSNR, C.KOMPTEILENR, C.ABW_ANZAHL, C.MENGE, C.LAENGE, C.ZUSCHNITT, R.REGEL, B.Bedingung + FROM COMP C + + LEFT JOIN VAR_TEIL_BEDINGUNG_2 B + ON C.ID = B.COMP_ID + + LEFT JOIN STKL_REGEL R + ON C.ID = R.COMP_ID + + WHERE C.PARENT = :parent_code + ORDER BY C.POSNR + """, {"parent_code": self.parent_code}).fetchall() + return res + + def _sql_as_ascii_table(self, sql_data: list[dict[str]]): + """Daten aus NEW-BOM als Tabelle ausgeben""" + headers = list(sql_data[0].keys()) + rows = [[str(row.get(h, "") if row.get(h, "") is not None else "") for h in headers] for row in sql_data] + col_widths = [max(len(str(h)), max(len(str(row[i])) for row in rows)) for i, h in enumerate(headers)] + header_line = " | ".join(f"{h:<{col_widths[i]}}" for i, h in enumerate(headers)) + separator = "-+-".join("-" * col_widths[i] for i in range(len(headers))) + row_lines = [" | ".join(f"{row[i]:<{col_widths[i]}}" for i in range(len(headers))) for row in rows] + return "\n".join([header_line, separator] + row_lines) + + +class COMP: + + def __init__(self, db: Database): + self.db = db + + def rebuild_sivas(self, sql: str) -> None: + self.db.con.executescript(sql) + + +class BOMProcessing: + + def __init__(self, db: Database, parent_code: str): + self.db = db + self.parent_code = parent_code + + def _get_codes_case_12_or_13(self) -> None: + """ + Ermittelt alle, die unter Fall 1.2 oder Fall 1.3 fallen. + Es wird dafür geprüft, welche Teilenummern NICHT unter Fall 1.1 fallen. + + Wieso wird nicht geprüft, welche Teilenummern in Fall 1.1 liegen? + - Weil die SQL-Abfrage dann deutlich komplexer wird. + + Die Prüfung läuft so ab: + 1. Teilenr-Längen-Kombinationen aus der alten und neuen Stückliste mit UNION ALL zusammenfügen + 2. Mit COUNT ermitteln, wie oft jede Teilenummer-Längen-Kombination in der zusammengefügten Stückliste auftritt + 3. Die Teilenummern in die temporäre Tabelle TMP_CODES_CASE_12_OR_13 schreiben, deren Teilenr-Längen-Kombination 1x vorkommen + => Diese Teilenummern unterliegen entweder Fall 1.2 oder Fall 1.3, aber definitiv nicht Fall 1.1, weil eine Zuordnung zu + Fall 1.1 nur dann möglich ist, wenn alle Teilenr-Längen-Kombinationen exakt 2x auftreten! + """ + self.db.con.execute("DROP TABLE IF EXISTS TMP_CODES_CASE_12_OR_13") + self.db.con.execute("CREATE TABLE TMP_CODES_CASE_12_OR_13 (KOMPTEILENR TEXT NOT NULL, PRIMARY KEY (KOMPTEILENR))") + self.db.con.execute(""" + WITH combined_bom AS ( + SELECT C.KOMPTEILENR, COALESCE(C.LAENGE, 0) AS LAENGE + FROM COMP C + WHERE C.PARENT = :parent_code + + UNION ALL + + SELECT ChildCode, Uebergabelaenge + FROM RD_COMP + WHERE ParentCode = :parent_code + ) + INSERT INTO TMP_CODES_CASE_12_OR_13 + SELECT DISTINCT KOMPTEILENR + FROM ( + SELECT KOMPTEILENR, LAENGE, COUNT(*) AS ANZAHL + FROM combined_bom + GROUP BY KOMPTEILENR, LAENGE + ) + WHERE ANZAHL = 1 + """, {"parent_code": self.parent_code}) + + def _get_codes_case_13(self) -> None: + """ + Ermittelt alle Teilenummern, die unter Fall 1.3 fallen + + Fall 1.3 umfasst alle Positionen, die entweder in der neuen Stückliste dazugekommen sind (Fall 1.31) oder deren + Teilenr-Längen-Kombination in beiden Stücklisten vorliegen (Fall 1.32). + + Wenn Positionen bereits unter Fall 1.1 oder Fall 1.2 fallen, dann werden sie nicht Fall 1.32 zugeordnet, selbst wenn + sie Fall 1.32 auf diese Positionen zutreffen würde (Vorrangsprinzip). + + Die Prüfung läuft so ab: + 1. Teilenr-Posnr-Kombinationen aus der alten und neuen Stückliste mit UNION ALL zusammenfügen, sofern sich die Teilenummern in der + Liste an Teilenummern befindet, die unter Fall 1.2 oder Fall 1.3 fallen (siehe Methode _get_codes_case_12_or_13()) + 2. Mit COUNT ermitteln, wie oft jede Teilenr-Posnr-Kombination in der zusammengefügten Stückliste auftritt + 3. Die Teilenummern ausgeben, deren Teilenr-Posnr-Kombination nur 1x vorkommen in die temporäre Tabelle TMP_CODES_CASE_13 schreiben + => Diese Teilenummern unterliegen definitv Fall 1.3, weil in Fall 1.2 alle Teilenr-Posnr-Kombinationen exakt 2x auftreten müssen! + """ + + self.db.con.execute("DROP TABLE IF EXISTS TMP_CODES_CASE_13") + self.db.con.execute("CREATE TABLE TMP_CODES_CASE_13 (KOMPTEILENR TEXT NOT NULL, PRIMARY KEY(KOMPTEILENR))") + self.db.con.execute(""" + WITH combined_bom AS ( + SELECT KOMPTEILENR, POSNR + FROM COMP + WHERE PARENT = :parent_code + AND KOMPTEILENR IN (SELECT * FROM TMP_CODES_CASE_12_OR_13) + + UNION ALL + + SELECT ChildCode, Positionsnummer + FROM RD_COMP + WHERE ParentCode = :parent_code + AND ChildCode IN (SELECT * FROM TMP_CODES_CASE_12_OR_13) + ) + + INSERT INTO TMP_CODES_CASE_13 (KOMPTEILENR) + SELECT DISTINCT KOMPTEILENR + FROM ( + SELECT KOMPTEILENR, COUNT(*) AS ANZAHL + FROM combined_bom + GROUP BY KOMPTEILENR, POSNR + ) + WHERE ANZAHL = 1 + """, {"parent_code": self.parent_code}) + + def _get_case_11_positions(self): + """ + Holt sich die Positionsnummern, die unter Fall 1.1 fallen. + + Deren Ermittlung funktioniert so: + 1. Teilenummern herausfiltern, die NICHT in der Tabelle TMP_CODES_CASE_12_OR_13 vorliegen + 2. Zu den gefundenen Teilenummern: + - Parent-Teilenr + - Posnr. laut neuer Stückliste + - ID laut alter Stückliste + in die temporäre Tabelle TMP_POSNRS_CASE_11 schreiben. + Hinweis: + - ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen + - Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können + """ + self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_11") + self.db.con.execute("""CREATE TABLE TMP_POSNRS_CASE_11 (COMP_ID INTEGER NOT NULL, + PARENT TEXT NOT NULL, + POSNR INTEGER NOT NULL, + PRIMARY KEY(POSNR))""") + self.db.con.execute(""" + INSERT INTO TMP_POSNRS_CASE_11 (COMP_ID, PARENT, POSNR) + + SELECT C.ID AS COMP_ID, C.PARENT, RD.Positionsnummer AS POSNR + FROM RD_COMP RD + + LEFT JOIN TMP_CODES_CASE_12_OR_13 TMP + ON RD.ChildCode = TMP.KOMPTEILENR + + LEFT JOIN COMP C + ON RD.ParentCode = C.PARENT + AND RD.ChildCode = C.KOMPTEILENR + AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) + + WHERE RD.ParentCode = :parent_code + AND TMP.KOMPTEILENR IS NULL + """, {"parent_code": self.parent_code}) + + def _get_case_12_positions(self) -> None: + """ + Holt sich die Positionsnummern, die unter Fall 1.2 fallen. + + Deren Ermittlung funktioniert so: + 1. Teilenummern herausfiltern, die in der Tabelle TMP_CODES_CASE_12_OR_13, aber NICHT in Tabelle TMP_CODES_CASE_13 existieren + 2. Zu den gefundenen Teilenummern: + - Parent-Teilenr + - Posnr. laut neuer Stückliste + - ID laut alter Stückliste + in die temporäre Tabelle TMP_POSNRS_CASE_12 schreiben. + Hinweis: + - ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen + - Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können + """ + + self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_12") + + self.db.con.execute(""" + CREATE TABLE TMP_POSNRS_CASE_12 (COMP_ID INTEGER NOT NULL, + PARENT TEXT NOT NULL, + POSNR INTEGER NOT NULL, + PRIMARY KEY(POSNR)) + """) + + self.db.con.execute(""" + INSERT INTO TMP_POSNRS_CASE_12 (COMP_ID, PARENT, POSNR) + + SELECT C.ID AS COMP_ID, C.PARENT, RD.Positionsnummer AS POSNR + FROM RD_COMP RD + + LEFT JOIN TMP_CODES_CASE_13 TMP + ON RD.ChildCode = TMP.KOMPTEILENR + + LEFT JOIN COMP C + ON RD.ParentCode = C.PARENT + AND RD.ChildCode = C.KOMPTEILENR + AND RD.Positionsnummer = C.POSNR + + WHERE RD.ParentCode = :parent_code + AND TMP.KOMPTEILENR IS NULL + AND RD.Positionsnummer NOT IN (SELECT POSNR FROM TMP_POSNRS_CASE_11); + """, {"parent_code": self.parent_code}) + + def _get_case_131_positions(self): + """ + Fall 1.31 - Positionsnummern holen, die neu angelegt werden müssen. + + Deren Ermittlung funktioniert so: + 1. Teilenummern herausfiltern, deren Teilenr-Längen-Kombination nur in der neuen Stückliste existieren und die zudem in der Tabelle + TMP_CODES_CASE_13 vorliegen. + Hinweis: + Es reicht nicht aus, lediglich zu prüfen, ob die Teilenr-Längen-Kombination nur in der neuen Stücklist vorliegt + Grund: + Positionsnummern die unter Fall 1.2 fallen, dürfen in der neuen Stückliste eine andere Teilenr-Längen-Kombination + als in der alten Stückliste aufweisen, d.h. die Teilenr-Längen-Kombi laut neuer Stückliste liegt auch nur dort vor. Trotzdem + werden diese Positionsnummern nicht neu angelegt, sondern von der alten in die neuen Stückliste verschoben! + + 2. Zu den gefundenen Teilenummern: + - Parent-Teilenr + - Posnr. laut neuer Stückliste + - ID laut alter Stückliste + in die temporäre Tabelle TMP_POSNRS_CASE_131 schreiben. + Hinweis: + - ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen + - Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können + """ + self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_131") + self.db.con.execute("CREATE TABLE TMP_POSNRS_CASE_131 (POSNR INTEGER NOT NULL, PRIMARY KEY(POSNR))") + self.db.con.execute(""" + INSERT INTO TMP_POSNRS_CASE_131 + + SELECT RD.Positionsnummer AS POSNR + FROM RD_COMP RD + + LEFT JOIN COMP C + ON RD.ParentCode = C.PARENT + AND RD.ChildCode = C.KOMPTEILENR + AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) + + WHERE RD.ParentCode = :parent_code + AND RD.ChildCode IN (SELECT * FROM TMP_CODES_CASE_13) + AND C.KOMPTEILENR IS NULL; + """, {"parent_code": self.parent_code}) + + def _get_case_132_positions(self) -> None: + """ + Fall 1.32 - Positionsnummern holen, die von der alten in die neue Stückliste verschoben werden + + Deren Ermittlung funktioniert so: + 1. Teilenummern herausfiltern, deren Teilenr-Längen-Kombination in beiden Stückliste existieren und die zudem in der Tabelle + TMP_CODES_CASE_13 vorliegen. Lediglich zu prüfen, ob die Teilenr-Längen-Kombination nur in der neuen Stückliste vorliegt reicht nicht aus! + Grund: + Positionsnummern die Fall 1.1 zugeordnet werden können oder Positionsnummern, die unter Fall 1.2 fallen und deren Länge dabei in beiden + Stücklisten gleich bleibt würden grundlegend ebenfalls unter Fall 1.32 fallen und somit identifiziert werden. + Es gilt jedoch das Vorrangsprinzip, d.h. die Zuordnung zu Fall 1.1 oder Fall 1.2 erfolgt (sofern möglich) immer vor der Zuordnung + zu Fall 1.32! + + 2. Zu den gefundenen Teilenummern: + - Parent-Teilenr + - Posnr. laut neuer Stückliste + - ID laut alter Stückliste + in die temporäre Tabelle TMP_POSNRS_CASE_132 schreiben. + Hinweis: + - ID dient in der späteren UPDATE-Methode dazu, die Positionsnummer zu finden, die aktualisiert werden sollen + - Parent und POSNR werden benötigt, um die neuen Stücklistendaten laut neuer Stückliste identifizieren zu können + """ + + self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_132") + self.db.con.execute(""" + CREATE TABLE TMP_POSNRS_CASE_132 (COMP_ID INTEGER NOT NULL, + PARENT TEXT NOT NULL, + POSNR INTEGER NOT NULL, + PRIMARY KEY(POSNR))""") + self.db.con.execute(""" + INSERT INTO TMP_POSNRS_CASE_132 (COMP_ID, PARENT, POSNR) + + SELECT C.ID AS COMP_ID, C.PARENT, RD.Positionsnummer AS POSNR + FROM RD_COMP RD + + INNER JOIN COMP C + ON RD.ParentCode = C.PARENT + AND RD.ChildCode = C.KOMPTEILENR + AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) + + WHERE RD.ParentCode = :parent_code + AND RD.ChildCode IN (SELECT * FROM TMP_CODES_CASE_13) + """, {"parent_code": self.parent_code}) + + def delete(self): + """ + Entfernt die Positionsnummern aus der alten Stückliste, die nicht mehr in der neuen Stückliste vorliegen. + + Hinweis: + Bei der Ermittlung der zu löschenden Datensätze dürfen die Positionsnummern, die verschoben werden NICHT berücksichtigt werden. + Die alte Lösch-Logik: "Die Teile die in der neuen Stückliste nicht drin sind, werden rausgelöscht" nur noch eingeschränkt anwendbar! + + Grund: + Das liegt vor allem an Fall 1.2, denn per Definition darf in diesem Fall die Länge in der neuen Stückliste abweichen, solange alle + Teilenr-Posnr-Kombinationen in beiden Stücklisten identsich sind, siehe hierzu z.B. Pos. 3 in Beispiel 3 unter doc/Beispiele.pdf: + Dort liegt die Länge 200 aus der alten Stückliste nicht mehr in der neuen Stückliste vor. Pos. 3 würde also laut alter Logik gelöscht werden. + Da diese Positionsnummer aber unter Fall 1.2 fällt, wird sie in die neue Stückliste verschoben, d.h. sie darf nicht gelöscht werden! + + """ + + self.db.con.execute("DROP TABLE IF EXISTS TMP_IDS_TO_DELETE") + self.db.con.execute("CREATE TABLE TMP_IDS_TO_DELETE (ID INTEGER NOT NULL, PRIMARY KEY(ID))") + self.db.con.execute(""" + WITH COMBINED_UPDATE_DATA AS ( + SELECT * FROM TMP_POSNRS_CASE_11 + UNION + SELECT * FROM TMP_POSNRS_CASE_12 + UNION + SELECT * FROM TMP_POSNRS_CASE_132 + ) + INSERT INTO TMP_IDS_TO_DELETE + + SELECT C.ID + FROM COMP C + + LEFT JOIN COMBINED_UPDATE_DATA CUD + ON C.ID = CUD.COMP_ID + + WHERE C.PARENT = :parent_code + AND CUD.COMP_ID IS NULL + """, {"parent_code": self.parent_code}) + + # Entfernung der Stücklistenpositionen aus COMP + self.db.con.execute("DELETE FROM COMP WHERE ID IN (SELECT ID FROM TMP_IDS_TO_DELETE)") + # Entfernung der TOS-Bedingungen aus VAR_TEIL_BEDINGUNG_2 + self.db.con.execute("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID IN (SELECT ID FROM TMP_IDS_TO_DELETE)") + # Entfernung der TOS-Regeln aus STKL_REGEL + self.db.con.execute("DELETE FROM STKL_REGEL WHERE COMP_ID IN (SELECT ID FROM TMP_IDS_TO_DELETE)") + + def update(self): + """ + Aktualisiert die Positionsnummern, die von der alten in die neue Stückliste verschoben werden sollen. + + Dies umfasst alle Positionsnummern, die unter die Fälle 1.1, 1.2 und 1.32 fallen. Sie wurden in den Methoden + _get_case_11_positions, _get_case_12_positions & _get_case_132_positions ermittelt und in die temporären + Tabellen TMP_POSNRS_CASE_11, TMP_POSNRS_CASE_12, TMP_POSNRS_CASE_132 geschrieben. + + Jetzt passiert folgendes: + 1. Die Positionen aus den 3 Tabellen werden zu einer temporären Tabelle, ALL_POSNRS_TO_UPDATE kombiniert. + 2. Zu allen Positionen aus in dieser kombinierten, temporären Tabelle werden dann die neuen Stücklisteninformationen aus COMP, + bzw. RD_COMP bestimmt und in eine weitere, temporäre, aber persistente Tabelle TMP_UPDATE_DATA geschrieben. + 3. Durchführen der Updates in den Tabellen COMP (für die Stücklisteninformationen), VAR_TEIL_BEDINGUNG_2 (für die TOS-Bedingungen) + und STKL_REGEL für die TOS-Regeln + """ + + self.db.con.execute("DROP TABLE IF EXISTS TMP_UPDATE_DATA") + self.db.con.execute("CREATE TABLE TMP_UPDATE_DATA (ID, POSNR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE, BEDINGUNG, REGEL)") + + # Neue Stücklisteninformationen ermitteln + self.db.con.execute(""" + WITH ALL_POSNRS_TO_UPDATE AS ( + SELECT * FROM TMP_POSNRS_CASE_11 + UNION + SELECT * FROM TMP_POSNRS_CASE_12 + UNION + SELECT * FROM TMP_POSNRS_CASE_132 + ) + INSERT INTO TMP_UPDATE_DATA (ID, POSNR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE, BEDINGUNG, REGEL) + + SELECT C.ID, + RD.Positionsnummer AS POSNR, + + -- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV + CASE + WHEN RD.Uebergabelaenge = 0 THEN NULL + ELSE RD.Menge + END AS ABW_ANZAHL, + + -- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV + CASE + WHEN RD.Uebergabelaenge = 0 THEN RD.Menge + ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT, 0)) / 1000 + END AS MENGE, + + -- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV + CASE + WHEN RD.Uebergabelaenge = 0 THEN NULL + ELSE RD.Uebergabelaenge + END AS LAENGE, + + C.ZUSCHNITT, + C.DISPO_AKTIV_JN, + C.RUESTCODE, + B.BEDINGUNG, + R.REGEL + + FROM ALL_POSNRS_TO_UPDATE CUD + INNER JOIN COMP C + ON CUD.COMP_ID = C.ID + + INNER JOIN RD_COMP RD + ON CUD.PARENT = RD.ParentCode + AND CUD.POSNR = RD.Positionsnummer + + LEFT JOIN VAR_TEIL_BEDINGUNG_2 B + ON CUD.COMP_ID = B.COMP_ID + + LEFT JOIN STKL_REGEL R + ON CUD.COMP_ID = R.COMP_ID + """) + + # Stücklisteninformationen updaten + self.db.con.execute(""" + UPDATE COMP + SET POSNR = UD.POSNR, + ABW_ANZAHL = UD.ABW_ANZAHL, + MENGE = UD.MENGE, + LAENGE = UD.LAENGE, + ZUSCHNITT = UD.ZUSCHNITT, + DISPO_AKTIV_JN = UD.DISPO_AKTIV_JN, + RUESTCODE = UD.RUESTCODE + FROM TMP_UPDATE_DATA UD + WHERE COMP.ID = UD.ID + """) + + # TOS-Bedingungen updaten + self.db.con.execute(""" + UPDATE VAR_TEIL_BEDINGUNG_2 + SET BEDINGUNG = UD.BEDINGUNG + FROM TMP_UPDATE_DATA UD + WHERE COMP_ID = UD.ID + """) + + # TOS-Regeln updaten + self.db.con.execute(""" + UPDATE STKL_REGEL + SET REGEL = UD.REGEL + FROM TMP_UPDATE_DATA UD + WHERE COMP_ID = UD.ID + """) + + def insert(self): + """ + Neuanlage aller Positionen, die nur in der neuen Stückliste vorliegen. + Die Positionsnummern wurde bereits in der Methode _get_codes_case_131 ermittelt und in die temporäre Tabelle + TMP_POSNRS_CASE_131 geschrieben. + + Hinweise: + - Neue Positionen werden immer ohne TOS-Regeln oder -Bedingungen angelegt, d.h. es erfolgen keine INSERT-Statements + in den Tabellen VAR_TEIL_BEDINGUNG_2 oder STKL_REGEL! + - Zur Verdeutlichung, welche Werte z.B. in der Spalte Ruestcode oder Dispoaktiv nach der Neuanlage vorliegen, werden + die Defaultwerte aus der PARTS-Tabelle geholt. Ob ein solcher INSERT auch in SIVAS erforderlich ist, war zum Zeitpunkt + der Erstellung nicht klar. + """ + self.db.con.execute(""" + INSERT INTO COMP (PARENT, POSNR, KOMPTEILENR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE) + + SELECT RD.ParentCode AS PARENT, + RD.Positionsnummer AS POSNR, + RD.ChildCode AS KOMPTEILENR, + + -- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV + CASE + WHEN RD.Uebergabelaenge = 0 THEN NULL + ELSE RD.Menge + END AS ABW_ANZAHL, + + -- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV + CASE + WHEN RD.Uebergabelaenge = 0 THEN RD.Menge + ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT,0)) / 1000 + END AS MENGE, + + -- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV + CASE + WHEN RD.Uebergabelaenge = 0 THEN NULL + ELSE RD.Uebergabelaenge + END AS LAENGE, + NULL AS ZUSCHNITT, + P.DISPO_AKTIV_JN, + P.RUESTCODE + + FROM RD_COMP RD + + LEFT JOIN COMP C + ON RD.ChildCode = C.KOMPTEILENR + AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) + AND RD.ParentCode = C.PARENT + + LEFT JOIN PARTS P + ON RD.ChildCode = P.TEILENR + + WHERE C.KOMPTEILENR IS NULL + AND RD.ParentCode = :parent_code + AND RD.Positionsnummer IN (SELECT * FROM TMP_POSNRS_CASE_131) + """, {"parent_code": self.parent_code}) + + def drop_tmp_tables(self): + self.db.con.execute("DROP TABLE TMP_CODES_CASE_12_OR_13") + self.db.con.execute("DROP TABLE TMP_CODES_CASE_13") + self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_11") + self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_12") + self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_131") + self.db.con.execute("DROP TABLE TMP_POSNRS_CASE_132") + self.db.con.execute("DROP TABLE TMP_IDS_TO_DELETE") + self.db.con.execute("DROP TABLE TMP_UPDATE_DATA") + + + def write_new_bom(self) -> None: + self._get_codes_case_12_or_13() + self._get_codes_case_13() + self._get_case_11_positions() + self._get_case_12_positions() + self._get_case_131_positions() + self._get_case_132_positions() + self.delete() + self.insert() + self.update() + self.drop_tmp_tables() +if __name__ == "__main__": + main() diff --git a/lib/verschiebelogik.py b/lib/verschiebelogik.py deleted file mode 100644 index 49f8dda..0000000 --- a/lib/verschiebelogik.py +++ /dev/null @@ -1,695 +0,0 @@ -import argparse -import csv -import os -import sqlite3 - -def main(): - parser = argparse.ArgumentParser(description='Programm bearbeitet Stücklisten-CSV aus RuleDesigner, sodass Spaltenüberschreibungen + TOS-Bedingungen und -Regeln (falls vorhanden) mit verschoben werden', - prog='verschiebelogik', - formatter_class=lambda prog: argparse.HelpFormatter(prog, max_help_position=50)) - parser.register("action", "store_after_check", ArgparseDirectoryCheck) - parser.add_argument("parent_code", action="store", type=str, help="Teilenummer der Baugruppe") - parser.add_argument("-cid", "--in_csv_dir", action="store_after_check", default=os.environ.get("PROJECT_DATA"), help="Eingabeverzeichnis der Stücklisten-CSV aus RuleDesigner") - subparsers = parser.add_subparsers(title="output type", help="Ausgabe der fertigen Stückliste in Konsole oder als CSV", dest="output_type") - sp_cmd = subparsers.add_parser("console", help="Ausgabe der fertig bearbeiteten Stückliste in der Konsole") - sp_csv = subparsers.add_parser("csv", help="Ausgabe der fertig bearbeiteten Stückliste als CSV-Datei") - sp_csv.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben") - sp_csv.register("action", "store_after_check", ArgparseDirectoryCheck) - sp_cmd.add_argument("-v", "--verbose", action="store_true", help="Alle Spalten ausgeben") - sp_csv.add_argument("-cod", "--out_csv_dir", action="store_after_check", default=os.environ.get("PROJECT_WORK"), help="Ausgabeverzeichnis fertigen Stücklisten-CSV") - args: dict = vars(parser.parse_args()) - - if os.environ.get("DB_PATH") is not None: - # nur 1 Datenbankinstanz über gesamte Laufzeit (ähnlich Singleton) - db = Database(os.environ.get("DB_PATH")) - else: - raise ValueError("Bitte bin/setenv.bat vor diesem Programm für erforderliche Umgebungsvariablen ausführen!") - - try: - old_bom = COMP(db) - new_bom = RD_COMP(db, args["parent_code"]) - csv_handler = CSVHandler() - sql_handler = SQLFileHandler() - bom_processing = BOMProcessing(db, args["parent_code"]) - - # SIVAS "nachbauen" - sql_source_filename = "sivas_replika.sql" - sivas_source_data = sql_handler.read_from_file(filename=sql_source_filename) - old_bom.rebuild_sivas(sivas_source_data) - - # CSV-Daten in Datenbank schreiben - csv_data = csv_handler.read_as_dict(filename=f"BOM_{args["parent_code"]}") - new_bom.create() - new_bom.insert_csv_data(csv_data) - - # Neue Stückliste erstellen / schreiben - bom_processing.write_new_bom() - - if args["output_type"] == "console": - if args["verbose"]: - output = repr(new_bom) - print(output) - else: - print(new_bom) - elif args["output_type"] == "csv": - if args["verbose"]: - bom_data = new_bom.get_verbose_data() - else: - bom_data = new_bom.get_compact_data() - - csv_handler.write_from_dict(data=bom_data, - filename=f"BOM_{args["parent_code"]}_final", - filedir=args["out_csv_dir"]) - else: - NotImplementedError - - finally: - db.close() - db.remove() # Damit bei jedem Lauf eine neue erstellt wird -> mit :memory: traten Bugs auf - - -class ArgparseDirectoryCheck(argparse.Action): - def __call__(self, parser, namespace, values: list[str], option_string = None): - if not os.path.isdir(values[0]): - raise NotADirectoryError - setattr(namespace, self.dest, values[0]) - -class FileHandler: - def check_filedir(self, filedir: str) -> str: - if filedir is None: - raise ValueError("Bitte bin/setenv.bat ausführen, oder Verzeichnis angegeben, in dem die Eingabedatei liegt!") - if not os.path.isdir(filedir): - raise NotADirectoryError - return filedir - - def check_filepath(self, filepath: str) -> str: - if not os.path.isfile(filepath): - raise FileNotFoundError("Angegebener Dateiname existiert nicht!") - return filepath - -class CSVHandler(FileHandler): - - def __init__(self, encoding="utf-8", delimiter: str = ";"): - self.delimiter = delimiter - self.encoding = encoding - - def read_as_dict(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA"), ) -> list[dict[str]]: - - filename = f"{filename.removesuffix(".csv")}.csv" - filedir = self.check_filedir(filedir) - - filepath = os.path.join(filedir, filename) - if not filepath: - raise FileNotFoundError("Zur angegebenen Teilenummer existiert keine Stücklisten-CSV!") - - with open(filepath, "r", encoding=self.encoding, newline="") as csv_file: - reader = csv.DictReader(csv_file, delimiter=self.delimiter) - - required_col_names = ["ParentCode", "Positionsnummer", "ChildCode", "Menge", "Uebergabelaenge"] - if reader.fieldnames != required_col_names: - raise ValueError(f"Spaltennamen falsch. Korrekt: {required_col_names}") - - csv_data = [] - for line in reader: - line["Positionsnummer"] += "0" # Angleichung Positionsnummer an SIVAS-Format (Pos. * 10) - csv_data.append(line) - return csv_data - - def write_from_dict(self, filename: str, data: list[dict[str]], filedir: str = os.environ.get("PROJECT_WORK")) -> None: - - filedir = self.check_filedir(filedir) - filepath = os.path.join(filedir, f"{filename.removesuffix(".csv")}.csv") - - with open(filepath, "w", encoding=self.encoding, newline="") as csv_file: - writer = csv.DictWriter(csv_file, - fieldnames=data[0].keys(), - delimiter=self.delimiter) - writer.writeheader() - writer.writerows(data) - - self.check_filepath(filepath) - -class SQLFileHandler(FileHandler): - - def __init__(self, encoding: str = "utf-8"): - self.encoding = encoding - - def read_from_file(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA")) -> str: - - filedir = self.check_filedir(filedir) - - filepath = os.path.join(filedir, filename) - filepath = self.check_filepath(filepath) - - with open(filepath, "r", encoding="utf-8", newline="") as sql_file: - sql = sql_file.read() - return sql - -class Database: - def __init__(self, db_path: str): - if not os.path.isdir(os.path.dirname(db_path)): - raise NotADirectoryError - - self.db_path = db_path - self.con = sqlite3.connect(self.db_path, autocommit=True) - self.con.row_factory = self.dict_factory - - @staticmethod - def dict_factory(cursor, row): - d = {} - for idx, col in enumerate(cursor.description): - d[col[0]] = row[idx] - return d - - def close(self): - self.con.close() - - def remove(self): - if os.path.isfile(self.db_path): - os.remove(self.db_path) - - -class RD_COMP: - - """ - Temporäre Tabelle, in die die Daten der Stücklisten-CSV aus RuleDesigner geschrieben werden - Tabelle wird für die Verarbeitung der Stückliste benötigt. Sobald die finale Stückliste steht und in die Tabelle 'COMP' - wird sie nicht mehr benötigt. - """ - - def __init__(self, db: Database, parent_code: str): - self.db = db - self.parent_code = parent_code - - def __repr__(self): - res = self.get_verbose_data() - output = self._sql_as_ascii_table(res) - return output - - def __str__(self): - res = self.get_compact_data() - output = self._sql_as_ascii_table(res) - return output - - - def create(self) -> None: - """Schreibt neue Stücklistendaten aus CSV in temporäre Datenbanktabelle 'RD_COMP'.""" - - self.db.con.execute("""CREATE TABLE RD_COMP (ParentCode TEXT NOT NULL, - Positionsnummer INTEGER NOT NULL, - ChildCode TEXT NOT NULL, - Menge INTEGER NOT NULL, - Uebergabelaenge REAL NOT NULL, - PRIMARY KEY(Positionsnummer))""") - - def drop(self): - pass - - def insert_csv_data(self, csv_data: list[dict[str]]): - self.db.con.executemany(""" INSERT INTO RD_COMP (ParentCode, Positionsnummer, ChildCode, Menge, Uebergabelaenge) - VALUES (:ParentCode, :Positionsnummer, :ChildCode, :Menge, :Uebergabelaenge)""", csv_data) - - def get_all_bom_data(self) -> list[dict[str]]: - res = self.db.con.execute("SELECT * FROM RD_COMP WHERE ParentCode = :parent ORDER BY Positionsnummer", - {"parent": self.parent_code}).fetchall() - return res - - def get_verbose_data(self) -> list[dict[str]]: - res = self.db.con.execute(""" - SELECT C.*, - R.REGEL, - B.BEDINGUNG - FROM COMP C - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID - WHERE C.PARENT = :parent_code - ORDER BY C.POSNR - """, {"parent_code": self.parent_code}).fetchall() - return res - - def get_compact_data(self) -> list[dict[str]]: - res = self.db.con.execute(""" - SELECT C.POSNR, - C.KOMPTEILENR, - C.ABW_ANZAHL, - C.MENGE, - C.LAENGE, - C.ZUSCHNITT, - R.REGEL, - B.Bedingung - FROM COMP C - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID - WHERE C.PARENT = :parent_code - ORDER BY C.POSNR - """, {"parent_code": self.parent_code}).fetchall() - return res - - def _sql_as_ascii_table(self, sql_data: list[dict[str]]): - """SQL-Rohdaten als Tabelle ausgeben""" - headers = list(sql_data[0].keys()) - rows = [[str(row.get(h, "") if row.get(h, "") is not None else "") for h in headers] for row in sql_data] - col_widths = [max(len(str(h)), max(len(str(row[i])) for row in rows)) for i, h in enumerate(headers)] - header_line = " | ".join(f"{h:<{col_widths[i]}}" for i, h in enumerate(headers)) - separator = "-+-".join("-" * col_widths[i] for i in range(len(headers))) - row_lines = [" | ".join(f"{row[i]:<{col_widths[i]}}" for i in range(len(headers))) for row in rows] - return "\n".join([header_line, separator] + row_lines) - - def insert(self, new_data: dict[str, str]) -> None: - self.db.con.executemany("""INSERT INTO COMP (PARENT, POSNR, KOMPTEILENR, DISPO_AKTIV_JN, MENGE, ABW_ANZAHL, LAENGE, ZUSCHNITT, RUESTCODE) - VALUES (:PARENT, :POSNR, :KOMPTEILENR, :DISPO_AKTIV_JN, :MENGE, :ABW_ANZAHL, :LAENGE, :ZUSCHNITT, :RUESTCODE)""", - new_data) - - def delete(self, ids: list[dict[str]]) -> None: - self.db.con.executemany("DELETE FROM COMP WHERE ID = :ID", ids) - - def update(self, pos_data: list[dict[str]]) -> None: - self.db.con.executemany(""" - UPDATE COMP - SET POSNR = :POSNR, - ABW_ANZAHL = :ABW_ANZAHL, - MENGE = :MENGE, - LAENGE = :LAENGE, - ZUSCHNITT = :ZUSCHNITT, - DISPO_AKTIV_JN = :DISPO_AKTIV_JN, - RUESTCODE = :RUESTCODE - WHERE ID = :ID - """, pos_data) - - -class COMP: - - """Beinhaltet Methoden, zur Interaktion mit Tabelle COMP (gleichnamig wie in SIVAS)""" - - def __init__(self, db: Database): - self.db = db - - def rebuild_sivas(self, sql: str) -> None: - self.db.con.executescript(sql) - - def get_child_ids(self, parent_code: str) -> list[dict[str]]: - res = self.db.con.execute("SELECT ID FROM COMP WHERE PARENT = :parent", - {"parent": parent_code}).fetchall() - return res - - -class VAR_TEIL_BEDINGUNGEN_2: - - """ - Beinhaltet Methoden, zur Interaktion mit Tabelle VAR_TEIL_BEDINGUNG_2 (gleichnamig wie in SIVAS), - in der TOS-Bedingungen gespeichert sind - """ - - def __init__(self, db: Database) -> None: - self.db = db - - def update(self, data: list[dict[str]]) -> None: - self.db.con.executemany("UPDATE VAR_TEIL_BEDINGUNG_2 SET BEDINGUNG = :BEDINGUNG WHERE COMP_ID = :ID", data) - - def delete(self, data: list[dict[str]]) -> None: - self.db.con.executemany("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID = :ID", data) - - -class STKL_REGEL: - - """ - Beinhaltet Methoden, zur Interaktion mit Tabelle STKL_REGEL (gleichnamig wie in SIVAS), - in der TOS-Regeln gespeichert sind - """ - - def __init__(self, db: Database) -> None: - self.db = db - - def update(self, data: list[dict[str]]) -> None: - self.db.con.executemany("UPDATE STKL_REGEL SET REGEL = :REGEL WHERE COMP_ID = :ID", data) - - def delete(self, data: list[dict[str]]) -> None: - self.db.con.executemany("DELETE FROM STKL_REGEL WHERE COMP_ID = :ID", data) - -class BOMProcessing: - - """ - Die Methoden dieser Klasse verarbeiten die Daten aus der RuleDesigner-Stücklisten CSV und initiiert das Schreiben der verarbeiteten - Daten in COMP. - """ - - def __init__(self, db: Database, parent_code: str): - self.db = db - self.parent_code = parent_code - self.new_bom = RD_COMP(db, parent_code) - self.old_bom = COMP(db) - self.tos_bedingungen = VAR_TEIL_BEDINGUNGEN_2(db) - self.tos_regeln = STKL_REGEL(db) - - def extract_col_data(self, query_results: list[dict[str]], *col_names: tuple) -> list[dict[str]]: - extracted_data = [] - for result in query_results: - data = {col_name: result[col_name] for col_name in col_names if col_name in result} - extracted_data.append(data) - return extracted_data - - def _get_positions_to_move_old_logic(self) -> set | None: - """ - Gibt alle Positionsnummern zurück, die gemäß alter Logik verschoben werden - """ - res = self.db.con.execute(""" - SELECT RD.Positionsnummer AS POS - - FROM RD_COMP RD - INNER JOIN COMP C - ON RD.ParentCode = C.PARENT - AND RD.ChildCode = C.KOMPTEILENR - AND RD.Uebergabelaenge = COALESCE(C.LAENGE,0) - - WHERE RD.ParentCode = :parent - """, {"parent": self.parent_code}).fetchall() - - if res is not None: - res = set([line["POS"] for line in res]) - return res - - def _check_identical_code_len_combo(self, child_code: str) -> bool: - """ - Prüfung Fall 1.1: - """ - res = self.db.con.execute(""" - SELECT CASE - -- Prüfung 1: Liegen alle Teilenr.+Länge-Kombinationen aus der neuen Stückliste auch in der alten Stückliste vor? - WHEN NOT EXISTS ( - SELECT 1 - FROM RD_COMP - LEFT JOIN COMP - ON RD_COMP.ParentCode = COMP.PARENT - AND RD_COMP.ChildCode = COMP.KOMPTEILENR - AND RD_COMP.Uebergabelaenge = COALESCE(COMP.LAENGE, 0) - - WHERE COMP.KOMPTEILENR IS NULL - AND RD_COMP.ChildCode = :child - AND RD_COMP.ParentCode = :parent - ) - -- Prüfung 2: Liegen alle Teilenr.+Länge-Kombinationen aus der alten Stückliste auch in der neuen Stückliste vor? - AND NOT EXISTS ( - SELECT 1 - FROM COMP - LEFT JOIN RD_COMP - ON COMP.PARENT = RD_COMP.ParentCode - AND COMP.KOMPTEILENR = RD_COMP.ChildCode - AND COALESCE(COMP.LAENGE, 0) = RD_COMP.Uebergabelaenge - - WHERE RD_COMP.ChildCode IS NULL - AND COMP.KOMPTEILENR = :child - AND COMP.PARENT = :parent - ) - -- Wenn Prüfung 1 und 2 zutreffen, dann liegen alle Teilenr.+Länge-Kombinationen in beiden Stücklisten vor (Fall 1.1 trifft zu) - THEN 'True' - ELSE 'False' - END AS 'Fall 1.1' - """, - {"parent": self.parent_code, "child": child_code}).fetchone() - - return True if res["Fall 1.1"] == 'True' else False - - def _check_same_pos(self, child_code: str) -> bool: - """ - Prüfung Fall 1.2 - """ - res = self.db.con.execute(""" - SELECT CASE - WHEN NOT EXISTS ( - -- Prüfung, ob in der neuen Stückliste an denselben Positionen wie in der alten Stückliste Längenvarianten zur angegebenen Teilenr. vorliegen - SELECT 1 - FROM COMP - LEFT JOIN RD_COMP - ON COMP.PARENT = RD_COMP.ParentCode - AND COMP.KOMPTEILENR = RD_COMP.ChildCode - AND COMP.POSNR = RD_COMP.Positionsnummer - - WHERE RD_COMP.ChildCode IS NULL - AND COMP.PARENT = :parent - AND COMP.KOMPTEILENR = :child - ) - AND NOT EXISTS ( - -- Prüfung, ob in der alten Stückliste an denselben Positionen wie in der neuen Stückliste Längenvarianten zur angegebenen Teilenr. vorliegen - SELECT 1 - FROM RD_COMP LEFT JOIN COMP - ON RD_COMP.ParentCode = COMP.PARENT - AND RD_COMP.ChildCode = COMP.KOMPTEILENR - AND RD_COMP.Positionsnummer = COMP.POSNR - - WHERE COMP.KOMPTEILENR IS NULL - AND RD_COMP.ParentCode = :parent - AND RD_COMP.ChildCode = :child - ) - -- Wenn beide Fallprüfungen zutreffend liegt die längenabhängige Teilenummer in beiden Stücklisten gleich oft vor und sie liegt in beiden - -- Stücklisten an den gleichen Positionsnummern - THEN 'True' - ELSE 'False' - END AS 'Fall 1.2' - """, - {"parent": self.parent_code, "child": child_code}).fetchone() - return True if res["Fall 1.2"] == 'True' else False - - def _get_update_data_for_diff_pos(self, pos_nr: str) -> list[dict[str]]: - """ - Ermittelt die Daten, die an die neue Position verschoben werden sollen. - (Für Fall 1.1 und Fall 1.3) - """ - res = self.db.con.execute(""" - SELECT C.ID, - C.DISPO_AKTIV_JN, - C.ZUSCHNITT, - C.RUESTCODE, - RD.Positionsnummer AS POSNR, - B.BEDINGUNG, - R.REGEL, - - -- Bestimmt Wert 'Abw. Anzahl': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Menge - END AS ABW_ANZAHL, - - -- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN RD.Menge - ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT,0)) / 1000 - END AS MENGE, - - -- Bestimmt Wert 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Uebergabelaenge - END AS LAENGE - - FROM RD_COMP RD - INNER JOIN COMP C - ON RD.ParentCode = C.PARENT - AND RD.ChildCode = C.KOMPTEILENR - AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) - - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID - - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID - - WHERE RD.ParentCode = :parent - AND RD.Positionsnummer = :posnr - """, - {"parent": self.parent_code, "posnr": pos_nr}).fetchone() - return res - - def _get_update_data_for_same_pos(self, pos_nr: str) -> list[dict[str]]: - """ - Ermittelt die Daten, die an dieselbe Position verschoben werden sollen - (Für Fall 1.2) - """ - res = self.db.con.execute(""" - SELECT C.ID, - C.ZUSCHNITT, - C.DISPO_AKTIV_JN, - C.RUESTCODE, - RD.Positionsnummer AS POSNR, - B.BEDINGUNG, - R.REGEL, - - -- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Menge - END AS ABW_ANZAHL, - - -- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN RD.Menge - ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT, 0)) / 1000 - END AS MENGE, - - -- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Uebergabelaenge - END AS LAENGE - - FROM RD_COMP RD - INNER JOIN COMP C - ON RD.ParentCode = C.PARENT - AND RD.ChildCode = C.KOMPTEILENR - AND RD.Positionsnummer = C.POSNR - - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID - - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID - - WHERE RD.ParentCode = :parent - AND RD.Positionsnummer = :posnr - """, - {"parent": self.parent_code, "posnr": pos_nr}).fetchone() - return res - - def _get_ids_to_del(self, ignored_ids: list = []) -> list[dict[str]]: - """ - Die zu löschenden ID's ergeben sich, indem man die ID's der Positionen aus der alten Stückliste betrachtet - und die ID's aus der neuen Stückliste abzieht, die verschoben werden. Positionen, die in der neuen Stückliste - neu dazukommen sind, wirken sich nicht auf die zu löschenden ID's aus! - """ - old_bom_ids = self.old_bom.get_child_ids(self.parent_code) - ids_to_delete = [idx for idx in old_bom_ids if idx["ID"] not in ignored_ids] - return ids_to_delete - - def _get_data_for_insert(self, posnr: str) -> list[dict[str]]: - """ - Fall 1.3 (Neuanlage von Positionen) - """ - res = self.db.con.execute(""" - SELECT RD.Positionsnummer AS POSNR, - RD.ChildCode AS KOMPTEILENR, - RD.ParentCode AS PARENT, - NULL AS ZUSCHNITT, - P.DISPO_AKTIV_JN, - P.RUESTCODE, - - -- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Menge - END AS ABW_ANZAHL, - - -- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN RD.Menge - ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT,0)) / 1000 - END AS MENGE, - - -- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV - CASE - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Uebergabelaenge - END AS LAENGE - - FROM RD_COMP RD - LEFT JOIN COMP C - ON RD.ChildCode = C.KOMPTEILENR - AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) - AND RD.ParentCode = C.PARENT - - -- Da es sich um neue Positionen handelt, werden Defaultwerte verwendet, die z.T. aus dem Teilestamm (PARTS) kommen - LEFT JOIN PARTS P - ON RD.ChildCode = P.TEILENR - - WHERE C.KOMPTEILENR IS NULL - AND RD.ParentCode = :parent - AND RD.Positionsnummer = :posnr - """, - {"parent": self.parent_code, "posnr": posnr}).fetchone() - return res - - def write_new_bom(self) -> None: - """ - Diese Methode binhaltet die Hauptlogik: - Hier wird festgelegt, welche Positionsnummern eingefügt, gelöscht oder verschoben werden müssen - Am Ende werden die daraus erforderlichen Datenbankaktionen (insert / update / delete) initiiert, - um die neue, fertig verarbeite Stückliste in COMP zu überführen. - """ - update_data_comp = [] - insert_data_comp = [] - update_data_regel = [] - update_data_bedingung = [] - ids_to_delete = [] - ids_not_to_delete = [] - childcodes_case_11 = set() # vermeidet erneute Prüfung einer Teilenummer durch _check_identical_code_len_combo - childcodes_case_12 = set() # vermeidet erneute Prüfung einer Teilenummer durch _check_same_pos() - - new_bom_data = self.new_bom.get_all_bom_data() - - for pos_data in new_bom_data: - - child_code: str = pos_data["ChildCode"] - posnr: int = pos_data["Positionsnummer"] - - # Prüfung Fall 1.1 - if child_code in childcodes_case_11 or self._check_identical_code_len_combo(child_code) is True: - logic_case = 1.1 - childcodes_case_11.add(child_code) - data = self._get_update_data_for_diff_pos(posnr) - - # Prüfung Fall 1.2 - elif child_code in childcodes_case_12 or self._check_same_pos(child_code) is True: - logic_case = 1.2 - childcodes_case_12.add(child_code) - data = self._get_update_data_for_same_pos(posnr) - - # Prüfungen Fall 1.3 - else: - - # Prüfung Fall 1.31 - Verschiebung bestehender Positionen - posnrs_to_move = self._get_positions_to_move_old_logic() - if posnr in posnrs_to_move: - logic_case = 1.31 - data = self._get_update_data_for_diff_pos(posnr) - - # Prüfung Fall 1.32 - Einfügen neuer Positionen - else: - logic_case = 1.32 - data = self._get_data_for_insert(posnr) - - # Teilenummern der Stückliste, die verschoben werden (Fal 1.1 / 1.2 / 1.3.1): - # - werden nie gelöscht (konkret: deren ID's in COMP), weil IMMER eine Verschiebung der Positionsnummer erfolgt - if logic_case in (1.1, 1.2, 1.31): - ids_not_to_delete.append(data["ID"]) - update_data_comp.append(data) - - if logic_case == 1.32: - insert_data_comp.append(data) - - - # Zu löschende Positionen ermitteln - ids_to_delete = self._get_ids_to_del(ignored_ids=ids_not_to_delete) - - # Positionen löschen (falls erforderlich) - if ids_to_delete: - self.new_bom.delete(ids_to_delete) - self.tos_bedingungen.delete(ids_to_delete) - self.tos_regeln.delete(ids_to_delete) - - # Positionen einfügen (falls erforderlich). Neue Positionen haben nie TOS-Regeln oder -Bedingungen - if insert_data_comp: - self.new_bom.insert(insert_data_comp) - - # Positionen sowie deren TOS-Regeln und -Bedingungen updaten (falls erforderlich) - if update_data_comp: - update_data_regel = self.extract_col_data(update_data_comp, "ID", "REGEL") - update_data_bedingung = self.extract_col_data(update_data_comp, "ID", "BEDINGUNG") - - self.new_bom.update(update_data_comp) - self.tos_regeln.update(update_data_regel) - self.tos_bedingungen.update(update_data_bedingung) - -if __name__ == "__main__": - main() \ No newline at end of file