461 lines
14 KiB
Python
461 lines
14 KiB
Python
"""
|
|
create_examples.py - Liest Stuecklisten-Daten aus Excel-Dateien im Examples-Ordner.
|
|
|
|
Verwendung:
|
|
python create_examples.py [--example-dir <pfad>]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import json
|
|
import logging
|
|
import re
|
|
import sys
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class StkItem:
|
|
position: str
|
|
teilenummer: str
|
|
bezeichnung: str
|
|
menge: Any
|
|
|
|
|
|
def _cell_str(value: Any) -> str:
|
|
return str(value).strip() if value is not None else ""
|
|
|
|
|
|
def _find_col_index(header_row: tuple[Any, ...], candidates: list[str]) -> int | None:
|
|
normalized = [_cell_str(v).lower() for v in header_row]
|
|
for candidate in candidates:
|
|
candidate_norm = candidate.lower()
|
|
if candidate_norm in normalized:
|
|
return normalized.index(candidate_norm)
|
|
return None
|
|
|
|
|
|
def _extract_stk_liste_ist(xlsx_path: Path) -> list[StkItem]:
|
|
import openpyxl
|
|
|
|
wb = openpyxl.load_workbook(str(xlsx_path), read_only=True, data_only=True)
|
|
if "Stk.-Liste IST" not in wb.sheetnames:
|
|
wb.close()
|
|
raise ValueError("Reiter 'Stk.-Liste IST' nicht gefunden")
|
|
|
|
ws = wb["Stk.-Liste IST"]
|
|
|
|
header_idx: tuple[int | None, int | None, int | None, int | None] | None = None
|
|
items: list[StkItem] = []
|
|
|
|
for row in ws.iter_rows(values_only=True):
|
|
if header_idx is None:
|
|
pos_idx = _find_col_index(list(row), ["position", "pos", "pos."])
|
|
part_idx = _find_col_index(
|
|
list(row),
|
|
["teilenummer", "teilnummer", "article", "artikelnummer"],
|
|
)
|
|
desc_idx = _find_col_index(
|
|
list(row),
|
|
["bezeichnung", "beschreibung", "description", "benennung"],
|
|
)
|
|
qty_idx = _find_col_index(list(row), ["menge", "anzahl", "qty", "quantity"])
|
|
if (
|
|
pos_idx is not None
|
|
and part_idx is not None
|
|
and desc_idx is not None
|
|
and qty_idx is not None
|
|
):
|
|
header_idx = (pos_idx, part_idx, desc_idx, qty_idx)
|
|
continue
|
|
|
|
pos_idx, part_idx, desc_idx, qty_idx = header_idx
|
|
assert (
|
|
pos_idx is not None
|
|
and part_idx is not None
|
|
and desc_idx is not None
|
|
and qty_idx is not None
|
|
)
|
|
|
|
position = _cell_str(row[pos_idx]) if pos_idx < len(row) else ""
|
|
teilenummer = _cell_str(row[part_idx]) if part_idx < len(row) else ""
|
|
bezeichnung = _cell_str(row[desc_idx]) if desc_idx < len(row) else ""
|
|
menge = row[qty_idx] if qty_idx < len(row) else None
|
|
|
|
# Leere Zeilen ueberspringen
|
|
if not position and not teilenummer and not bezeichnung and menge in ("", None):
|
|
continue
|
|
if not position or not teilenummer:
|
|
continue
|
|
|
|
items.append(
|
|
StkItem(
|
|
position=position,
|
|
teilenummer=teilenummer,
|
|
bezeichnung=bezeichnung,
|
|
menge=menge,
|
|
)
|
|
)
|
|
|
|
wb.close()
|
|
return items
|
|
|
|
|
|
def _to_number(value: Any) -> float:
|
|
if value is None:
|
|
return 0.0
|
|
if isinstance(value, (int, float)):
|
|
return float(value)
|
|
text = str(value).strip().replace(",", ".")
|
|
if not text:
|
|
return 0.0
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def _common_path_prefix(paths: list[str]) -> str:
|
|
parts_list = [p.split("/") for p in paths if p]
|
|
if not parts_list:
|
|
return ""
|
|
common = parts_list[0]
|
|
for parts in parts_list[1:]:
|
|
i = 0
|
|
max_i = min(len(common), len(parts))
|
|
while i < max_i and common[i] == parts[i]:
|
|
i += 1
|
|
common = common[:i]
|
|
if not common:
|
|
break
|
|
return "/".join(common)
|
|
|
|
|
|
def _load_product_numbers(cfg_path: Path) -> dict[str, str]:
|
|
"""
|
|
Laedt 9-stellige Produktnummern aus cfg/product_numbers.cfg.
|
|
|
|
Erwartetes Format:
|
|
[Sektion]
|
|
123456789
|
|
987654321
|
|
"""
|
|
allowed_sections = {"omniflo", "ils", "cpc", "trolley", "geruest"}
|
|
products: dict[str, str] = {}
|
|
current_section = ""
|
|
|
|
if not cfg_path.exists():
|
|
return products
|
|
|
|
for raw_line in cfg_path.read_text(encoding="utf-8").splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
|
|
if line.startswith("[") and line.endswith("]"):
|
|
current_section = line[1:-1].strip().lower()
|
|
continue
|
|
|
|
if current_section not in allowed_sections:
|
|
continue
|
|
|
|
if re.match(r"^\d{9}$", line):
|
|
products[line] = current_section
|
|
|
|
return products
|
|
|
|
|
|
def write_results(
|
|
xlsx_path: Path,
|
|
filtered_items: list[dict[str, Any]],
|
|
log: logging.Logger,
|
|
write_json: bool = False,
|
|
write_csv: bool = False,
|
|
write_xlsx: bool = False,
|
|
) -> None:
|
|
if write_json:
|
|
json_path = xlsx_path.with_suffix(".json")
|
|
with json_path.open("w", encoding="utf-8") as fh:
|
|
json.dump(filtered_items, fh, indent=2, ensure_ascii=False, default=str)
|
|
log.info("JSON geschrieben: %s", json_path)
|
|
|
|
if write_csv:
|
|
csv_path = xlsx_path.with_suffix(".csv")
|
|
with csv_path.open("w", encoding="utf-8-sig", newline="") as fh:
|
|
writer = csv.writer(fh, delimiter=";")
|
|
writer.writerow(["position", "teilenummer", "bezeichnung", "menge"])
|
|
for row in filtered_items:
|
|
writer.writerow(
|
|
[
|
|
row.get("position", ""),
|
|
row.get("teilenummer", ""),
|
|
row.get("bezeichnung", ""),
|
|
row.get("menge", ""),
|
|
]
|
|
)
|
|
log.info("CSV geschrieben: %s", csv_path)
|
|
|
|
if write_xlsx:
|
|
import openpyxl
|
|
|
|
xlsx_out = xlsx_path.with_name(f"{xlsx_path.stem}_result.xlsx")
|
|
wb = openpyxl.Workbook()
|
|
ws_3 = wb.active
|
|
ws_3.title = "3er"
|
|
ws_4 = wb.create_sheet("4er")
|
|
ws_result = wb.create_sheet("Result")
|
|
ws_product = wb.create_sheet("product")
|
|
|
|
headers = ["position", "teilenummer", "bezeichnung", "menge"]
|
|
ws_3.append(headers)
|
|
ws_4.append(headers)
|
|
|
|
aggregated: dict[str, dict[str, Any]] = defaultdict(
|
|
lambda: {
|
|
"teilenummer": "",
|
|
"bezeichnung": "",
|
|
"gesamtmenge": 0.0,
|
|
"positions": [],
|
|
}
|
|
)
|
|
|
|
for row in filtered_items:
|
|
slash_count = str(row.get("position", "")).count("/")
|
|
if slash_count == 3:
|
|
ws_3.append([row.get(h, "") for h in headers])
|
|
elif slash_count == 4:
|
|
ws_4.append([row.get(h, "") for h in headers])
|
|
|
|
tn = str(row.get("teilenummer", "")).strip()
|
|
if not tn:
|
|
continue
|
|
if not aggregated[tn]["teilenummer"]:
|
|
aggregated[tn]["teilenummer"] = tn
|
|
aggregated[tn]["bezeichnung"] = row.get("bezeichnung", "")
|
|
aggregated[tn]["gesamtmenge"] += _to_number(row.get("menge"))
|
|
aggregated[tn]["positions"].append(str(row.get("position", "")).strip())
|
|
|
|
ws_result.append(["teilenummer", "bezeichnung", "gesamtmenge", "mainpath", "endpath"])
|
|
|
|
project_root = Path(__file__).resolve().parent.parent
|
|
product_cfg = project_root / "cfg" / "product_numbers.cfg"
|
|
product_numbers = _load_product_numbers(product_cfg)
|
|
ws_product.append(["teilenummer", "bezeichnung", "gesamtmenge", "mainpath", "endpath"])
|
|
|
|
for tn in sorted(aggregated.keys()):
|
|
positions = [p for p in aggregated[tn]["positions"] if p]
|
|
mainpath = _common_path_prefix(positions)
|
|
variable_parts: list[str] = []
|
|
for pos in positions:
|
|
if mainpath:
|
|
suffix = pos[len(mainpath):].lstrip("/")
|
|
else:
|
|
suffix = pos
|
|
if suffix and suffix not in variable_parts:
|
|
variable_parts.append(suffix)
|
|
endpath = ",".join(variable_parts)
|
|
result_row = [
|
|
aggregated[tn]["teilenummer"],
|
|
aggregated[tn]["bezeichnung"],
|
|
aggregated[tn]["gesamtmenge"],
|
|
mainpath,
|
|
endpath,
|
|
]
|
|
ws_result.append(result_row)
|
|
if tn in product_numbers:
|
|
ws_product.append(result_row)
|
|
|
|
wb.save(str(xlsx_out))
|
|
wb.close()
|
|
log.info("XLSX geschrieben: %s", xlsx_out)
|
|
|
|
|
|
def _process_single_folder(
|
|
folder: Path,
|
|
example_dir: Path,
|
|
result: dict[str, list[dict[str, Any]]],
|
|
log: logging.Logger,
|
|
write_json: bool = False,
|
|
write_csv: bool = False,
|
|
write_xlsx: bool = False,
|
|
) -> None:
|
|
xlsx_files = sorted(folder.glob("*.xlsx"))
|
|
if not xlsx_files:
|
|
return
|
|
|
|
xlsx_path = xlsx_files[0]
|
|
rel_folder = str(folder.relative_to(example_dir))
|
|
try:
|
|
items = _extract_stk_liste_ist(xlsx_path)
|
|
except ValueError as exc:
|
|
log.warning("%s/%s uebersprungen: %s", rel_folder, xlsx_path.name, exc)
|
|
return
|
|
size_path = (3, 4)
|
|
|
|
filtered_items = [
|
|
{
|
|
"position": item.position,
|
|
"teilenummer": item.teilenummer,
|
|
"bezeichnung": item.bezeichnung,
|
|
"menge": item.menge,
|
|
}
|
|
for item in items
|
|
if item.position.count("/") in size_path
|
|
]
|
|
result[rel_folder] = filtered_items
|
|
write_results(
|
|
xlsx_path,
|
|
filtered_items,
|
|
log,
|
|
write_json=write_json,
|
|
write_csv=write_csv,
|
|
write_xlsx=write_xlsx,
|
|
)
|
|
|
|
def process_all(
|
|
example_dir: Path,
|
|
write_json: bool = False,
|
|
write_csv: bool = False,
|
|
write_xlsx: bool = False,
|
|
) -> dict[str, list[dict[str, Any]]]:
|
|
"""
|
|
Durchsucht rekursiv alle Unterverzeichnisse in example_dir.
|
|
|
|
Pro Unterverzeichnis wird die erste gefundene .xlsx-Datei gelesen und aus
|
|
dem Reiter "Stk.-Liste IST" werden Position, Teilenummer, Menge extrahiert.
|
|
"""
|
|
if not example_dir.exists():
|
|
log.error("Beispiel-Verzeichnis existiert nicht: %s", example_dir)
|
|
sys.exit(1)
|
|
|
|
subdirs = sorted([p for p in example_dir.rglob("*") if p.is_dir()])
|
|
if not subdirs:
|
|
log.error("Keine Unterverzeichnisse in %s gefunden", example_dir)
|
|
sys.exit(1)
|
|
|
|
log.info("Gefunden: %d Unterverzeichnisse in %s", len(subdirs), example_dir)
|
|
|
|
result: dict[str, list[dict[str, Any]]] = {}
|
|
for folder in subdirs:
|
|
log.info("Verarbeite %s", folder)
|
|
_process_single_folder(
|
|
folder,
|
|
example_dir,
|
|
result,
|
|
log,
|
|
write_json=write_json,
|
|
write_csv=write_csv,
|
|
write_xlsx=write_xlsx,
|
|
)
|
|
|
|
log.info("Fertig. Gelesene Verzeichnisse mit Daten: %d", len(result))
|
|
return result
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Liest Position/Teilenummer/Menge aus Example-Exceldateien"
|
|
)
|
|
parser.add_argument(
|
|
"--example-dir",
|
|
type=Path,
|
|
default=None,
|
|
help=(
|
|
"Unterordnername in PV_EXAMPLES oder direkter Verzeichnispfad "
|
|
"(default: alle Unterordner in PV_EXAMPLES)"
|
|
),
|
|
)
|
|
parser.add_argument("--verbose", "-v", action="store_true",
|
|
help="Ausfuehrliche Ausgabe")
|
|
parser.add_argument(
|
|
"--2json",
|
|
dest="to_json",
|
|
action="store_true",
|
|
help="Schreibt Ausgabe als .json neben die jeweilige .xlsx",
|
|
)
|
|
parser.add_argument(
|
|
"--2csv",
|
|
dest="to_csv",
|
|
action="store_true",
|
|
help="Schreibt Ausgabe als .csv neben die jeweilige .xlsx",
|
|
)
|
|
parser.add_argument(
|
|
"--2xlsx",
|
|
dest="to_xlsx",
|
|
action="store_true",
|
|
help="Schreibt Ausgabe als _result.xlsx neben die jeweilige .xlsx",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(levelname)-7s %(message)s",
|
|
)
|
|
|
|
import os
|
|
|
|
pv_examples = os.environ.get("PV_EXAMPLES")
|
|
if not pv_examples:
|
|
log.error("PV_EXAMPLES ist nicht gesetzt")
|
|
sys.exit(1)
|
|
|
|
examples_root = Path(pv_examples).resolve()
|
|
if not examples_root.exists() or not examples_root.is_dir():
|
|
log.error("PV_EXAMPLES ist kein gueltiges Verzeichnis: %s", examples_root)
|
|
sys.exit(1)
|
|
|
|
if args.example_dir is None:
|
|
log.info("Beispiel-Verzeichnis (alle Unterordner): %s", examples_root)
|
|
data = process_all(
|
|
examples_root,
|
|
write_json=args.to_json,
|
|
write_csv=args.to_csv,
|
|
write_xlsx=args.to_xlsx,
|
|
)
|
|
log.info("Objekt erzeugt mit %d Schluesseln", len(data))
|
|
print(json.dumps(data, indent=2, ensure_ascii=False, default=str))
|
|
return
|
|
|
|
# --example-dir: entweder direkter Verzeichnispfad oder Unterordnername in PV_EXAMPLES
|
|
candidate = args.example_dir
|
|
if candidate.is_absolute():
|
|
target_dir = candidate.resolve()
|
|
else:
|
|
as_path = candidate.resolve()
|
|
if as_path.exists() and as_path.is_dir():
|
|
target_dir = as_path
|
|
else:
|
|
target_dir = (examples_root / candidate).resolve()
|
|
|
|
if not target_dir.exists() or not target_dir.is_dir():
|
|
log.error(
|
|
"--example-dir muss ein gueltiger Verzeichnispfad oder Unterordner in PV_EXAMPLES sein: %s",
|
|
args.example_dir,
|
|
)
|
|
sys.exit(1)
|
|
|
|
log.info("Beispiel-Verzeichnis (einzeln): %s", target_dir)
|
|
result: dict[str, list[dict[str, Any]]] = {}
|
|
_process_single_folder(
|
|
target_dir,
|
|
target_dir.parent,
|
|
result,
|
|
log,
|
|
write_json=args.to_json,
|
|
write_csv=args.to_csv,
|
|
write_xlsx=args.to_xlsx,
|
|
)
|
|
log.info("Objekt erzeugt mit %d Schluesseln", len(result))
|
|
#print(json.dumps(result, indent=2, ensure_ascii=False, default=str))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|