Pfad auflösen zum dictionary

app/services/edge_registry.py

@@ -1,8 +1,9 @@
 """
 FILE: app/services/edge_registry.py
-DESCRIPTION: Single Source of Truth für Kanten-Typen. Parst '01_User_Manual/01_edge_vocabulary.md'.
+DESCRIPTION: Single Source of Truth für Kanten-Typen. Parst '01_edge_vocabulary.md'.
              WP-22 Teil B: Registry & Validation.
-             FIX: Beachtet MINDNET_VAULT_ROOT aus .env korrekt.
+             FIX: Beachtet MINDNET_VOCAB_PATH und MINDNET_VAULT_ROOT aus .env sicher.
+VERSION: 0.6.5 (Path Stability Update)
 """
 import re
 import os
@@ -27,12 +28,32 @@ class EdgeRegistry:
         if self.initialized: 
             return
             
-        # Priorität: 1. Parameter (Test) -> 2. ENV -> 3. Default
         settings = get_settings()
-        self.vault_root = vault_root or getattr(settings, "MINDNET_VAULT_ROOT", "./vault")
-        self.vocab_rel_path = os.path.join("01_User_Manual", "01_edge_vocabulary.md")
-        self.unknown_log_path = "data/logs/unknown_edges.jsonl"
         
+        # --- WP-22 Pfad-Priorisierung ---
+        # 1. Expliziter Vokabular-Pfad (höchste Prio)
+        vocab_env = os.getenv("MINDNET_VOCAB_PATH")
+        
+        # 2. Vault Root Auflösung
+        # Wir prüfen os.getenv direkt, falls Pydantic Settings (settings.MINDNET_VAULT_ROOT) 
+        # das Feld nicht definiert haben.
+        self.vault_root = (
+            vault_root or 
+            vocab_env or 
+            os.getenv("MINDNET_VAULT_ROOT") or 
+            getattr(settings, "MINDNET_VAULT_ROOT", "./vault")
+        )
+        
+        # Falls vocab_env eine komplette Datei ist, nutzen wir sie direkt, 
+        # ansonsten bauen wir den Pfad relativ zur vault_root.
+        if vocab_env and vocab_env.endswith(".md"):
+            self.full_vocab_path = os.path.abspath(vocab_env)
+        else:
+            self.full_vocab_path = os.path.abspath(
+                os.path.join(self.vault_root, "01_User_Manual", "01_edge_vocabulary.md")
+            )
+        
+        self.unknown_log_path = "data/logs/unknown_edges.jsonl"
         self.canonical_map: Dict[str, str] = {} 
         self.valid_types: Set[str] = set()
         
@@ -41,17 +62,17 @@ class EdgeRegistry:
 
     def _load_vocabulary(self):
         """Parst die Markdown-Tabelle im Vault."""
-        full_path = os.path.abspath(os.path.join(self.vault_root, self.vocab_rel_path))
+        if not os.path.exists(self.full_vocab_path):
-        
+            logger.warning(f"Edge Vocabulary NOT found at: {self.full_vocab_path}. Registry is empty.")
-        if not os.path.exists(full_path):
-            logger.warning(f"Edge Vocabulary NOT found at: {full_path}. Registry is empty.")
             return
 
         # Regex für Markdown Tabellen: | **canonical** | Aliases | ...
         pattern = re.compile(r"\|\s*\*\*([a-z_]+)\*\*\s*\|\s*([^|]+)\|")
 
         try:
-            with open(full_path, "r", encoding="utf-8") as f:
+            with open(self.full_vocab_path, "r", encoding="utf-8") as f:
+                count_types = 0
+                count_aliases = 0
                 for line in f:
                     match = pattern.search(line)
                     if match:
@@ -60,26 +81,34 @@ class EdgeRegistry:
                         
                         self.valid_types.add(canonical)
                         self.canonical_map[canonical] = canonical
+                        count_types += 1
                         
                         if aliases_str and "Kein Alias" not in aliases_str:
+                            # Aliase säubern und mappen
                             aliases = [a.strip() for a in aliases_str.split(",") if a.strip()]
                             for alias in aliases:
                                 clean_alias = alias.replace("`", "").lower().strip()
                                 self.canonical_map[clean_alias] = canonical
+                                count_aliases += 1
             
-            logger.info(f"EdgeRegistry loaded from {full_path}: {len(self.valid_types)} types.")
+            logger.info(f"EdgeRegistry: Loaded {count_types} types and {count_aliases} aliases from {self.full_vocab_path}.")
 
         except Exception as e:
-            logger.error(f"Failed to parse Edge Vocabulary at {full_path}: {e}")
+            logger.error(f"Failed to parse Edge Vocabulary at {self.full_vocab_path}: {e}")
 
     def resolve(self, edge_type: str) -> str:
         """Normalisiert Kanten-Typen via Registry oder loggt Unbekannte."""
-        if not edge_type: return "related_to"
+        if not edge_type: 
+            return "related_to"
+            
+        # Normalisierung: Kleinschreibung und Unterstriche statt Leerzeichen
         clean_type = edge_type.lower().strip().replace(" ", "_")
         
+        # 1. Direkter Match (Kanonisch oder Alias)
         if clean_type in self.canonical_map:
             return self.canonical_map[clean_type]
         
+        # 2. Unbekannt -> Logging & Fallback
         self._log_unknown(clean_type)
         return clean_type 
 

vault_master/01_User_Manual/01_edge_vocabulary.md

@@ -27,4 +27,5 @@ context: "Zentrales Wörterbuch für Kanten-Bezeichner. Dient als Single Source
 | **`prev`** | `davor`, `vorgänger`, `preceded_by` | Prozess: B <- A. |
 | **`related_to`** | `siehe_auch`, `kontext`, `thematisch` | Lose Assoziation. |
 | **`similar_to`** | `ähnlich_wie`, `vergleichbar` | Synonym / Ähnlichkeit. |
+| **`experienced_in`** | `erfahren_in`, `spezialisiert_in` | Synonym / Ähnlichkeit. |
 | **`references`** | *(Kein Alias)* | Standard-Verweis (Fallback). |