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()