From 0d71f41a135a6fdc841208c32c2ebd6fbee859a6 Mon Sep 17 00:00:00 2001 From: Lars Date: Thu, 18 Dec 2025 18:53:01 +0100 Subject: [PATCH] dictionary parsen --- app/services/edge_registry.py | 47 ++++++++++++++++------------------- 1 file changed, 21 insertions(+), 26 deletions(-) diff --git a/app/services/edge_registry.py b/app/services/edge_registry.py index e50bdbd..4e3f674 100644 --- a/app/services/edge_registry.py +++ b/app/services/edge_registry.py @@ -1,9 +1,8 @@ """ FILE: app/services/edge_registry.py DESCRIPTION: Single Source of Truth für Kanten-Typen. - WP-22 Teil B: Registry & Validation. - FIX: Erzwingt sofortige Log-Ausgabe beim Laden des Moduls. -VERSION: 0.6.8 (Emergency Debug) + FIX: Regex angepasst auf Format **`canonical`** (Bold + Backticks). +VERSION: 0.6.10 (Regex Precision Update) """ import re import os @@ -11,8 +10,7 @@ import json import logging from typing import Dict, Optional, Set -# Sofortiger Print beim Laden des Moduls (erscheint garantiert in journalctl) -print(">>> MODULE_LOAD: edge_registry.py is being read by Python <<<", flush=True) +print(">>> MODULE_LOAD: edge_registry.py initialized <<<", flush=True) from app.config import get_settings @@ -23,7 +21,6 @@ class EdgeRegistry: def __new__(cls, *args, **kwargs): if cls._instance is None: - print(">>> SINGLETON: Creating new EdgeRegistry instance <<<", flush=True) cls._instance = super(EdgeRegistry, cls).__new__(cls) cls._instance.initialized = False return cls._instance @@ -32,10 +29,7 @@ class EdgeRegistry: if self.initialized: return - print(">>> INIT: EdgeRegistry.__init__ started <<<", flush=True) - settings = get_settings() - # Prio: MINDNET_VOCAB_PATH > MINDNET_VAULT_ROOT > Default env_vocab_path = os.getenv("MINDNET_VOCAB_PATH") env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault_master") @@ -55,16 +49,16 @@ class EdgeRegistry: def _load_vocabulary(self): """Parst die Markdown-Tabelle im Vault.""" - print(f">>> CHECK: Searching for Vocabulary at {self.full_vocab_path}", flush=True) + print(f">>> CHECK: Loading Vocabulary from {self.full_vocab_path}", flush=True) if not os.path.exists(self.full_vocab_path): - msg = f"!!! [DICT-ERROR] Edge Vocabulary NOT found at: {self.full_vocab_path} !!!" - print(msg, flush=True) - logger.warning(msg) + print(f"!!! [DICT-ERROR] File not found: {self.full_vocab_path} !!!", flush=True) return - # Regex für Markdown Tabellen - pattern = re.compile(r"\|\s*\*\*([a-z_]+)\*\*\s*\|\s*([^|]+)\|") + # WP-22 Precision Regex: + # Sucht nach | **`typ`** | oder | **typ** | + # Die Backticks `? sind jetzt optional enthalten. + pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|") try: with open(self.full_vocab_path, "r", encoding="utf-8") as f: @@ -72,7 +66,7 @@ class EdgeRegistry: for line in f: match = pattern.search(line) if match: - canonical = match.group(1).strip() + canonical = match.group(1).strip().lower() aliases_str = match.group(2).strip() self.valid_types.add(canonical) @@ -80,26 +74,29 @@ class EdgeRegistry: c_types += 1 if aliases_str and "Kein Alias" not in aliases_str: + # Aliase säubern (entfernt Backticks auch hier) aliases = [a.strip() for a in aliases_str.split(",") if a.strip()] for alias in aliases: - clean_alias = alias.replace("`", "").lower().strip() + clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_") self.canonical_map[clean_alias] = canonical c_aliases += 1 - success_msg = f"=== [DICT-SUCCESS] Loaded {c_types} Types and {c_aliases} Aliases ===" - print(success_msg, flush=True) - logger.info(success_msg) + if c_types == 0: + print("!!! [DICT-WARN] Pattern mismatch! Ensure types are **`canonical`** or **canonical**. !!!", flush=True) + else: + print(f"=== [DICT-SUCCESS] Registered {c_types} Canonical Types and {c_aliases} Aliases ===", flush=True) except Exception as e: - err_msg = f"!!! [DICT-FATAL] Failed to parse: {e} !!!" - print(err_msg, flush=True) - logger.error(err_msg) + print(f"!!! [DICT-FATAL] Error reading file: {e} !!!", flush=True) def resolve(self, edge_type: str) -> str: + """Normalisiert Kanten-Typen via Registry oder loggt Unbekannte.""" if not edge_type: return "related_to" - clean_type = edge_type.lower().strip().replace(" ", "_") + clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_") + if clean_type in self.canonical_map: return self.canonical_map[clean_type] + self._log_unknown(clean_type) return clean_type @@ -111,6 +108,4 @@ class EdgeRegistry: f.write(json.dumps(entry) + "\n") except Exception: pass -# Singleton Instanz - Dies triggert beim Import die Initialisierung -print(">>> INSTANTIATING REGISTRY SINGLETON <<<", flush=True) registry = EdgeRegistry() \ No newline at end of file