import argparse import csv import os import sqlite3 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") parser.add_argument("-bom", action="store_true", help="Ausgabe der neuen, aufbereiteten Stückliste") parser.add_argument("-av", action="store_true", help="Ausgabe der AV-Workflowmeldungen (falls vorhanden)") parser.add_argument("-tos", action="store_true", help="Ausgabe der TOS-Workflowmeldungen (falls vorhanden)") 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 über die Flags -bom / -av / -tos ausgewählten Elemente in der Konsole") sp_csv = subparsers.add_parser("csv", help="Ausgabe der über die Flags -bom / -av / -tos ausgewählten Elemente als CSV-Datei(en)") 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") 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"]) av_workflow_msg = AVWorkflowMessage(db, args["parent_code"]) tos_workflow_msg = TOSWorkflowMessage(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) # Positionen für AV- bzw. TOS-Workflowmeldungen generieren (muss vor dem Schreiben der neuen Stücklist erfolgen!) av_workflow_msg.get_av_workflow_posnrs_and_codes() tos_workflow_msg.get_tos_workflow_posnrs_and_codes() # Neue Stückliste erstellen / schreiben bom_processing.write_new_bom() # Ausgabe in Konsole if args["output_type"] == "cmd": if args["bom"]: if args["verbose"]: print(repr(new_bom)) else: print(new_bom) if args["av"]: print(av_workflow_msg) if args["tos"]: print(tos_workflow_msg) # Ausgabe als CSV-Datei(en) elif args["output_type"] == "csv": filedir = args["out_dir"] filenames = [] file_data = [] # CSV-Ausgabe der aufbereiteten Stückliste if args["bom"]: if args["verbose"]: bom_data = new_bom.get_verbose_data() else: bom_data = new_bom.get_compact_data() filenames.append(f"BOM_{args["parent_code"]}_new.csv") file_data.append(bom_data) # CSV-Ausgabe der Positionsnummern aus der alten Stückliste, die Bestandteil der AV-Workflowmeldung sind if args["av"]: msg_data = av_workflow_msg.get_data() filenames.append(f"AV_{args["parent_code"]}.csv") file_data.append(msg_data) # CSV-Ausgabe der Positionsnummern aus der alten Stückliste, die Bestandteil der TOS-Workflowmeldung sind if args["tos"]: msg_data = tos_workflow_msg.get_data() filenames.append(f"TOS_{args["parent_code"]}.csv") file_data.append(msg_data) for name, data in zip(filenames, file_data): csv_handler.write_from_dict(data=data, filedir=filedir, filename=name) 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.LFD_NR, B.UEB_LEISTEN_NR FROM COMP C LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON C.UEBTEILENR = B.UEBTEILENR AND C.POSNR = B.POSNR LEFT JOIN STKL_REGEL R ON C.UEBTEILENR = R.UEBTEILENR AND C.POSNR = R.POSNR WHERE C.UEBTEILENR = :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.LFD_NR, B.UEB_LEISTEN_NR FROM COMP C LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON C.UEBTEILENR = B.UEBTEILENR AND C.POSNR = B.POSNR LEFT JOIN STKL_REGEL R ON C.UEBTEILENR = R.UEBTEILENR AND C.POSNR = R.POSNR WHERE C.UEBTEILENR = :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 Teilenummern, 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 komplexer wird. Die Prüfung läuft so ab: 1. Teilenr-Längen-Kombinationen aus der alten und neuen Stückliste mit UNION ALL verschmelzen 2. Mit COUNT ermitteln, wie oft jede Teilenummer-Längen-Kombination in der verschmolzenen Stückliste auftritt 3. Die Teilenummern in die temporäre Tabelle TMP_CODES_CASE_12_OR_13 schreiben, sobald 1x eine Teilenr-Längen-Kombination auftritt, die nur 1x vorkommt => 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 einer Teilenummer 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)") self.db.con.execute(""" WITH combined_bom AS ( SELECT C.KOMPTEILENR, COALESCE(C.LAENGE, 0) AS LAENGE FROM COMP C WHERE C.UEBTEILENR = :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 Fall 1.32 auf diese Positionen zutreffen würde. Die Prüfung läuft so ab: 1. Alle Teilenr-Posnr-Kombinationen aus der alten und neuen Stückliste mit UNION ALL verschmelzen, 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, deren Teilenr-Posnr-Kombination nur 1x auftritt, 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 einer Teilenummer 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)") self.db.con.execute(""" WITH combined_bom AS ( SELECT KOMPTEILENR, POSNR FROM COMP WHERE UEBTEILENR = :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 folgende Spaltenwerte ermitteln und in die temporäre Tabelle TMP_POSNRS_CASE_11 schreiben: - Teilenr des Parents - Positionsnummer laut alter Stückliste - Positionsnummer laut neuer Stückliste Hinweis: - Die Positionsnummern aus beiden Stücklisten werden benötigt, um bei der Ermittlung der Update-Daten die Zuordnung zwischen der alten und neuen Stückliste herstellen zu können - Die Teilenummer des Parents wird zusamment mit der Positionsnummer laut alter Stückliste dafür benötigt, um in den Tabellen STKL_REGEL und VAR_TEIL_BEDINGUNG_2 die Zeilen zu ermitteln, die aktualisiert werden müssen """ self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_11") self.db.con.execute(""" CREATE TABLE TMP_POSNRS_CASE_11 (UEBTEILENR TEXT NOT NULL, POS_ALT INTEGER NOT NULL, POS_NEU INTEGER NOT NULL) """) self.db.con.execute(""" INSERT INTO TMP_POSNRS_CASE_11 (UEBTEILENR, POS_ALT, POS_NEU) SELECT C.UEBTEILENR, C.POSNR AS POS_ALT, RD.Positionsnummer AS POS_NEU 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.UEBTEILENR 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 folgende Spaltenwerte ermitteln und in die temporäre Tabelle TMP_POSNRS_CASE_12 schreiben: - Teilenr des Parents - Positionsnummer laut alter Stückliste - Positionsnummer laut neuer Stückliste Hinweis: - Die Positionsnummern aus beiden Stücklisten werden benötigt, um bei der Ermittlung der Update-Daten die Zuordnung zwischen der alten und neuen Stückliste herstellen zu können - Die Teilenummer des Parents wird zusamment mit der Positionsnummer laut alter Stückliste dafür benötigt, um in den Tabellen STKL_REGEL und VAR_TEIL_BEDINGUNG_2 die Zeilen zu ermitteln, die aktualisiert werden müssen """ self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_12") self.db.con.execute(""" CREATE TABLE TMP_POSNRS_CASE_12 (UEBTEILENR TEXT NOT NULL, POS_ALT INTEGER NOT NULL, POS_NEU INTEGER NOT NULL) """) self.db.con.execute(""" INSERT INTO TMP_POSNRS_CASE_12 (UEBTEILENR, POS_ALT, POS_NEU) SELECT C.UEBTEILENR, C.POSNR AS POS_ALT, RD.Positionsnummer AS POS_NEU FROM RD_COMP RD LEFT JOIN TMP_CODES_CASE_13 TMP ON RD.ChildCode = TMP.KOMPTEILENR LEFT JOIN COMP C ON RD.ParentCode = C.UEBTEILENR 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 POS_NEU FROM TMP_POSNRS_CASE_11) """, {"parent_code": self.parent_code}) def _get_case_131_positions(self): """ Fall 1.31 - Positionsnummern ermitteln, die neu angelegt werden müssen. Deren Ermittlung funktioniert so: 1. Alle Positionsnummern herausfiltern, auf die folgende Kriterien zutreffen: - Teilenr-Längen-Kombination der Position existiert nur in der neuen Stückliste existiert - Teilenummer der Position ist in der Tabelle TMP_CODES_CASE_13 enthalten Hinweis: Es reicht nicht aus, lediglich zu prüfen, ob die Teilenr-Längen-Kombination nur in der neuen Stückliste 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. Alle gefundenen Positionen in die temporäre Tabelle TMP_POSNRS_CASE_131 schreiben (mehr Informationen werden nicht benötigt) """ 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)") 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.UEBTEILENR AND RD.ChildCode = C.KOMPTEILENR AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) WHERE RD.ParentCode = :parent_code AND RD.ChildCode IN (SELECT KOMPTEILENR 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. Alle Positionsnummern herausfiltern, auf die folgende Kriterien zutreffen: - Teilenr-Längen-Kombination existiert in beiden Stücklisten - Teilenummer der Position ist in der Tabelle TMP_CODES_CASE_13 enthalten Hinweis: Es reicht nicht aus, lediglich zu prüfen, ob die Teilenr-Längen-Kombination in beiden Stücklisten vorliegt 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 Prioritätsprinzip, 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 folgende Spaltenwerte ermitteln und in die temporäre Tabelle TMP_POSNRS_CASE_132 schreiben: - Teilenr des Parents - Positionsnummer laut alter Stückliste - Positionsnummer laut neuer Stückliste Hinweis: - Die Positionsnummern aus beiden Stücklisten werden benötigt, um bei der Ermittlung der Update-Daten die Zuordnung zwischen der alten und neuen Stückliste herstellen zu können - Die Teilenummer des Parents wird zusamment mit der Positionsnummer laut alter Stückliste dafür benötigt, um in den Tabellen STKL_REGEL und VAR_TEIL_BEDINGUNG_2 die Zeilen zu ermitteln, die aktualisiert werden müssen """ self.db.con.execute("DROP TABLE IF EXISTS TMP_POSNRS_CASE_132") self.db.con.execute(""" CREATE TABLE TMP_POSNRS_CASE_132 (UEBTEILENR TEXT NOT NULL, POS_ALT INTEGER NOT NULL, POS_NEU INTEGER NOT NULL) """) self.db.con.execute(""" INSERT INTO TMP_POSNRS_CASE_132 (UEBTEILENR, POS_ALT, POS_NEU) SELECT C.UEBTEILENR, C.POSNR AS POS_ALT, RD.Positionsnummer AS POS_NEU FROM RD_COMP RD INNER JOIN COMP C ON RD.ParentCode = C.UEBTEILENR AND RD.ChildCode = C.KOMPTEILENR AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) WHERE RD.ParentCode = :parent_code AND RD.ChildCode IN (SELECT KOMPTEILENR FROM TMP_CODES_CASE_13) """, {"parent_code": self.parent_code}) def _get_ids_to_delete(self) -> None: """ Ermittelt alle Positionsnummern, die entfernt werden sollen Hinweis: Bei der Ermittlung 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" ist daher 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, darf also 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 (COMP_ROWID INTEGER NOT NULL, VAR_TEIL_BED_2_ROWID INTEGER, STKL_REGEL_ROWID INTEGER) """) 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 (COMP_ROWID, VAR_TEIL_BED_2_ROWID, STKL_REGEL_ROWID) SELECT C.ROWID AS COMP_ROWID, B.ROWID AS VAR_TEIL_BED_2_ROWID, R.ROWID AS STKL_REGEL_ROWID FROM COMP C LEFT JOIN COMBINED_UPDATE_DATA CUD ON C.UEBTEILENR = CUD.UEBTEILENR AND C.POSNR = CUD.POS_ALT LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON C.UEBTEILENR = B.UEBTEILENR AND C.POSNR = B.POSNR LEFT JOIN STKL_REGEL R ON C.UEBTEILENR = R.UEBTEILENR AND C.POSNR = R.POSNR WHERE C.UEBTEILENR = :parent_code AND CUD.UEBTEILENR IS NULL """, {"parent_code": self.parent_code}) def _get_update_data(self): """ Ermittlung der neuen Stücklisteninformationen, mit denen die bestehenden Datzensätze überschrieben werden Die neuen Informationen werden für alle Positionsnummern bestimmt, die unter die Fälle 1.1, 1.2 und 1.32 fallen, also die, die sich in den Tabellen TMP_POSNRS_CASE_11, TMP_POSNRS_CASE_12 & TMP_POSNRS_CASE_132 befinden. Hinweis: Es werden in diesem Zuge auch die zugehörigen Zeilen-ID's zu den Positionsnummern in den Tabellen VAR_TEIL_BEDINGUNG_2 und STKL_REGEL ermittelt, da in diesen Tabellen ebenfalls die Positionsnummern aktualisiert werden müssen, sofern welche vorliegen. Die Regeln und Bedingungen selbst bleiben unverändert, da sie nur verschoben werden. """ self.db.con.execute("DROP TABLE IF EXISTS TMP_UPDATE_DATA") self.db.con.execute(""" CREATE TABLE TMP_UPDATE_DATA (COMP_ROWID INTEGER NOT NULL, VAR_TEIL_BED_2_ROWID INTEGER, STKL_REGEL_ROWID INTEGER, POS_ALT INTEGER NOT NULL, POS_NEU INTEGER NOT NULL, ABW_ANZAHL INTEGER, MENGE REAL, LAENGE REAL, ZUSCHNITT REAL, DISPO_AKTIV_JN TEXT, RUESTCODE TEXT) """) 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 (COMP_ROWID, VAR_TEIL_BED_2_ROWID, STKL_REGEL_ROWID, POS_ALT, POS_NEU, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE) SELECT C.ROWID AS COMP_ROWID, B.ROWID AS VAR_TEIL_BED_2_ROWID, R.ROWID AS STKL_REGEL_ROWID, P.POS_ALT, RD.Positionsnummer AS POS_NEU, -- 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 FROM ALL_POSNRS_TO_UPDATE P INNER JOIN COMP C ON P.UEBTEILENR = C.UEBTEILENR AND P.POS_ALT = C.POSNR INNER JOIN RD_COMP RD ON P.UEBTEILENR = RD.ParentCode AND P.POS_NEU = RD.Positionsnummer LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON P.UEBTEILENR = B.UEBTEILENR AND P.POS_ALT = B.POSNR LEFT JOIN STKL_REGEL R ON P.UEBTEILENR = R.UEBTEILENR AND P.POS_ALT = R.POSNR """) def delete(self): """ - Entfernt die Positionsnummern aus COMP, die nicht mehr in der neuen Stückliste vorliegen - Entfernt die TOS-Regeln und -Bedingungen der Positionsnummern aus VAR_TEIL_BEDINGUNG_2 und STKL_REGEL """ # Stücklistenpositionen entfernen self.db.con.execute("DELETE FROM COMP WHERE ROWID IN (SELECT COMP_ROWID FROM TMP_IDS_TO_DELETE)") # TOS-Bedingungen entfernen self.db.con.execute("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE ROWID IN (SELECT VAR_TEIL_BED_2_ROWID FROM TMP_IDS_TO_DELETE)") # TOS-Regeln entfernen self.db.con.execute("DELETE FROM STKL_REGEL WHERE ROWID IN (SELECT STKL_REGEL_ROWID FROM TMP_IDS_TO_DELETE)") def update(self): """ Aktualisiert die Positionsnummern, deren Spaltenanpassungen und TOS-Regeln / -Bedingungen von der alten in die neue Stückliste übernommen werden sollen. """ # Stücklisteninformationen updaten self.db.con.execute(""" UPDATE COMP SET POSNR = UD.POS_NEU, 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 ROWID = UD.COMP_ROWID """) # TOS-Bedingungen updaten self.db.con.execute(""" UPDATE VAR_TEIL_BEDINGUNG_2 SET POSNR = UD.POS_NEU FROM TMP_UPDATE_DATA UD WHERE ROWID = UD.VAR_TEIL_BED_2_ROWID """) # TOS-Regeln updaten self.db.con.execute(""" UPDATE STKL_REGEL SET POSNR = UD.POS_NEU FROM TMP_UPDATE_DATA UD WHERE ROWID = UD.STKL_REGEL_ROWID """) def insert(self): """ Neuanlage aller Positionen, die nur in der neuen Stückliste vorliegen (das sind die aus der Tabelle TMP_POSNRS_CASE_131) 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 dieses Scripts nicht klar. """ self.db.con.execute(""" INSERT INTO COMP (UEBTEILENR, POSNR, KOMPTEILENR, ABW_ANZAHL, MENGE, LAENGE, ZUSCHNITT, DISPO_AKTIV_JN, RUESTCODE) SELECT RD.ParentCode AS UEBTEILENR, 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.UEBTEILENR LEFT JOIN PARTS P ON RD.ChildCode = P.TEILENR WHERE C.KOMPTEILENR IS NULL AND RD.ParentCode = :parent_code AND RD.Positionsnummer IN (SELECT POSNR 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._get_ids_to_delete() self._get_update_data() self.delete() self.insert() self.update() self.drop_tmp_tables() class AVWorkflowMessage: def __init__(self, db: Database, parent_code: str): self.db = db self.parent_code = parent_code def __str__(self): res = self.get_data() msg = self._create_msg(res) return msg def drop_tmp_table(self): self.db.con.execute("DROP TABLE TMP_AV_WORKFLOW_MESSAGE") def get_av_workflow_posnrs_and_codes(self): """ Wenn Positionsnummern folgende Kriterien erfüllen, sollen Sie in einer Workflowmeldung an die AV-Verantwortlichen ausgegeben werden: Kriterien: WENN die Positionen zu einer längenabhängigen Teilenummer MIT Sägeliste gehören (d.h. Haken bei 'Halbzeug ohne Sägeliste' im Teilestamm ist nicht gesetzt) UND diese längenabhängige Teilenummer in der alten und der neuen Stückliste vertreten ist UND die Länge-Teilenummer-Kombination dieser Positionsnummer nur in der alten Stückliste vorliegt (aber nicht in der Neuen) UND bei der Position mindestens eine der nachfolgenden Kriterien erfüllt ist: - der Wert in der Spalte 'Rüstcode' weicht vom Rüstcode-Wert ab, den die zugehörige Teilenummer im Teilstamm aufweist - in der Spalte 'Dispoaktiv' liegt kein Haken vor - in der Spalte 'Zuschnitt' ist ein Wert definiert """ self.db.con.execute("DROP TABLE IF EXISTS TMP_AV_WORKFLOW_MESSAGE") self.db.con.execute(""" CREATE TABLE TMP_AV_WORKFLOW_MESSAGE (POSNR INTEGER, KOMPTEILENR TEXT) """) test = self.db.con.execute(""" --längenabhängige Teilenummern ermitteln, die in beiden Stücklisten vorliegen WITH laengenabh_codes_in_both_boms AS ( SELECT DISTINCT ChildCode FROM RD_COMP WHERE ParentCode = :parent_code AND ChildCode in ( SELECT DISTINCT KOMPTEILENR FROM COMP WHERE UEBTEILENR = :parent_code AND COALESCE(LAENGE, 0) != 0 ) ) INSERT INTO TMP_AV_WORKFLOW_MESSAGE (POSNR, KOMPTEILENR) SELECT C.POSNR, C.KOMPTEILENR FROM COMP C LEFT JOIN RD_COMP RD ON C.UEBTEILENR = RD.ParentCode AND C.KOMPTEILENR = RD.ChildCode AND COALESCE(C.LAENGE, 0) = RD.Uebergabelaenge INNER JOIN PARTS P ON C.KOMPTEILENR = P.TEILENR WHERE C.UEBTEILENR = :parent_code AND C.KOMPTEILENR IN laengenabh_codes_in_both_boms -- Nur längenabhängige Teilenummern berücksichtigen, die in beiden Stücklisten vorliegen AND RD.ParentCode IS NULL -- Nur Positionen berücksichtigen, deren Länge ausschließlich in der alten Stückliste vorliegt AND P.INFO2_JN = "N" -- Nur Positionen berücksichtigen, deren zugehörige Teilenummer keinen Haken bei "Halbzeug ohne Sägeliste" im Teilestamm aufweist AND ( -- Nur Positionen berücksichtigen, auf die min. eines der nachfolgenden Kriterien zutrifft: C.ZUSCHNITT IS NOT NULL -- Zuschnittswert liegt vor OR C.RUESTCODE != P.RUESTCODE -- Rüstcode der Position in der Stammstückliste ist abweichend zum Rüstcode der zugehörigen Teilenummer im Teilestamm OR C.DISPO_AKTIV_JN = "N" -- Stücklistenpositionen besitzt keinen Haken in der Spalte "Dispoaktiv" ) ORDER BY C.POSNR """, {"parent_code": self.parent_code}).fetchall() return test def get_data(self) -> list[dict[str]]: res = self.db.con.execute("SELECT POSNR, KOMPTEILENR FROM TMP_AV_WORKFLOW_MESSAGE").fetchall() return res def _create_msg(self, data: list[dict[str]] | None) -> str: if not data: return "\nAV-Meldung:\nKeine" affected_positions_str = ", ".join([f"{pos["POSNR"]} - {pos["KOMPTEILENR"]}" for pos in data]) return f"\nAV-Meldung:\nBitte Stückliste {self.parent_code} prüfen. Vor der Änderung lagen bei folgenden Positionen manuelle Eintragungen / Änderungen in den Spalten „Zuschnitt“, „Rüstcode“ oder „Dispoaktiv“ vor: {affected_positions_str}" class TOSWorkflowMessage: def __init__(self, db: Database, parent_code: str): self.db = db self.parent_code = parent_code def __str__(self): res = self.get_data() msg = self._create_msg(res) return msg def drop_tmp_table(self): self.db.con.execute("DROP TABLE TMP_TOS_WORKFLOW_MESSAGE") def get_tos_workflow_posnrs_and_codes(self) -> None: """ Wenn Positionsnummern in der Stückliste TOS-Regeln oder -Bedingungen aufweisen, sollen Sie in einer Workflowmeldung an die TOS-Verantwortlichen ausgegeben werden. """ self.db.con.execute("DROP TABLE IF EXISTS TMP_TOS_WORKFLOW_MESSAGE") self.db.con.execute(""" CREATE TABLE TMP_TOS_WORKFLOW_MESSAGE (POSNR INTEGER, KOMPTEILENR TEXT) """) self.db.con.execute(""" INSERT INTO TMP_TOS_WORKFLOW_MESSAGE (POSNR, KOMPTEILENR) SELECT C.POSNR, C.KOMPTEILENR FROM COMP C LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON C.UEBTEILENR = B.UEBTEILENR AND C.POSNR = B.POSNR LEFT JOIN STKL_REGEL R ON C.UEBTEILENR = R.UEBTEILENR AND C.POSNR = R.POSNR WHERE C.UEBTEILENR = :parent_code AND ( B.UEB_LEISTEN_NR IS NOT NULL OR R.LFD_NR IS NOT NULL ) ORDER BY C.POSNR """, {"parent_code": self.parent_code}) def get_data(self) -> list[dict[str]]: res = self.db.con.execute("SELECT POSNR, KOMPTEILENR FROM TMP_TOS_WORKFLOW_MESSAGE").fetchall() return res def _create_msg(self, data: list[dict[str]] | None) -> str: if not data: return "\nTOS-Meldung:\nKeine" affected_positions_str = ", ".join([f"{pos["POSNR"]} - {pos["KOMPTEILENR"]}" for pos in data]) return f"\nTOS-Meldung:\nBitte Stückliste {self.parent_code} prüfen. Vor der Änderung lagen bei folgenden Positionen TOS-Bedingungen vor: {affected_positions_str}" if __name__ == "__main__": main()