291 lines
9.9 KiB
Python
291 lines
9.9 KiB
Python
"""
|
|
collect_partnumbers.py - Sammelt 9-stellige Teilenummern aus Produktprogramm-RSTs.
|
|
|
|
Erkennt auch Bereiche wie 834372007-011 und expandiert diese zu
|
|
834372007, 834372008, 834372009, 834372010, 834372011.
|
|
|
|
Gleicht die Nummern gegen eine Excel-Stueckliste ab und schreibt
|
|
eine CSV mit Teilenummer, Quelldatei und Bezeichnung.
|
|
|
|
Verwendung:
|
|
python collect_partnumbers.py [--data-dir <pfad>]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def expand_ranges(text: str) -> list[str]:
|
|
"""Findet alle 9-stelligen Nummern und Bereiche in einem Text.
|
|
|
|
Unterstuetzte Formate:
|
|
- Einzeln: 834372007
|
|
- Bereich: 834372007-011 -> 834372007..834372011
|
|
- OCR-verklebt: "834372001-005834372 007-011" bedeutet eigentlich
|
|
"834372001-005 834372007-011" -> zwei Bereiche mit gleicher Stammnummer
|
|
- Kurzbereich mit Stammnummer: "834372 007-011" -> 834372007..834372011
|
|
|
|
Returns:
|
|
Liste der gefundenen/expandierten 9-stelligen Nummern.
|
|
"""
|
|
results: list[str] = []
|
|
|
|
# Phase 1: OCR-verklebte Nummern trennen.
|
|
# "005834372" -> "005 834372" (3 Suffix-Ziffern gefolgt von 6-stelliger Stammnr)
|
|
# Wiederhole bis stabil.
|
|
cleaned = text
|
|
for _ in range(10):
|
|
prev = cleaned
|
|
# Nach einem Bereichs-Ende (-NNN) klebt eine Stammnummer (6 Ziffern) dran
|
|
cleaned = re.sub(r'-(\d{3})(\d{6})(?=\s|\d{3}-|\d{3}\s|$)', r'-\1 \2', cleaned)
|
|
if cleaned == prev:
|
|
break
|
|
|
|
# Phase 2: Tokenisieren und mit Kontext parsen.
|
|
# Tokens: 9-stellige Nummern, 6-stellige Stammnummern, NNN-NNN Bereiche,
|
|
# einzelne NNN Suffixe.
|
|
# Die Stammnummer wird aus dem letzten vollstaendigen 9-Steller oder
|
|
# expliziten 6-Steller uebernommen.
|
|
|
|
current_prefix = "" # 6-stellige Stammnummer
|
|
|
|
# Tokenize: Zifferngruppen und Bindestriche extrahieren
|
|
# Wir verarbeiten die Zeile sequentiell
|
|
token_pat = re.compile(r'\d+')
|
|
dash_pat = re.compile(r'(\d+)-(\d+)')
|
|
|
|
# Strategie: alle Zifferngruppen und ihre Kontexte finden
|
|
# Zuerst alle "Nummer-Nummer" Paare und einzelne Nummern identifizieren
|
|
segments: list[str] = []
|
|
# Aufteilen an Nicht-Ziffern-nicht-Bindestrich Grenzen
|
|
# aber Bindestrich zwischen Ziffern beibehalten
|
|
seg_pat = re.compile(r'[\d]+(?:-[\d]+)*')
|
|
segments = seg_pat.findall(cleaned)
|
|
|
|
for seg in segments:
|
|
# Fall 1: "834372007-011" (9 Stellen - 3 Stellen)
|
|
m = re.match(r'^(\d{9})-(\d{3})$', seg)
|
|
if m:
|
|
base = m.group(1)
|
|
current_prefix = base[:6]
|
|
start = int(base[6:])
|
|
end = int(m.group(2))
|
|
if end >= start:
|
|
for i in range(start, end + 1):
|
|
results.append(f"{current_prefix}{i:03d}")
|
|
else:
|
|
results.append(base)
|
|
continue
|
|
|
|
# Fall 2: "834372007" (einzelne 9-stellige Nummer)
|
|
m = re.match(r'^(\d{9})$', seg)
|
|
if m:
|
|
num = m.group(1)
|
|
current_prefix = num[:6]
|
|
if num not in results:
|
|
results.append(num)
|
|
continue
|
|
|
|
# Fall 3: "834372" (6-stellige Stammnummer, setzt Prefix)
|
|
m = re.match(r'^(\d{6})$', seg)
|
|
if m:
|
|
current_prefix = m.group(1)
|
|
continue
|
|
|
|
# Fall 4: "007-011" (3-3 Bereich mit bekanntem Prefix)
|
|
m = re.match(r'^(\d{3})-(\d{3})$', seg)
|
|
if m and current_prefix:
|
|
start = int(m.group(1))
|
|
end = int(m.group(2))
|
|
if end >= start:
|
|
for i in range(start, end + 1):
|
|
results.append(f"{current_prefix}{i:03d}")
|
|
continue
|
|
|
|
# Fall 5: "420" (einzelnes 3-stelliges Suffix mit bekanntem Prefix)
|
|
m = re.match(r'^(\d{3})$', seg)
|
|
if m and current_prefix:
|
|
num = f"{current_prefix}{m.group(1)}"
|
|
if num not in results:
|
|
results.append(num)
|
|
continue
|
|
|
|
# Fall 6: "200-202834372" -> nach Phase 1 Trennung sollte das
|
|
# nicht mehr vorkommen, aber als Fallback:
|
|
# Mehrere Bereiche hintereinander
|
|
sub_segs = re.findall(r'(\d{9})-(\d{3})|(\d{9})|(\d{3})-(\d{3})', seg)
|
|
for ss in sub_segs:
|
|
if ss[0] and ss[1]: # 9-3 Bereich
|
|
base = ss[0]
|
|
current_prefix = base[:6]
|
|
start = int(base[6:])
|
|
end = int(ss[1])
|
|
if end >= start:
|
|
for i in range(start, end + 1):
|
|
results.append(f"{current_prefix}{i:03d}")
|
|
elif ss[2]: # Einzelne 9-stellige
|
|
current_prefix = ss[2][:6]
|
|
if ss[2] not in results:
|
|
results.append(ss[2])
|
|
elif ss[3] and ss[4] and current_prefix: # 3-3 Bereich
|
|
start = int(ss[3])
|
|
end = int(ss[4])
|
|
if end >= start:
|
|
for i in range(start, end + 1):
|
|
results.append(f"{current_prefix}{i:03d}")
|
|
|
|
return results
|
|
|
|
|
|
RE_UNDERLINE = re.compile(r"^([=\-~\^])\1{2,}$")
|
|
|
|
|
|
def collect_from_rst(rst_path: Path) -> dict[str, str]:
|
|
"""Sammelt alle 9-stelligen Teilenummern aus einer RST-Datei.
|
|
|
|
Returns:
|
|
Dict nummer -> kapitelueberschrift in der die Nummer gefunden wurde.
|
|
"""
|
|
lines = rst_path.read_text(encoding="utf-8").splitlines()
|
|
num_to_chapter: dict[str, str] = {}
|
|
current_heading = ""
|
|
|
|
i = 0
|
|
while i < len(lines):
|
|
line = lines[i].rstrip()
|
|
|
|
# RST-Ueberschriften erkennen:
|
|
# Variante 1: overline + title + underline (Level 1)
|
|
# Variante 2: title + underline (Level 2+)
|
|
if RE_UNDERLINE.match(line):
|
|
# Pruefen ob dies eine Underline nach einem Titel ist
|
|
if i > 0 and lines[i - 1].strip():
|
|
title = lines[i - 1].strip()
|
|
# Nur nummerierte Ueberschriften als Kapitel verwenden
|
|
if re.match(r"^\d+(\.\d+)*\s+", title):
|
|
current_heading = title
|
|
# Oder eine Overline vor einem Titel
|
|
elif i + 1 < len(lines) and lines[i + 1].strip():
|
|
title = lines[i + 1].strip()
|
|
if re.match(r"^\d+(\.\d+)*\s+", title):
|
|
current_heading = title
|
|
i += 1
|
|
continue
|
|
|
|
# Teilenummern aus der Zeile extrahieren
|
|
nums = expand_ranges(line)
|
|
for n in nums:
|
|
if n[0] in ("8", "9") and n not in num_to_chapter:
|
|
num_to_chapter[n] = current_heading
|
|
|
|
i += 1
|
|
|
|
return num_to_chapter
|
|
|
|
|
|
def load_excel_descriptions(xlsx_path: Path) -> dict[str, str]:
|
|
"""Laedt Teilenummer -> Bezeichnung aus Excel-Stueckliste."""
|
|
import openpyxl
|
|
|
|
wb = openpyxl.load_workbook(str(xlsx_path), read_only=True)
|
|
ws = wb["Stk.-Liste SOLL"]
|
|
desc_map: dict[str, str] = {}
|
|
|
|
for row in ws.iter_rows(min_row=5, values_only=True):
|
|
tn = str(row[1]).strip() if row[1] else ""
|
|
if re.match(r"^\d{9}$", tn) and tn not in desc_map:
|
|
bez = str(row[2]).strip() if row[2] else ""
|
|
if bez:
|
|
desc_map[tn] = bez
|
|
|
|
wb.close()
|
|
return desc_map
|
|
|
|
|
|
def process(data_dir: Path) -> None:
|
|
pp_dir = data_dir / "Produktprogramm"
|
|
rst_files = sorted(pp_dir.glob("*.de.rst"))
|
|
|
|
if not rst_files:
|
|
log.error("Keine RST-Dateien in %s gefunden", pp_dir)
|
|
sys.exit(1)
|
|
|
|
# Teilenummern sammeln: nummer -> list[(quelldatei, kapitel)]
|
|
numbers: dict[str, list[tuple[str, str]]] = {}
|
|
for rst_path in rst_files:
|
|
basename = rst_path.name
|
|
num_chapters = collect_from_rst(rst_path)
|
|
log.info("%s: %d Teilenummern gefunden", basename, len(num_chapters))
|
|
for num, chapter in num_chapters.items():
|
|
numbers.setdefault(num, []).append((basename, chapter))
|
|
|
|
log.info("Gesamt: %d eindeutige Teilenummern", len(numbers))
|
|
|
|
# Excel-Bezeichnungen laden
|
|
xlsx_files = list((data_dir).glob("KA*.xlsx"))
|
|
desc_map: dict[str, str] = {}
|
|
if xlsx_files:
|
|
xlsx_path = xlsx_files[0]
|
|
log.info("Lade Excel: %s", xlsx_path.name)
|
|
desc_map = load_excel_descriptions(xlsx_path)
|
|
log.info(" %d Bezeichnungen geladen", len(desc_map))
|
|
else:
|
|
log.warning("Keine KA*.xlsx Datei gefunden in %s", data_dir)
|
|
|
|
# CSV schreiben
|
|
outfile = pp_dir / "teilenummern.csv"
|
|
matched = 0
|
|
with open(outfile, "w", encoding="utf-8-sig", newline="") as fh:
|
|
writer = csv.writer(fh, delimiter=";")
|
|
writer.writerow(["Teilenummer", "Quelldatei", "Kapitel", "Bezeichnung"])
|
|
for num in sorted(numbers.keys()):
|
|
entries = sorted(numbers[num], key=lambda e: e[0])
|
|
sources = ", ".join(e[0] for e in entries)
|
|
chapters = ", ".join(e[1] for e in entries if e[1])
|
|
bez = desc_map.get(num, "")
|
|
if bez:
|
|
matched += 1
|
|
writer.writerow([num, sources, chapters, bez])
|
|
|
|
log.info("CSV geschrieben: %s", outfile)
|
|
log.info(" %d Nummern, %d mit Bezeichnung, %d ohne",
|
|
len(numbers), matched, len(numbers) - matched)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(
|
|
description="Sammelt 9-stellige Teilenummern aus Produktprogramm-RSTs"
|
|
)
|
|
parser.add_argument("--data-dir", type=Path, default=None,
|
|
help="data/ Verzeichnis (default: aus PV_DATA oder ./data)")
|
|
parser.add_argument("--verbose", "-v", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(levelname)-7s %(message)s",
|
|
)
|
|
|
|
if args.data_dir:
|
|
data_dir = args.data_dir.resolve()
|
|
elif os.environ.get("PV_DATA"):
|
|
data_dir = Path(os.environ["PV_DATA"]).resolve()
|
|
else:
|
|
data_dir = Path("data").resolve()
|
|
|
|
log.info("Data-Verzeichnis: %s", data_dir)
|
|
process(data_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|