import argparse import csv import os import sqlite3 from contextlib import closing def main(asm_partnum, sql_filedir, csv_filedir): 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) def csv_to_db(csv_data: list[dict[str, str]]) -> None: """Schreibt neue Stücklistendaten aus CSV in temporäre Datenbanktabelle.""" 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)) """) for line in csv_data: con.execute(""" INSERT INTO RD_COMP (ParentCode, Positionsnummer, ChildCode, Menge, Uebergabelaenge) VALUES (:ParentCode, :Positionsnummer, :ChildCode, :Menge, :Uebergabelaenge)""", line) def dict_factory(cursor, row): d = {} for idx, col in enumerate(cursor.description): d[col[0]] = row[idx] return d 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 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) return csv_data 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 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 """ 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) WHERE RD.ParentCode = :parent """, {"parent": parent_partnum}).fetchall() if res is not None: res = set([line["POS"] for line in res]) return res def identical_occurences_in_both_boms(parent_partnum: str, child_id: str) -> bool: """ Prüfung Fall 1.1: """ 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 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() 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: """ Prüfung Fall 1.2 """ 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 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 get_data_for_new_pos(parent_partnum: str, pos: str) -> list[dict[str, str]]: """ Fall 1.3 (Neuanlage von Positionen) """ 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 WHERE C.KOMPTEILENR IS NULL AND RD.ParentCode = :parent AND RD.Positionsnummer = :posnr """, {"parent": parent_partnum, "posnr": pos}).fetchone() return res 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 INNER JOIN COMP C ON RD.ParentCode = C.PARENT AND RD.ChildCode = C.KOMPTEILENR AND RD.Uebergabelaenge = COALESCE(C.LAENGE, 0) LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON C.ID = B.COMP_ID LEFT JOIN STKL_REGEL R ON C.ID = R.COMP_ID WHERE RD.ParentCode = :parent AND RD.Positionsnummer = :posnr """, {"parent": parent_partnum, "posnr": pos}).fetchone() return res 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, -- Bestimmt Wert für Spalte 'ABW_ANZAHL': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Menge' aus CSV CASE WHEN RD.Uebergabelaenge = 0 THEN NULL ELSE RD.Menge END AS ABW_ANZAHL, -- Bestimmt Wert für Spalte 'MENGE': Berechnung Dezimalwert, wenn Pos. längenabhängig, sonst Wert 'Menge' aus CSV CASE WHEN RD.Uebergabelaenge = 0 THEN RD.Menge ELSE RD.Menge * (RD.Uebergabelaenge + COALESCE(C.ZUSCHNITT, 0)) / 1000 END AS MENGE, -- Bestimmt Wert für Spalte 'LAENGE': NULL, wenn Pos. nicht längenabhängig, sonst Wert 'Uebergabelaenge' aus CSV CASE WHEN RD.Uebergabelaenge = 0 THEN NULL ELSE RD.Uebergabelaenge END AS LAENGE FROM RD_COMP RD INNER JOIN COMP C ON RD.ParentCode = C.PARENT AND RD.ChildCode = C.KOMPTEILENR AND RD.Positionsnummer = C.POSNR LEFT JOIN VAR_TEIL_BEDINGUNG_2 B ON C.ID = B.COMP_ID LEFT JOIN STKL_REGEL R ON C.ID = R.COMP_ID WHERE RD.ParentCode = :parent AND RD.Positionsnummer = :posnr """, {"parent": parent_partnum, "posnr": pos}).fetchone() return res def create_new_bom(parent_partnum:str) -> None: 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() for line in res: print(line) def write_bom_as_csv(parent_partnum: str) -> None: out_path = f".data/BOM_{parent_partnum}_new.csv" 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 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": args["partnumber"]}).fetchall() 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) 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"])