"""Extract 9-digit Sivas IDs from JSON tree files generated by jsonFromXlsx.py. Collects Sivas numbers that sit one hierarchy level above the leaves and writes them as 'nummer: bezeichnung' into a text file. Usage: python extract_ids.py --filename KA135776 python extract_ids.py --filename KA135776 --tab soll python extract_ids.py --all python extract_ids.py --all --sum2json python extract_ids.py --all --sum2txt """ import argparse import json import os import re import sys from pathlib import Path RE_SIVAS = re.compile(r"^(\d{9})") def is_leaf_value(v) -> bool: """True if v is a leaf entry (string, anzahl-dict, or laenge-dict).""" if isinstance(v, str): return True if isinstance(v, dict) and ("anzahl" in v or "laenge" in v): return True return False def extract_ids_from_tree(tree: dict) -> dict[str, str]: """Collect 9-digit Sivas numbers one level above leaves. Returns dict of nummer -> bezeichnung. """ ids: dict[str, str] = {} def walk(node: dict): for key, value in node.items(): if not isinstance(value, dict) or "anzahl" in value or "laenge" in value: continue # leaf, skip # Check if all children are leaves all_leaves = all(is_leaf_value(v) for v in value.values()) m = RE_SIVAS.match(key) if all_leaves and value and m: nummer = m.group(1) # Bezeichnung from key: part after "nummer-", underscores back to spaces if len(key) > 9: bez = key[10:].replace("_", " ") else: bez = nummer ids[nummer] = bez else: walk(value) walk(tree) return ids def get_project_name(json_path: Path) -> str: """Derive project name from JSON filename by stripping _ist/_soll suffix.""" project_name = json_path.stem for suffix in ("_ist", "_soll"): if project_name.endswith(suffix): project_name = project_name[: -len(suffix)] break return project_name def process_json(json_path: Path, tab: str, results_dir: Path) -> tuple[dict[str, str], str, Path]: """Extract IDs from one JSON file. Returns (ids dict, project_name, ids_path).""" with open(json_path, encoding="utf-8") as f: tree = json.load(f) ids = extract_ids_from_tree(tree) project_name = get_project_name(json_path) ids_path = results_dir / f"{project_name}_ids_{tab}.txt" with open(ids_path, "w", encoding="utf-8") as f: for nummer, bez in sorted(ids.items()): f.write(f"{nummer}: {bez}\n") print(f" {json_path.name} -> {ids_path.name} ({len(ids)} IDs)") return ids, project_name, ids_path def write_sum2json(all_ids: dict[str, str], projekte: dict[str, list[str]], tab: str, results_dir: Path): """Write aggregated summary as JSON file.""" # projekte dict: convert lists to comma-separated strings projekte_str = {k: ", ".join(v) for k, v in sorted(projekte.items())} data = { "ids": dict(sorted(all_ids.items())), "projekte": projekte_str, } out_path = results_dir / f"summary_ids_{tab}.json" with open(out_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) print(f"\n Summary JSON -> {out_path.name} ({len(all_ids)} IDs, {len(projekte)} mit Projektzuordnung)") def write_sum2txt(all_ids: dict[str, str], projekte: dict[str, list[str]], tab: str, results_dir: Path): """Write aggregated summary as TXT file with [Sivasnummern] and [Anzahl] sections.""" out_path = results_dir / f"summary_ids_{tab}.txt" with open(out_path, "w", encoding="utf-8") as f: f.write("[Sivasnummern]\n") for nummer, bez in sorted(all_ids.items()): f.write(f"{nummer}: {bez}\n") f.write(f"\n[Anzahl]\n") for nummer, prj_list in sorted(projekte.items(), key=lambda x: len(x[1]), reverse=True): f.write(f"{nummer}: {len(prj_list)}\n") print(f"\n Summary TXT -> {out_path.name} ({len(all_ids)} IDs)") def main(): parser = argparse.ArgumentParser( description="Extrahiert 9-stellige Sivas-Nummern aus JSON-Baeumen im results/-Ordner." ) parser.add_argument( "--filename", default=None, help="Suchmuster fuer JSON-Dateien im results/-Ordner (z.B. 'KA135776')", ) parser.add_argument( "--tab", choices=["soll", "ist"], default="ist", help="Welches Tab gelesen wird: 'ist' oder 'soll' (default: ist)", ) parser.add_argument( "--all", action="store_true", dest="process_all", help="Alle *_.json Dateien im results/-Ordner verarbeiten", ) parser.add_argument( "--sum2json", action="store_true", help="Alle IDs in eine summary_ids_.json zusammenfassen (ids + projekte)", ) parser.add_argument( "--sum2txt", action="store_true", help="Alle IDs in eine summary_ids_.txt zusammenfassen ([Sivasnummern] + [Anzahl])", ) args = parser.parse_args() if not args.process_all and not args.filename: parser.error("Entweder --filename oder --all muss angegeben werden.") project_root = Path(os.environ.get("PROJECT", Path(__file__).resolve().parent.parent)) results_dir = project_root / "results" if not results_dir.exists(): print(f"FEHLER: results/-Ordner nicht gefunden: {results_dir}") sys.exit(1) pattern = f"*_{args.tab}.json" json_files = sorted(results_dir.glob(pattern)) if args.filename: json_files = [f for f in json_files if args.filename.lower() in f.name.lower()] if not json_files: print(f"FEHLER: Keine JSON-Dateien mit Muster '{pattern}' in {results_dir} gefunden.") if args.filename: print(f" (Filter: '{args.filename}')") sys.exit(1) print(f"Verarbeite {len(json_files)} JSON-Dateien aus {results_dir}\n") total_ids = 0 # Aggregation dicts for --sum2json / --sum2txt all_ids: dict[str, str] = {} projekte: dict[str, list[str]] = {} individual_txt_files: list[Path] = [] for json_path in json_files: ids, project_name, ids_path = process_json(json_path, args.tab, results_dir) total_ids += len(ids) individual_txt_files.append(ids_path) for nummer, bez in ids.items(): all_ids[nummer] = bez projekte.setdefault(nummer, []).append(project_name) print(f"\nFertig: {total_ids} IDs aus {len(json_files)} Dateien extrahiert.") if args.sum2json: write_sum2json(all_ids, projekte, args.tab, results_dir) if args.sum2txt: write_sum2txt(all_ids, projekte, args.tab, results_dir) # Einzelne _ids_ txt-Dateien loeschen wenn Summary erzeugt wurde if args.sum2json or args.sum2txt: for txt_path in individual_txt_files: txt_path.unlink(missing_ok=True) print(f" {len(individual_txt_files)} einzelne _ids_ Dateien geloescht.") if __name__ == "__main__": main()