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

dictionary parsen

Browse Source
This commit is contained in:
Lars 2025-12-18 18:53:01 +01:00
parent f5bfb0cfb4
commit 0d71f41a13
1 changed files with 21 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
app/services/edge_registry.py
View File

@ -1,9 +1,8 @@
"""
FILE: app/services/edge_registry.py
DESCRIPTION: Single Source of Truth für Kanten-Typen.
WP-22 Teil B: Registry & Validation.
FIX: Erzwingt sofortige Log-Ausgabe beim Laden des Moduls.
VERSION: 0.6.8 (Emergency Debug)
FIX: Regex angepasst auf Format **`canonical`** (Bold + Backticks).
VERSION: 0.6.10 (Regex Precision Update)
"""
import re
import os
@ -11,8 +10,7 @@ import json
import logging
from typing import Dict, Optional, Set
# Sofortiger Print beim Laden des Moduls (erscheint garantiert in journalctl)
print(">>> MODULE_LOAD: edge_registry.py is being read by Python <<<", flush=True)
print(">>> MODULE_LOAD: edge_registry.py initialized <<<", flush=True)
from app.config import get_settings
@ -23,7 +21,6 @@ class EdgeRegistry:
def __new__(cls, *args, **kwargs):
if cls._instance is None:
print(">>> SINGLETON: Creating new EdgeRegistry instance <<<", flush=True)
cls._instance = super(EdgeRegistry, cls).__new__(cls)
cls._instance.initialized = False
return cls._instance
@ -32,10 +29,7 @@ class EdgeRegistry:
if self.initialized:
return
print(">>> INIT: EdgeRegistry.__init__ started <<<", flush=True)
settings = get_settings()
# Prio: MINDNET_VOCAB_PATH > MINDNET_VAULT_ROOT > Default
env_vocab_path = os.getenv("MINDNET_VOCAB_PATH")
env_vault_root = os.getenv("MINDNET_VAULT_ROOT") or getattr(settings, "MINDNET_VAULT_ROOT", "./vault_master")
@ -55,16 +49,16 @@ class EdgeRegistry:
def _load_vocabulary(self):
"""Parst die Markdown-Tabelle im Vault."""
print(f">>> CHECK: Searching for Vocabulary at {self.full_vocab_path}", flush=True)
print(f">>> CHECK: Loading Vocabulary from {self.full_vocab_path}", flush=True)
if not os.path.exists(self.full_vocab_path):
msg = f"!!! [DICT-ERROR] Edge Vocabulary NOT found at: {self.full_vocab_path} !!!"
print(msg, flush=True)
logger.warning(msg)
print(f"!!! [DICT-ERROR] File not found: {self.full_vocab_path} !!!", flush=True)
return
# Regex für Markdown Tabellen
pattern = re.compile(r"\|\s*\*\*([a-z_]+)\*\*\s*\|\s*([^|]+)\|")
# WP-22 Precision Regex:
# Sucht nach | **`typ`** | oder | **typ** |
# Die Backticks `? sind jetzt optional enthalten.
pattern = re.compile(r"\|\s*\*\*`?([a-zA-Z0-9_-]+)`?\*\*\s*\|\s*([^|]+)\|")
try:
with open(self.full_vocab_path, "r", encoding="utf-8") as f:
@ -72,7 +66,7 @@ class EdgeRegistry:
for line in f:
match = pattern.search(line)
if match:
canonical = match.group(1).strip()
canonical = match.group(1).strip().lower()
aliases_str = match.group(2).strip()
self.valid_types.add(canonical)
@ -80,26 +74,29 @@ 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:
clean_alias = alias.replace("`", "").lower().strip()
clean_alias = alias.replace("`", "").lower().strip().replace(" ", "_")
self.canonical_map[clean_alias] = canonical
c_aliases += 1
success_msg = f"=== [DICT-SUCCESS] Loaded {c_types} Types and {c_aliases} Aliases ==="
print(success_msg, flush=True)
logger.info(success_msg)
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)
except Exception as e:
err_msg = f"!!! [DICT-FATAL] Failed to parse: {e} !!!"
print(err_msg, flush=True)
logger.error(err_msg)
print(f"!!! [DICT-FATAL] Error reading file: {e} !!!", flush=True)
def resolve(self, edge_type: str) -> str:
"""Normalisiert Kanten-Typen via Registry oder loggt Unbekannte."""
if not edge_type: return "related_to"
clean_type = edge_type.lower().strip().replace(" ", "_")
clean_type = edge_type.lower().strip().replace(" ", "_").replace("-", "_")
if clean_type in self.canonical_map:
return self.canonical_map[clean_type]
self._log_unknown(clean_type)
return clean_type
@ -111,6 +108,4 @@ class EdgeRegistry:
f.write(json.dumps(entry) + "\n")
except Exception: pass
# Singleton Instanz - Dies triggert beim Import die Initialisierung
print(">>> INSTANTIATING REGISTRY SINGLETON <<<", flush=True)
registry = EdgeRegistry()