diff --git a/app/services/edge_registry.py b/app/services/edge_registry.py index 4e9cb85..2859baa 100644 --- a/app/services/edge_registry.py +++ b/app/services/edge_registry.py @@ -1,16 +1,15 @@ """ FILE: app/services/edge_registry.py -DESCRIPTION: Single Source of Truth für Kanten-Typen. - FIX: Regex angepasst auf Format **`canonical`** (Bold + Backticks). -VERSION: 0.6.10 (Regex Precision Update) +DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload. + WP-22: Transparente Status-Meldungen für Dev-Umgebungen. +VERSION: 0.7.2 (Fix: Restore Console Visibility & Entry Counts) """ import re import os import json import logging -from typing import Dict, Optional, Set - -print(">>> MODULE_LOAD: edge_registry.py initialized <<<", flush=True) +import time +from typing import Dict, Optional, Set, Tuple from app.config import get_settings @@ -18,6 +17,8 @@ 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): if cls._instance is None: @@ -33,31 +34,50 @@ 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") + # Pfad-Priorität: 1. ENV -> 2. _system/dictionary -> 3. 01_User_Manual if env_vocab_path: self.full_vocab_path = os.path.abspath(env_vocab_path) else: - self.full_vocab_path = os.path.abspath( + 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]) + 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 - self._load_vocabulary() + # Initialer Lade-Versuch mit Konsolen-Feedback + print(f"\n>>> [EDGE-REGISTRY] Initializing with Path: {self.full_vocab_path}", flush=True) + self.ensure_latest() self.initialized = True - def _load_vocabulary(self): - """Parst die Markdown-Tabelle im Vault.""" - print(f">>> CHECK: Loading Vocabulary from {self.full_vocab_path}", flush=True) - + def ensure_latest(self): + """Prüft den Zeitstempel und lädt bei Bedarf neu.""" if not os.path.exists(self.full_vocab_path): - print(f"!!! [DICT-ERROR] File not found: {self.full_vocab_path} !!!", flush=True) + print(f"!!! [EDGE-REGISTRY ERROR] File not found: {self.full_vocab_path} !!!", flush=True) return - # WP-22 Precision Regex: - # Sucht nach | **`typ`** | oder | **typ** | - # Die Backticks `? sind jetzt optional enthalten. + current_mtime = os.path.getmtime(self.full_vocab_path) + if current_mtime > self._last_mtime: + self._load_vocabulary() + self._last_mtime = current_mtime + + def _load_vocabulary(self): + """Parst das Wörterbuch und meldet die Anzahl der gelesenen Einträge.""" + self.canonical_map.clear() + self.valid_types.clear() + + # Regex deckt | **canonical** | Aliase | ab pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|") try: @@ -74,36 +94,58 @@ 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: + # Normalisierung: Kleinschreibung und Unterstriche clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_") self.canonical_map[clean_alias] = canonical c_aliases += 1 - 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) + # Erfolgskontrolle für das Dev-Terminal + print(f"=== [EDGE-REGISTRY SUCCESS] Loaded {c_types} Canonical Types and {c_aliases} Aliases ===", flush=True) + logger.info(f"Registry reloaded from {self.full_vocab_path}") except Exception as e: - print(f"!!! [DICT-FATAL] Error reading file: {e} !!!", flush=True) + print(f"!!! [EDGE-REGISTRY FATAL] Error reading file: {e} !!!", flush=True) + logger.error(f"Error reading vocabulary: {e}") - def resolve(self, edge_type: str) -> str: - """Normalisiert Kanten-Typen via Registry oder loggt Unbekannte.""" + def resolve(self, edge_type: str, provenance: str = "explicit", context: dict = None) -> str: + """Validierung mit Fundort-Logging.""" + self.ensure_latest() if not edge_type: return "related_to" - clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_") + clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_") + ctx = context or {} + + # 1. Schutz der Systemkanten (Verbot für manuelle Nutzung) + if provenance == "explicit" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: + self._log_issue(clean_type, "forbidden_system_usage", ctx) + return "related_to" + + # 2. Akzeptanz interner Strukturkanten + if provenance == "structure" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: + return clean_type + + # 3. Mapping via Wörterbuch if clean_type in self.canonical_map: return self.canonical_map[clean_type] - self._log_unknown(clean_type) + # 4. Unbekannte Kante + self._log_issue(clean_type, "unknown_type", ctx) return clean_type - def _log_unknown(self, edge_type: str): + def _log_issue(self, edge_type: str, error_kind: str, ctx: dict): + """Detailliertes JSONL-Logging für Debugging.""" try: os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True) - entry = {"unknown_type": edge_type, "status": "new"} + entry = { + "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), + "edge_type": edge_type, + "error": error_kind, + "file": ctx.get("file", "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") except Exception: pass