Lars/mindnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

debug mit print

Browse Source
This commit is contained in:
Lars 2025-12-23 11:40:48 +01:00
parent 81d005d969
commit f8ed7bb62e
1 changed files with 24 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
app/services/edge_registry.py
View File

@ -1,7 +1,8 @@
""" """
FILE: app/services/edge_registry.py FILE: app/services/edge_registry.py
DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload. DESCRIPTION: Single Source of Truth für Kanten-Typen mit dynamischem Reload.
VERSION: 0.7.1 (Fix: Silent Path Error & Path Alignment) WP-22: Transparente Status-Meldungen für Dev-Umgebungen.
VERSION: 0.7.2 (Fix: Restore Console Visibility & Entry Counts)
""" """
import re import re
import os import os
@ -16,6 +17,7 @@ logger = logging.getLogger(__name__)
class EdgeRegistry: class EdgeRegistry:
_instance = None _instance = None
# System-Kanten, die NIEMALS manuell im Markdown stehen dürfen
FORBIDDEN_SYSTEM_EDGES = {"next", "prev", "belongs_to"} FORBIDDEN_SYSTEM_EDGES = {"next", "prev", "belongs_to"}
def __new__(cls, *args, **kwargs): def __new__(cls, *args, **kwargs):
@ -32,11 +34,10 @@ class EdgeRegistry:
env_vocab_path = os.getenv("MINDNET_VOCAB_PATH") env_vocab_path = os.getenv("MINDNET_VOCAB_PATH")
env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault") 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) # Pfad-Priorität: 1. ENV -> 2. _system/dictionary -> 3. 01_User_Manual
if env_vocab_path: if env_vocab_path:
self.full_vocab_path = os.path.abspath(env_vocab_path) self.full_vocab_path = os.path.abspath(env_vocab_path)
else: else:
# Suche an zwei Orten: Erst System-Dir, dann User-Manual
possible_paths = [ possible_paths = [
os.path.join(env_vault_root, "_system", "dictionary", "edge_vocabulary.md"), os.path.join(env_vault_root, "_system", "dictionary", "edge_vocabulary.md"),
os.path.join(env_vault_root, "01_User_Manual", "01_edge_vocabulary.md") os.path.join(env_vault_root, "01_User_Manual", "01_edge_vocabulary.md")
@ -48,21 +49,22 @@ class EdgeRegistry:
break break
if not self.full_vocab_path: if not self.full_vocab_path:
self.full_vocab_path = os.path.abspath(possible_paths[0]) # Fallback self.full_vocab_path = os.path.abspath(possible_paths[0])
self.unknown_log_path = "data/logs/unknown_edges.jsonl" self.unknown_log_path = "data/logs/unknown_edges.jsonl"
self.canonical_map: Dict[str, str] = {} self.canonical_map: Dict[str, str] = {}
self.valid_types: Set[str] = set() self.valid_types: Set[str] = set()
self._last_mtime = 0.0 self._last_mtime = 0.0
# Initialer Lade-Versuch mit Konsolen-Feedback
print(f"\n>>> [EDGE-REGISTRY] Initializing with Path: {self.full_vocab_path}", flush=True)
self.ensure_latest() self.ensure_latest()
self.initialized = True self.initialized = True
def ensure_latest(self): def ensure_latest(self):
"""Prüft den Zeitstempel und meldet Fehler nun explizit.""" """Prüft den Zeitstempel und lädt bei Bedarf neu."""
if not os.path.exists(self.full_vocab_path): if not os.path.exists(self.full_vocab_path):
# WP-22 FIX: Fehlermeldung statt silent return print(f"!!! [EDGE-REGISTRY ERROR] File not found: {self.full_vocab_path} !!!", flush=True)
logger.error(f"❌ EdgeRegistry: Vocabulary file NOT FOUND at {self.full_vocab_path}")
return return
current_mtime = os.path.getmtime(self.full_vocab_path) current_mtime = os.path.getmtime(self.full_vocab_path)
@ -71,11 +73,11 @@ class EdgeRegistry:
self._last_mtime = current_mtime self._last_mtime = current_mtime
def _load_vocabulary(self): def _load_vocabulary(self):
"""Parst das Wörterbuch mit verbessertem Logging.""" """Parst das Wörterbuch und meldet die Anzahl der gelesenen Einträge."""
self.canonical_map.clear() self.canonical_map.clear()
self.valid_types.clear() self.valid_types.clear()
# Regex deckt **typ** und **`typ`** ab # Regex deckt | **canonical** | Aliase | ab
pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|") pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|")
try: try:
@ -94,35 +96,46 @@ class EdgeRegistry:
if aliases_str and "Kein Alias" not in aliases_str: if aliases_str and "Kein Alias" not in aliases_str:
aliases = [a.strip() for a in aliases_str.split(",") if a.strip()] aliases = [a.strip() for a in aliases_str.split(",") if a.strip()]
for alias in aliases: for alias in aliases:
# Normalisierung: Kleinschreibung und Unterstriche
clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_") clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_")
self.canonical_map[clean_alias] = canonical self.canonical_map[clean_alias] = canonical
c_aliases += 1 c_aliases += 1
logger.info(f"✅ EdgeRegistry reloaded: {c_types} types and {c_aliases} aliases from {self.full_vocab_path}") # 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: except Exception as e:
logger.error(f"❌ EdgeRegistry: Error reading vocabulary: {e}") print(f"!!! [EDGE-REGISTRY FATAL] Error reading file: {e} !!!", flush=True)
logger.error(f"Error reading vocabulary: {e}")
def resolve(self, edge_type: str, provenance: str = "explicit", context: dict = None) -> str: def resolve(self, edge_type: str, provenance: str = "explicit", context: dict = None) -> str:
"""Validierung mit Fundort-Logging."""
self.ensure_latest() self.ensure_latest()
if not edge_type: return "related_to" 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 {} ctx = context or {}
# 1. Schutz der Systemkanten (Verbot für manuelle Nutzung)
if provenance == "explicit" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: if provenance == "explicit" and clean_type in self.FORBIDDEN_SYSTEM_EDGES:
self._log_issue(clean_type, "forbidden_system_usage", ctx) self._log_issue(clean_type, "forbidden_system_usage", ctx)
return "related_to" return "related_to"
# 2. Akzeptanz interner Strukturkanten
if provenance == "structure" and clean_type in self.FORBIDDEN_SYSTEM_EDGES: if provenance == "structure" and clean_type in self.FORBIDDEN_SYSTEM_EDGES:
return clean_type return clean_type
# 3. Mapping via Wörterbuch
if clean_type in self.canonical_map: if clean_type in self.canonical_map:
return self.canonical_map[clean_type] return self.canonical_map[clean_type]
# 4. Unbekannte Kante
self._log_issue(clean_type, "unknown_type", ctx) self._log_issue(clean_type, "unknown_type", ctx)
return clean_type return clean_type
def _log_issue(self, edge_type: str, error_kind: str, ctx: dict): def _log_issue(self, edge_type: str, error_kind: str, ctx: dict):
"""Detailliertes JSONL-Logging für Debugging."""
try: try:
os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True) os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True)
entry = { entry = {