From 727f629e76edd2d765515f6c327527d2f9b66cdb Mon Sep 17 00:00:00 2001 From: "s.steuer" Date: Mon, 11 Aug 2025 02:10:05 +0200 Subject: [PATCH] =?UTF-8?q?Komplett=20=C3=BCberarbeitet=20und=20in=20Klass?= =?UTF-8?q?en=20aufgeteilt,=20um=20Verantwortlichkeiten=20besser=20aufteil?= =?UTF-8?q?en=20zu=20k=C3=B6nnen.=20Beachte=20-=20Die=20Verarbeitungslogik?= =?UTF-8?q?=20der=20neuen=20St=C3=BCckliste=20(die=20nun=20in=20der=20Klas?= =?UTF-8?q?se=20BOMProcessing=20und=20haupts=C3=A4chlich=20in=20der=20dari?= =?UTF-8?q?n=20enthaltenen=20Methode=20'write=5Fnew=5Fbom()'=20befindet)?= =?UTF-8?q?=20hat=20sich=20durch=20das=20Refactoring=20nicht=20ge=C3=A4nde?= =?UTF-8?q?rt!?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/verschiebelogik.py | 1076 +++++++++++++++++++++++----------------- 1 file changed, 631 insertions(+), 445 deletions(-) diff --git a/lib/verschiebelogik.py b/lib/verschiebelogik.py index e0f78fb..49f8dda 100644 --- a/lib/verschiebelogik.py +++ b/lib/verschiebelogik.py @@ -2,508 +2,694 @@ import argparse import csv import os import sqlite3 -from contextlib import closing -def main(asm_partnum, sql_filedir, csv_filedir): +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()) - build_sivas_from_file(sql_filedir) - csv_data = read_csv(csv_filedir, asm_partnum) - csv_to_db(csv_data) - create_new_bom(asm_partnum) - # print_new_bom(asm_partnum) - write_bom_as_csv(asm_partnum) + 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 csv_to_db(csv_data: list[dict[str, str]]) -> None: - """Schreibt neue Stücklistendaten aus CSV in temporäre Datenbanktabelle.""" + def check_filepath(self, filepath: str) -> str: + if not os.path.isfile(filepath): + raise FileNotFoundError("Angegebener Dateiname existiert nicht!") + return filepath - 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)) - """) +class CSVHandler(FileHandler): - for line in csv_data: - con.execute(""" INSERT INTO RD_COMP (ParentCode, Positionsnummer, ChildCode, Menge, Uebergabelaenge) - VALUES (:ParentCode, :Positionsnummer, :ChildCode, :Menge, :Uebergabelaenge)""", line) + def __init__(self, encoding="utf-8", delimiter: str = ";"): + self.delimiter = delimiter + self.encoding = encoding -def dict_factory(cursor, row): - d = {} - for idx, col in enumerate(cursor.description): - d[col[0]] = row[idx] - return d + def read_as_dict(self, filename: str, filedir: str = os.environ.get("PROJECT_DATA"), ) -> list[dict[str]]: -def read_csv(filedir: str, partnumber: str) -> list[dict[str, str]]: - csv_path = os.path.join(filedir, f"BOM_{partnumber}.csv") - if not csv_path: - raise FileNotFoundError + 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) - with open(csv_path, "r", encoding="utf-8", newline="") as csv_file: - reader = csv.DictReader(csv_file, delimiter=";") - csv_data = [] - for line in reader: - line["Positionsnummer"] += "0" # Angleichung Positionsnummer an SIVAS-Format (Pos. * 10) - csv_data.append(line) + 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 - return csv_data + def close(self): + self.con.close() + + def remove(self): + if os.path.isfile(self.db_path): + os.remove(self.db_path) -def build_sivas_from_file(sql_filedir: str, sql_filename="sivas_replika.sql") -> None: - filepath = os.path.join(sql_filedir, sql_filename) - if not filepath: - raise FileNotFoundError +class RD_COMP: - with open(filepath, "r", encoding="utf-8", newline="") as sql_file: - sql = sql_file.read() - - con.executescript(sql) - -def get_new_bom_data(parent_partnum:str) -> list[dict[str, str]]: - """ Positionsnummern und zugehörige Teilenummern aus neuer Stückliste holen""" - res = con.execute("SELECT * FROM RD_COMP WHERE ParentCode = :parent ORDER BY Positionsnummer", {"parent": parent_partnum}).fetchall() - return res - -def get_old_bom_ids(parent_partnum: str) -> list | None: - """Gibt ID's der Datensätze aus der alten Stückliste zurück""" - res = con.execute("SELECT ID FROM COMP WHERE PARENT = :parent", {"parent": parent_partnum}).fetchall() - return res - -def get_positions_to_move(parent_partnum: str) -> set | None: """ - Gibt alle Positionsnummern zurück, die gemäß alter Logik verschoben werden + 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. """ - res = 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) + 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 + - WHERE RD.ParentCode = :parent - """, - {"parent": parent_partnum}).fetchall() - if res is not None: - res = set([line["POS"] for line in res]) - return res + 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: -def identical_occurences_in_both_boms(parent_partnum: str, child_id: str) -> bool: """ - Prüfung Fall 1.1: + Beinhaltet Methoden, zur Interaktion mit Tabelle VAR_TEIL_BEDINGUNG_2 (gleichnamig wie in SIVAS), + in der TOS-Bedingungen gespeichert sind """ - res = 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 + 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) - 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": parent_partnum, "child": child_id}).fetchone() + def delete(self, data: list[dict[str]]) -> None: + self.db.con.executemany("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID = :ID", data) - return True if res["Fall 1.1"] == 'True' else False -def occurences_at_same_pos_in_both_boms(parent_partnum: str, child_id: str) -> bool: +class STKL_REGEL: + """ - Prüfung Fall 1.2 + Beinhaltet Methoden, zur Interaktion mit Tabelle STKL_REGEL (gleichnamig wie in SIVAS), + in der TOS-Regeln gespeichert sind """ - res = 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 + 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) - 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": parent_partnum, "child": child_id}).fetchone() - return True if res["Fall 1.2"] == 'True' else False + def delete(self, data: list[dict[str]]) -> None: + self.db.con.executemany("DELETE FROM STKL_REGEL WHERE COMP_ID = :ID", data) + +class BOMProcessing: -def get_data_for_new_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]: """ - Fall 1.3 (Neuanlage von Positionen) + Die Methoden dieser Klasse verarbeiten die Daten aus der RuleDesigner-Stücklisten CSV und initiiert das Schreiben der verarbeiteten + Daten in COMP. """ - res = con.execute(""" - SELECT RD.Positionsnummer AS POSNR, - RD.ChildCode AS KOMPTEILENR, - RD.ParentCode AS PARENT, - CASE -- Abw. Anz. = NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV - WHEN RD.Uebergabelaenge = 0 THEN NULL - ELSE RD.Menge - END AS ABW_ANZAHL, - CASE -- Mengenwert berechnen, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV - WHEN RD.Uebergabelaenge = 0 THEN RD.Menge - ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT,0)) / 1000 - END AS MENGE, - CASE -- LAENGE = NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV - 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 - - -- Da die Positionen neu angelegt werden, werden die Defaultwerte verwendet, die zum Teil aus dem Teilestamm (PARTS) kommen. - LEFT JOIN PARTS P - ON RD.ChildCode = P.TEILENR + 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 - WHERE C.KOMPTEILENR IS NULL - AND RD.ParentCode = :parent - AND RD.Positionsnummer = :posnr - """, - {"parent": parent_partnum, "posnr": pos}).fetchone() - return res + 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 -def get_update_data_for_diff_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]: - """ - Ermittelt die Daten, die an die neue Position verschoben werden sollen. - (Für Fall 1.1 und Fall 1.3) - """ - res = 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 + 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) + AND RD.Uebergabelaenge = COALESCE(C.LAENGE,0) - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID + WHERE RD.ParentCode = :parent + """, {"parent": self.parent_code}).fetchall() - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID + if res is not None: + res = set([line["POS"] for line in res]) + return res - WHERE RD.ParentCode = :parent - AND RD.Positionsnummer = :posnr - """, - {"parent": parent_partnum, "posnr": pos}).fetchone() - 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) -def get_update_data_for_same_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]: - """ - Ermittelt die Daten, die an dieselbe Position verschoben werden sollen - (Für Fall 1.2) - """ - res = con.execute(""" - SELECT C.ID, - C.ZUSCHNITT, - C.DISPO_AKTIV_JN, - C.RUESTCODE, - RD.Positionsnummer AS POSNR, - B.BEDINGUNG, - R.REGEL, + 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 - -- 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, + 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() - -- 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, + return True if res["Fall 1.1"] == 'True' else False - -- 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 + 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 - FROM RD_COMP RD - INNER JOIN COMP C - ON RD.ParentCode = C.PARENT - AND RD.ChildCode = C.KOMPTEILENR - AND RD.Positionsnummer = C.POSNR + 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 - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID + 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 - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID + 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, - WHERE RD.ParentCode = :parent - AND RD.Positionsnummer = :posnr - """, - {"parent": parent_partnum, "posnr": pos}).fetchone() - return res + -- 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, -def create_new_bom(parent_partnum:str) -> None: + -- 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, - new_bom_data = get_new_bom_data(parent_partnum) - - # Ermittlung, welche Teilenummern unter Fall 1.1 , 1.2 oder 1.3 fallen - # Nicht im folgenden Loop, um mehrfache Fallprüfung 1.1 und 1.2 zu vermeiden (wenn längenabhängige Teilenummern mehrfach in Stückliste vorliegen) - partnum_cases = {} - for result_line in new_bom_data: - child_partnum = result_line["ChildCode"] - if child_partnum not in partnum_cases: - if identical_occurences_in_both_boms(parent_partnum, child_partnum): - partnum_cases[child_partnum] = 1.1 - elif occurences_at_same_pos_in_both_boms(parent_partnum, child_partnum): - partnum_cases[child_partnum] = 1.2 - else: - partnum_cases[child_partnum] = 1.3 - - # Ermittlung unter welchen Fall der Logik, die jeweilige Positionsnummer der neuen Stückliste fällt - pos_to_update = [] # Sammelt Positionsnummern (inkl. Daten), die upgedated werden sollen - pos_to_insert = [] # Sammelt Positionsnummern (inkl. Daten), die eingefügt werden sollen - pos_to_del = [] # Sammelt Positionsnummern, die gelöscht werden sollen - moved_partnums = [] # Sammelt Positionsnummern, die verschoben werden. Dient später zur Ermittlung der zu löschenden Positionen - - for pos_data in new_bom_data: - child_partnum = pos_data["ChildCode"] - child_pos = pos_data["Positionsnummer"] - - # Fall 1.1 (Kann nicht mit 1. Fall in Fall 1.3.1 (siehe unten) fusioniert werden, weil sonst z.B. bei Beispiel 3 fälschlicherweise diese Logik greifen würde) - if partnum_cases[child_partnum] == 1.1: - data = get_update_data_for_diff_pos(parent_partnum, child_pos) - pos_to_update.append(data) - moved_partnums.append(data["ID"]) - - # Fall 1.2 - elif partnum_cases[child_partnum] == 1.2: - data = get_update_data_for_same_pos(parent_partnum, child_pos) - pos_to_update.append(data) - moved_partnums.append(data["ID"]) - - # Fall 1.3 - elif partnum_cases[child_partnum] == 1.3: - - # Fall 1.3.1 - Verschiebung von Positionen - pos_to_move = get_positions_to_move(parent_partnum) - if child_pos in pos_to_move: - data = get_update_data_for_diff_pos(parent_partnum, child_pos) - pos_to_update.append(data) - moved_partnums.append(data["ID"]) - - # Fall 1.3.2 - Neuanlage von Positionen - else: - data = get_data_for_new_pos(parent_partnum, child_pos) - pos_to_insert.append(data) - else: - raise NotImplementedError("Fall ist nicht vorgesehen (dürfte laut spezifizierter Logik nicht auftreten). Prüfung erforderlich!") - - # Positionen aus der alten Stückliste, die nicht verschoben werden müssen gelöscht werden. - # Positionen, die in der neuen Stückliste neu eingefügt werden wirken sich nicht auf die Positionen der alten Stückliste aus und bleiben - # deshalb bei der Bestimmung der zu löschenden Positionen unberücksichtigt. - old_bom_ids = get_old_bom_ids(parent_partnum) - for idx in old_bom_ids: - if idx["ID"] not in moved_partnums: - pos_to_del.append(idx) - - # Datenbankaktionen durchführen, um die neue Stückliste zu schreiben. - if pos_to_del: - delete_pos(pos_to_del) - if pos_to_insert: - insert_pos(pos_to_insert) - if pos_to_update: - update_pos(pos_to_update) - -def insert_pos(new_data: dict[str, str]) -> None: - """ - Fügt die Positionen in die Stücklist ein, bei denen keine Verschiebung von der alten Stückliste erfolgt ist. - Diese Positionen werden stets ohne TOS-Regeln oder -Bedingungen angelegt. - """ - 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_pos(ids: list[dict[str, str]]) -> None: - """ - - Entfernt die Positionen aus der Stückliste (COMP), die nicht verschoben werden und nicht mehr in der alten Stückliste vorliegen. - - Entfernt die TOS-Regeln aus STKL_REGEL und TOS-Bedingungen aus VAR_TEIL_BEDINGUNG_2 zu den oben genannten Positionen. - """ - con.executemany("DELETE FROM COMP WHERE ID = :ID", ids) - con.executemany("DELETE FROM VAR_TEIL_BEDINGUNG_2 WHERE COMP_ID = :ID", ids) - con.executemany("DELETE FROM STKL_REGEL WHERE COMP_ID = :ID", ids) - -def update_pos(new_data: list[dict[str, str]]) -> None: - """ - - Verschiebt Positionsdaten von einer Position in der alten Stückliste zu einer Position in der neuen Stückliste - - Aktualisiert TOS-Regeln in STKL_REGEL und - """ - 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 - """, new_data) - - update_bedingungen = [pos_data for pos_data in new_data if pos_data["BEDINGUNG"] is not None] - update_regeln = [pos_data for pos_data in new_data if pos_data["REGEL"] is not None] - - if update_bedingungen: - con.executemany("UPDATE VAR_TEIL_BEDINGUNG_2 SET BEDINGUNG = :BEDINGUNG WHERE COMP_ID = :ID", update_bedingungen) - if update_regeln: - con.executemany("UPDATE STKL_REGEL SET REGEL = :REGEL WHERE COMP_ID = :ID", update_regeln) - -def print_new_bom(parent_partnum: str) -> None: - res = con.execute(""" - SELECT C.POSNR AS Pos, - C.KOMPTEILENR AS 'Art.-Nr.', - C.MENGE As Menge, - C.ABW_ANZAHL AS 'Abw. Anzahl', - C.LAENGE AS Länge, - C.ZUSCHNITT AS Zuschnitt, - B.BEDINGUNG AS Bedingung, - R.REGEL AS Regel, - BSP.Beispielnr AS Beispiel - - 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 - - LEFT JOIN Beispiele BSP - ON C.PARENT = BSP.Parent_ID - - WHERE C.PARENT = :parent - ORDER BY POSNR - """, - {"parent": parent_partnum}).fetchall() + -- 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 - for line in res: - print(line) + 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) -def write_bom_as_csv(parent_partnum: str) -> None: - out_path = f".data/BOM_{parent_partnum}_new.csv" + LEFT JOIN VAR_TEIL_BEDINGUNG_2 B + ON C.ID = B.COMP_ID - res: list[dict[str, str]] - res = con.execute(""" - SELECT C.POSNR AS Pos, - C.KOMPTEILENR AS 'Art.-Nr.', - C.MENGE As Menge, - C.ABW_ANZAHL AS 'Abw. Anzahl', - C.LAENGE AS Länge, - C.ZUSCHNITT AS Zuschnitt, - B.BEDINGUNG AS Bedingung, - R.REGEL AS Regel, - BSP.Beispielnr AS Beispiel + LEFT JOIN STKL_REGEL R + ON C.ID = R.COMP_ID - FROM COMP C - LEFT JOIN VAR_TEIL_BEDINGUNG_2 B - ON C.ID = B.COMP_ID + WHERE RD.ParentCode = :parent + AND RD.Positionsnummer = :posnr + """, + {"parent": self.parent_code, "posnr": pos_nr}).fetchone() + return res - LEFT JOIN STKL_REGEL R - ON C.ID = R.COMP_ID - - LEFT JOIN Beispiele BSP - ON C.PARENT = BSP.Parent_ID + 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, - WHERE C.PARENT = :parent - ORDER BY POSNR - """, - {"parent": args["partnumber"]}).fetchall() + -- 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, - with open("test.csv", "w", encoding="utf-8", newline="") as csv_file: - writer = csv.DictWriter(csv_file, delimiter=";", fieldnames=res[0].keys()) - writer.writeheader() - writer.writerows(res) + -- 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__": - parser = argparse.ArgumentParser() - parser.add_argument("-cd", "--csv_dir", help="Verzeichnis, in der CSV-Datei mit den neuen Stücklisteninformationen liegt", default=r".\data") - parser.add_argument("-sd", "--sql_dir", help="Verzeichnis, in der SQL-Datei mit den Informationen zur Teilreplizierung der SIVAS-Datenbank liegt", default=r".\data") - parser.add_argument("-n", "--partnumber", help="Teilenummer der Baugruppe", type=str, default="000000382") - args: dict = vars(parser.parse_args()) - - # Vorerst wird echte bei jedem Lauf neue db-Datei erstellt, da es mit Datenbank im RAM (:memory:) teilweise zu fehlerhaften Abfragen kam - if "sivas_replika.db" in os.listdir("."): - os.remove("./sivas_replika.db") - - with closing(sqlite3.connect("./sivas_replika.db", autocommit=True)) as con: - con.row_factory = dict_factory - main(args["partnumber"], args["sql_dir"], args["csv_dir"]) \ No newline at end of file + main() \ No newline at end of file