From 81d005d9691a7eee905f7bb4cbadc5638739e836 Mon Sep 17 00:00:00 2001 From: Lars Date: Tue, 23 Dec 2025 11:34:55 +0100 Subject: [PATCH] =?UTF-8?q?pfad=20Aufl=C3=B6sung?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/edge_registry.py | 55 ++++++++++++++++++++++------------- 1 file changed, 34 insertions(+), 21 deletions(-) diff --git a/app/services/edge_registry.py b/app/services/edge_registry.py index c23c20c..c22bc0b 100644 --- a/app/services/edge_registry.py +++ b/app/services/edge_registry.py @@ -1,3 +1,8 @@ +""" +FILE: app/services/edge_registry.py +DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload. +VERSION: 0.7.1 (Fix: Silent Path Error & Path Alignment) +""" import re import os import json @@ -11,7 +16,6 @@ logger = logging.getLogger(__name__) class EdgeRegistry: _instance = None - # System-Kanten, die NIEMALS manuell im Markdown stehen dürfen FORBIDDEN_SYSTEM_EDGES = {"next", "prev", "belongs_to"} def __new__(cls, *args, **kwargs): @@ -28,24 +32,37 @@ class EdgeRegistry: env_vocab_path = os.getenv("MINDNET_VOCAB_PATH") env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault") + # WP-22 FIX: Pfad-Priorität angepasst auf _system/dictionary (Architektur-Standard) if env_vocab_path: self.full_vocab_path = os.path.abspath(env_vocab_path) else: - self.full_vocab_path = os.path.abspath( + # Suche an zwei Orten: Erst System-Dir, dann User-Manual + possible_paths = [ + os.path.join(env_vault_root, "_system", "dictionary", "edge_vocabulary.md"), os.path.join(env_vault_root, "01_User_Manual", "01_edge_vocabulary.md") - ) - + ] + self.full_vocab_path = None + for p in possible_paths: + if os.path.exists(p): + self.full_vocab_path = os.path.abspath(p) + break + + if not self.full_vocab_path: + self.full_vocab_path = os.path.abspath(possible_paths[0]) # Fallback + self.unknown_log_path = "data/logs/unknown_edges.jsonl" self.canonical_map: Dict[str, str] = {} self.valid_types: Set[str] = set() - self._last_mtime = 0.0 # Für dynamisches Neuladen + self._last_mtime = 0.0 self.ensure_latest() self.initialized = True def ensure_latest(self): - """Prüft den Zeitstempel der Datei und lädt ggf. neu.""" + """Prüft den Zeitstempel und meldet Fehler nun explizit.""" if not os.path.exists(self.full_vocab_path): + # WP-22 FIX: Fehlermeldung statt silent return + logger.error(f"❌ EdgeRegistry: Vocabulary file NOT FOUND at {self.full_vocab_path}") return current_mtime = os.path.getmtime(self.full_vocab_path) @@ -54,14 +71,16 @@ class EdgeRegistry: self._last_mtime = current_mtime def _load_vocabulary(self): - """Parst das Wörterbuch und baut die Canonical-Map auf.""" + """Parst das Wörterbuch mit verbessertem Logging.""" self.canonical_map.clear() self.valid_types.clear() + # Regex deckt **typ** und **`typ`** ab pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|") try: with open(self.full_vocab_path, "r", encoding="utf-8") as f: + c_types, c_aliases = 0, 0 for line in f: match = pattern.search(line) if match: @@ -70,47 +89,40 @@ class EdgeRegistry: self.valid_types.add(canonical) self.canonical_map[canonical] = canonical + c_types += 1 if aliases_str and "Kein Alias" not in aliases_str: aliases = [a.strip() for a in aliases_str.split(",") if a.strip()] for alias in aliases: clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_") self.canonical_map[clean_alias] = canonical - logger.info(f"EdgeRegistry reloaded: {len(self.valid_types)} types.") + c_aliases += 1 + + logger.info(f"✅ EdgeRegistry reloaded: {c_types} types and {c_aliases} aliases from {self.full_vocab_path}") except Exception as e: - logger.error(f"Error reading vocabulary: {e}") + logger.error(f"❌ EdgeRegistry: Error reading vocabulary: {e}") def resolve(self, edge_type: str, provenance: str = "explicit", context: dict = None) -> str: - """ - Validiert Kanten gegen System-Regeln und Wörterbuch. - - provenance: 'explicit' (User/Markdown) oder 'structure' (System-Pipeline) - - context: {'file': str, 'line': int} - """ self.ensure_latest() if not edge_type: return "related_to" clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_") ctx = context or {} - # 1. Check auf "verbotene" manuelle Nutzung von Systemkanten if provenance == "explicit" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: self._log_issue(clean_type, "forbidden_system_usage", ctx) - return "related_to" # Fallback, um System-Integrität zu schützen + return "related_to" - # 2. Stillschweigende Akzeptanz für echte System-Struktur if provenance == "structure" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: return clean_type - # 3. Mapping gegen das Wörterbuch if clean_type in self.canonical_map: return self.canonical_map[clean_type] - # 4. Unbekannte Kante loggen self._log_issue(clean_type, "unknown_type", ctx) return clean_type def _log_issue(self, edge_type: str, error_kind: str, ctx: dict): - """Schreibt detaillierte Fehler mit Fundort.""" try: os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True) entry = { @@ -118,7 +130,8 @@ class EdgeRegistry: "edge_type": edge_type, "error": error_kind, "file": ctx.get("file", "unknown"), - "line": ctx.get("line", "unknown") + "line": ctx.get("line", "unknown"), + "note_id": ctx.get("note_id", "unknown") } with open(self.unknown_log_path, "a", encoding="utf-8") as f: f.write(json.dumps(entry) + "\n")