diff --git a/app/services/edge_registry.py b/app/services/edge_registry.py index b9d3535..fac56b6 100644 --- a/app/services/edge_registry.py +++ b/app/services/edge_registry.py @@ -1,9 +1,9 @@ """ FILE: app/services/edge_registry.py -DESCRIPTION: Single Source of Truth für Kanten-Typen. Parst '01_edge_vocabulary.md'. +DESCRIPTION: Single Source of Truth für Kanten-Typen. WP-22 Teil B: Registry & Validation. - FIX: Beachtet MINDNET_VOCAB_PATH und MINDNET_VAULT_ROOT aus .env sicher. -VERSION: 0.6.5 (Path Stability Update) + HARD-LOGGING: Nutzt print() und logger für maximale Sichtbarkeit beim Systemstart. +VERSION: 0.6.6 (Diagnostic Update) """ import re import os @@ -18,7 +18,7 @@ logger = logging.getLogger(__name__) class EdgeRegistry: _instance = None - def __new__(cls, vault_root: Optional[str] = None): + def __new__(cls, *args, **kwargs): if cls._instance is None: cls._instance = super(EdgeRegistry, cls).__new__(cls) cls._instance.initialized = False @@ -28,29 +28,20 @@ class EdgeRegistry: if self.initialized: return + # Sofortige Ausgabe beim Start (erscheint in journalctl) + print("--- [DICT-INIT] EdgeRegistry startup sequence initiated ---") + + # Pfad-Ermittlung settings = get_settings() + env_vocab_path = os.getenv("MINDNET_VOCAB_PATH") + env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault_master") - # --- 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) + # Absolute Pfad-Konstruktion + if env_vocab_path: + self.full_vocab_path = os.path.abspath(env_vocab_path) else: self.full_vocab_path = os.path.abspath( - os.path.join(self.vault_root, "01_User_Manual", "01_edge_vocabulary.md") + os.path.join(env_vault_root, "01_User_Manual", "01_edge_vocabulary.md") ) self.unknown_log_path = "data/logs/unknown_edges.jsonl" @@ -62,11 +53,14 @@ class EdgeRegistry: def _load_vocabulary(self): """Parst die Markdown-Tabelle im Vault.""" + print(f"--- [DICT-CHECK] Attempting to load: {self.full_vocab_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.") + msg = f"!!! [DICT-ERROR] Edge Vocabulary NOT found at: {self.full_vocab_path} !!!" + print(msg) + logger.warning(msg) return - # Regex für Markdown Tabellen: | **canonical** | Aliases | ... pattern = re.compile(r"\|\s*\*\*([a-z_]+)\*\*\s*\|\s*([^|]+)\|") try: @@ -84,43 +78,39 @@ class EdgeRegistry: 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 {count_types} types and {count_aliases} aliases from {self.full_vocab_path}.") + success_msg = f"=== [DICT-SUCCESS] Loaded {count_types} Types and {count_aliases} Aliases ===" + print(success_msg) + logger.info(success_msg) except Exception as e: - logger.error(f"Failed to parse Edge Vocabulary at {self.full_vocab_path}: {e}") + err_msg = f"!!! [DICT-FATAL] Failed to parse Vocabulary: {e} !!!" + print(err_msg) + logger.error(err_msg) def resolve(self, edge_type: str) -> str: """Normalisiert Kanten-Typen via Registry oder loggt Unbekannte.""" - if not edge_type: - return "related_to" - - # Normalisierung: Kleinschreibung und Unterstriche statt Leerzeichen + if not edge_type: return "related_to" 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 def _log_unknown(self, edge_type: str): - """Schreibt unbekannte Typen für Review in ein Log.""" try: os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True) entry = {"unknown_type": edge_type, "status": "new"} with open(self.unknown_log_path, "a", encoding="utf-8") as f: f.write(json.dumps(entry) + "\n") - except Exception: - pass + except Exception: pass # Singleton Instanz registry = EdgeRegistry() \ No newline at end of file