137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
import argparse
|
||
import configparser
|
||
import os
|
||
import subprocess
|
||
import csv
|
||
import pandas as pd
|
||
|
||
|
||
def get_sivas_teilestamm(sivas_ids, overwrite=True, out_dir= os.environ.get('PROJECT_DATA')):
|
||
"""
|
||
Generiert für jede angegebene SIVAS-Teilenummer eine CSV-Datei, in der die Teilestamm-Informationen gespeichert sind
|
||
"""
|
||
sivas_teilestamm_exe = os.environ.get('SIVAS_TEILESTAMM')
|
||
|
||
if not os.path.isfile(sivas_teilestamm_exe):
|
||
raise FileNotFoundError(
|
||
f"Programmpfad '{sivas_teilestamm_exe}' nicht erreichbar. "
|
||
f"Bitte Zugriff und Variable SIVAS_TEILESTAMM in setenv.bat prüfen."
|
||
)
|
||
|
||
if not os.path.isdir(out_dir):
|
||
raise NotADirectoryError(
|
||
f"Ausgabeverzeichnis '{out_dir}' existiert nicht. "
|
||
f"Bitte VPN oder setenv.bat prüfen."
|
||
)
|
||
|
||
if isinstance(sivas_ids, str):
|
||
sivas_ids = list(set(sivas_ids.strip().split(",")))
|
||
|
||
ids_to_generate = []
|
||
for id in sivas_ids:
|
||
if not overwrite and os.path.isfile(os.path.join(out_dir, id + ".csv")):
|
||
continue
|
||
ids_to_generate.append(id)
|
||
|
||
if len(ids_to_generate) == 0:
|
||
print("Alle CSVs vorhanden oder keine neuen Nummern")
|
||
return
|
||
|
||
ids_to_generate_str = ",".join([f'"{id}"' for id in ids_to_generate])
|
||
|
||
os.environ["SIVAS_EXCEL_EXPORT_DIR"] = out_dir
|
||
print(f"Erstelle CSV(s) mit Teilestamminformationen...")
|
||
subprocess.run(f'{sivas_teilestamm_exe} -teilenr={ids_to_generate_str} -open_target_dir=n')
|
||
|
||
for id in ids_to_generate:
|
||
csv_path = os.path.join(out_dir, id + ".csv")
|
||
with open(csv_path, encoding="utf-8", newline="") as csv_file:
|
||
csv_data = csv.reader(csv_file, delimiter=";")
|
||
rows = list(csv_data)
|
||
if len(rows) != 2:
|
||
raise ValueError(f"SIVAS-Teilenummer '{id}' existiert nicht oder liefert keine gültige CSV.")
|
||
|
||
def read_bezeichner_from_csv(teilenr):
|
||
"""
|
||
Liest den Bezeichner aus einer SIVAS-CSV für eine gegebene Teilenummer.
|
||
|
||
:param teilenr: SIVAS Teilenummer als String
|
||
:return: Bezeichner-String, falls erfolgreich. Wirft Exception bei Fehler.
|
||
"""
|
||
csv_dir = os.environ.get("PROJECT_DATA", ".")
|
||
csv_path = os.path.join(csv_dir, f"{teilenr}.csv")
|
||
|
||
if not os.path.isfile(csv_path):
|
||
raise FileNotFoundError(f"CSV für Teilenummer '{teilenr}' nicht gefunden unter {csv_path}")
|
||
|
||
try:
|
||
df = pd.read_csv(csv_path, sep=";", encoding="utf-8")
|
||
except Exception as e:
|
||
raise ValueError(f"Fehler beim Einlesen der CSV: {e}")
|
||
|
||
if df.shape[0] != 1:
|
||
raise ValueError(f"Ungültiger Zeileninhalt für '{teilenr}' – erwartet genau eine Datenzeile.")
|
||
|
||
if "Bezeichnung1" not in df.columns:
|
||
raise ValueError(f"Spalte 'Bezeichnung1' nicht gefunden in CSV für '{teilenr}'.")
|
||
|
||
return str(df["Bezeichnung1"].iloc[0]).strip()
|
||
|
||
def update_bezeichner(names_config, config_path):
|
||
|
||
if "Missing" not in names_config or not names_config["Missing"]:
|
||
print("Keine fehlenden Einträge in [Missing]. Keine Abfrage angestoßen.")
|
||
return
|
||
|
||
missing_ids = list(names_config["Missing"].keys())
|
||
resolved_ids = []
|
||
|
||
print(f"Starte SIVAS-Abfrage für {len(missing_ids)} fehlende Teilenummer(n)...")
|
||
|
||
try:
|
||
get_sivas_teilestamm(missing_ids)
|
||
except Exception as e:
|
||
print(f"Fehler bei der SIVAS-Abfrage: {e}")
|
||
|
||
updated = 0
|
||
|
||
|
||
for id in missing_ids:
|
||
try:
|
||
bezeichner = read_bezeichner_from_csv(id)
|
||
if "Sivasnummern" not in names_config:
|
||
names_config["Sivasnummern"] = {}
|
||
names_config["Sivasnummern"][id] = bezeichner
|
||
resolved_ids.append(id)
|
||
updated += 1
|
||
except Exception as e:
|
||
print(f"WARNUNG: Bezeichner für {id} konnte nicht gelesen werden: {e}")
|
||
|
||
for id in resolved_ids:
|
||
names_config["Missing"].pop(id, None)
|
||
|
||
if "Missing" in names_config and not names_config["Missing"]:
|
||
names_config.remove_section("Missing")
|
||
|
||
with open(config_path, "w", encoding="utf-8") as f:
|
||
names_config.write(f)
|
||
|
||
print(f"{updated} Bezeichner erfolgreich ergänzt und Konfiguration aktualisiert.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
|
||
parser = argparse.ArgumentParser(description="Aktualisiert fehlende Bezeichner in bezeichner.cfg anhand von SIVAS-Teilestamm.")
|
||
parser.add_argument("-c", "--config", type=str, default=os.path.join(os.environ.get("PROJECT_CFG", "."), "bezeichner.cfg"), help="Pfad zur bezeichner.cfg")
|
||
|
||
args = parser.parse_args()
|
||
config_path = args.config
|
||
|
||
if not os.path.isfile(config_path):
|
||
raise FileNotFoundError(f"Config-Datei nicht gefunden: {config_path}")
|
||
|
||
config = configparser.ConfigParser()
|
||
config.optionxform = str # Groß-/Kleinschreibung erhalten
|
||
config.read(config_path, encoding="utf-8")
|
||
|
||
update_bezeichner(config) |