304 lines
11 KiB
Python
304 lines
11 KiB
Python
"""Builds a JSON hierarchy tree from the Position column of an Excel Stk.-Liste.
|
|
|
|
Usage:
|
|
python jsonFromXlsx.py --filename Fortna --tab soll
|
|
python jsonFromXlsx.py --filename Fortna --tab ist --output result.json
|
|
python jsonFromXlsx.py --all
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
import openpyxl
|
|
|
|
|
|
SHEET_NAMES = {
|
|
"soll": "Stk.-Liste SOLL",
|
|
"ist": "Stk.-Liste IST",
|
|
}
|
|
|
|
# Patterns to match sheet names (case-insensitive), ordered by priority
|
|
# Matches: Stk.-Liste SOLL, Stck.-Liste Soll, Stückliste SOLL, Stk. Liste IST, etc.
|
|
SHEET_PATTERNS = {
|
|
"soll": [
|
|
re.compile(r"st[ck]+[\.\-]?\s*liste[n]?\s+soll", re.IGNORECASE),
|
|
re.compile(r"st[üu]ckliste[n]?\s+soll", re.IGNORECASE),
|
|
],
|
|
"ist": [
|
|
re.compile(r"st[ck]+[\.\-]?\s*liste[n]?\s+ist", re.IGNORECASE),
|
|
re.compile(r"st[üu]ckliste[n]?\s+ist", re.IGNORECASE),
|
|
],
|
|
}
|
|
|
|
# Column indices (0-based) in the header row
|
|
COL_POSITION = 0
|
|
COL_TEILENUMMER = 1
|
|
COL_BEZEICHNUNG = 2
|
|
COL_MENGE = 3
|
|
HEADER_ROW = 3 # 1-based row number of the header (minimum, actual may be later)
|
|
RE_POSITION = re.compile(r"^\d+(/\d+)*$") # valid position pattern: "10" or "10/5/1"
|
|
|
|
# Sivas numbers where Menge represents length (meters) instead of piece count
|
|
LAENGENABHAENGIG = {
|
|
"821106001", "821106002",
|
|
"825084000", "825084007",
|
|
"825104000", "825104007",
|
|
"825144000", "825144007",
|
|
"821006026", "821006034", "821006035", "821006045",
|
|
"825004000",
|
|
}
|
|
|
|
|
|
def find_xlsx(examples_dir: str, pattern: str) -> Path:
|
|
"""Find an .xlsx file in examples/ whose name contains the given pattern."""
|
|
matches = []
|
|
for f in Path(examples_dir).glob("*.xlsx"):
|
|
if pattern.lower() in f.name.lower():
|
|
matches.append(f)
|
|
if not matches:
|
|
print(f"FEHLER: Keine .xlsx-Datei mit Muster '{pattern}' in {examples_dir} gefunden.")
|
|
sys.exit(1)
|
|
if len(matches) > 1:
|
|
print(f"Mehrere Dateien gefunden fuer Muster '{pattern}':")
|
|
for m in sorted(matches):
|
|
print(f" {m.name}")
|
|
print(f"Verwende: {matches[0].name}")
|
|
return sorted(matches)[0]
|
|
|
|
|
|
def make_key(teilenummer: str, bezeichnung: str, is_unique: bool) -> str:
|
|
"""Build a node key from Teilenummer, adding Bezeichnung if not unique."""
|
|
if is_unique:
|
|
return teilenummer
|
|
combined = f"{teilenummer}-{bezeichnung}"
|
|
return combined.replace(" ", "_")
|
|
|
|
|
|
def unique_key(parent: dict, key: str) -> str:
|
|
"""Return key with _1, _2, ... suffix if it already exists in parent."""
|
|
if key not in parent:
|
|
return key
|
|
suffix = 1
|
|
while f"{key}_{suffix}" in parent:
|
|
suffix += 1
|
|
return f"{key}_{suffix}"
|
|
|
|
|
|
def build_tree(ws, sheet_type: str) -> dict:
|
|
"""Read the worksheet and build a nested dict from Position paths.
|
|
|
|
Keys are Teilenummer-based (with Bezeichnung if non-unique).
|
|
Position paths are used only for hierarchy, not as keys.
|
|
"""
|
|
# First pass: collect all Teilenummern to determine uniqueness
|
|
teil_counter: Counter = Counter()
|
|
rows_data = []
|
|
for row in ws.iter_rows(min_row=HEADER_ROW + 1, max_row=ws.max_row, values_only=True):
|
|
pos = row[COL_POSITION]
|
|
teil = row[COL_TEILENUMMER]
|
|
bez = row[COL_BEZEICHNUNG]
|
|
menge = row[COL_MENGE] if len(row) > COL_MENGE else None
|
|
if pos is None or teil is None:
|
|
continue
|
|
pos_str = str(pos).strip()
|
|
if not RE_POSITION.match(pos_str):
|
|
continue # skip header rows and non-position entries
|
|
teil_str = str(teil).strip()
|
|
bez_str = str(bez).strip() if bez else ""
|
|
try:
|
|
menge_val = float(menge) if menge is not None else 0.0
|
|
except (ValueError, TypeError):
|
|
menge_val = 0.0
|
|
teil_counter[teil_str] += 1
|
|
rows_data.append((pos_str, teil_str, bez_str, menge_val))
|
|
|
|
unique_teile = {t for t, count in teil_counter.items() if count == 1}
|
|
|
|
# Second pass: build tree with Teilenummer-based keys
|
|
tree: dict = {}
|
|
pos_to_node: dict[str, dict] = {} # maps position path -> dict node
|
|
# Track metadata per node for leaf post-processing
|
|
node_meta: dict[int, tuple[str, str, float]] = {} # id(child) -> (teil_str, bez_str, menge)
|
|
|
|
for pos_str, teil_str, bez_str, menge_val in rows_data:
|
|
key = make_key(teil_str, bez_str, teil_str in unique_teile)
|
|
segments = pos_str.split("/")
|
|
|
|
# Find parent dict
|
|
if len(segments) == 1:
|
|
parent = tree
|
|
else:
|
|
parent_pos = "/".join(segments[:-1])
|
|
if parent_pos not in pos_to_node:
|
|
# Parent row missing in data, create placeholder
|
|
pos_to_node[parent_pos] = {}
|
|
parent = pos_to_node[parent_pos]
|
|
|
|
# Insert with unique key
|
|
final_key = unique_key(parent, key)
|
|
child: dict = {}
|
|
parent[final_key] = child
|
|
pos_to_node[pos_str] = child
|
|
node_meta[id(child)] = (teil_str, bez_str, menge_val)
|
|
|
|
is_sivas = re.compile(r"^\d{9}$").match
|
|
|
|
# Post-process: leaves (empty dicts) -> key=Teilenummer, value=Bezeichnung
|
|
def simplify_leaves(node: dict) -> dict:
|
|
result: dict = {}
|
|
for key, value in node.items():
|
|
if isinstance(value, dict) and len(value) == 0:
|
|
# Leaf node: use only Teilenummer as key, Bezeichnung as value
|
|
teil, bez, menge = node_meta.get(id(value), ("", "", 0.0))
|
|
if teil in LAENGENABHAENGIG:
|
|
# Length-dependent part: sum up Menge as laenge
|
|
if teil in result and isinstance(result[teil], dict) and "laenge" in result[teil]:
|
|
result[teil]["laenge"] += menge
|
|
elif teil in result and isinstance(result[teil], str):
|
|
result[teil] = {"laenge": menge, "beschreibung": result[teil]}
|
|
else:
|
|
leaf_key = unique_key(result, teil)
|
|
result[leaf_key] = {"laenge": menge, "beschreibung": bez}
|
|
elif teil in result and is_sivas(teil):
|
|
# Aggregate duplicate 9-digit Sivas numbers
|
|
existing = result[teil]
|
|
if isinstance(existing, str):
|
|
result[teil] = {"anzahl": 2, "beschreibung": existing}
|
|
else:
|
|
existing["anzahl"] += 1
|
|
else:
|
|
leaf_key = unique_key(result, teil)
|
|
result[leaf_key] = bez
|
|
elif isinstance(value, dict):
|
|
# Branch node: enrich bare 9-digit keys with Bezeichnung
|
|
branch_key = key
|
|
if is_sivas(key):
|
|
teil, bez, _ = node_meta.get(id(value), ("", "", 0.0))
|
|
if bez:
|
|
branch_key = f"{teil}-{bez}".replace(" ", "_")
|
|
branch_key = unique_key(result, branch_key)
|
|
else:
|
|
branch_key = unique_key(result, key)
|
|
result[branch_key] = simplify_leaves(value)
|
|
else:
|
|
result[key] = value
|
|
return result
|
|
|
|
return simplify_leaves(tree)
|
|
|
|
|
|
def process_file(xlsx_path: Path, sheet_name: str, tab: str, results_dir: Path,
|
|
out_path: Path | None = None) -> bool:
|
|
"""Process a single xlsx file. Returns True on success."""
|
|
print(f"Lade: {xlsx_path.name}")
|
|
print(f"Sheet: {sheet_name}")
|
|
|
|
wb = openpyxl.load_workbook(str(xlsx_path), data_only=True)
|
|
# Find matching sheet: try exact (case-insensitive), then pattern matching
|
|
actual_name = None
|
|
for name in wb.sheetnames:
|
|
if name.lower() == sheet_name.lower():
|
|
actual_name = name
|
|
break
|
|
if actual_name is None:
|
|
for pattern in SHEET_PATTERNS.get(tab, []):
|
|
for name in wb.sheetnames:
|
|
if pattern.search(name):
|
|
actual_name = name
|
|
break
|
|
if actual_name:
|
|
break
|
|
if actual_name is None:
|
|
print(f" SKIP: Kein passendes Sheet fuer '{tab}' gefunden. Vorhanden: {wb.sheetnames}")
|
|
return False
|
|
|
|
ws = wb[actual_name]
|
|
tree = build_tree(ws, tab)
|
|
|
|
if out_path is None:
|
|
stem = xlsx_path.stem.replace(" ", "_")
|
|
out_path = results_dir / f"{stem}_{tab}.json"
|
|
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
json.dump(tree, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f" Ausgabe: {out_path}")
|
|
return True
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Erzeugt eine JSON-Baumstruktur aus der Positionsspalte einer Excel-Stueckliste."
|
|
)
|
|
parser.add_argument(
|
|
"--filename",
|
|
default=None,
|
|
help="Suchmuster fuer die .xlsx-Datei im examples/-Ordner (z.B. 'Fortna')",
|
|
)
|
|
parser.add_argument(
|
|
"--tab",
|
|
choices=["soll", "ist"],
|
|
default="soll",
|
|
help="Welches Tabellenblatt gelesen wird: 'soll' oder 'ist' (default: soll)",
|
|
)
|
|
parser.add_argument(
|
|
"--output",
|
|
default=None,
|
|
help="Ausgabedatei (default: <filename>_<tab>.json im results/-Ordner)",
|
|
)
|
|
parser.add_argument(
|
|
"--all",
|
|
action="store_true",
|
|
dest="process_all",
|
|
help="Alle .xlsx-Dateien im examples/-Ordner in _ist und _soll JSON konvertieren",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
if not args.process_all and not args.filename:
|
|
parser.error("Entweder --filename oder --all muss angegeben werden.")
|
|
|
|
# Resolve paths relative to project root
|
|
project_root = Path(os.environ.get("PROJECT", Path(__file__).resolve().parent.parent))
|
|
examples_dir = project_root / "examples"
|
|
results_dir = project_root / "results"
|
|
results_dir.mkdir(exist_ok=True)
|
|
|
|
tab = args.tab
|
|
|
|
if args.process_all:
|
|
xlsx_files = sorted(f for f in Path(examples_dir).glob("*.xlsx")
|
|
if not f.name.startswith("~$"))
|
|
print(f"Verarbeite {len(xlsx_files)} Dateien aus {examples_dir}\n")
|
|
ok, skip = 0, 0
|
|
tabs = [tab] if tab != "soll" or "--tab" in sys.argv else ["soll", "ist"]
|
|
for xlsx_path in xlsx_files:
|
|
for t in tabs:
|
|
sheet_name = SHEET_NAMES[t]
|
|
if process_file(xlsx_path, sheet_name, t, results_dir):
|
|
ok += 1
|
|
else:
|
|
skip += 1
|
|
print()
|
|
print(f"Fertig: {ok} JSON erzeugt, {skip} uebersprungen.")
|
|
else:
|
|
xlsx_path = find_xlsx(str(examples_dir), args.filename)
|
|
sheet_name = SHEET_NAMES[tab]
|
|
|
|
out_path = None
|
|
if args.output:
|
|
out_path = Path(args.output)
|
|
if not out_path.is_absolute():
|
|
out_path = results_dir / out_path
|
|
|
|
if not process_file(xlsx_path, sheet_name, tab, results_dir, out_path):
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|