erste Fassung des Extracts von json files aus den Excel Dateien des Kontrolling geschrieben.
This commit is contained in:
@@ -0,0 +1,460 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user