109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
"""
|
|
FILE: app/services/edge_registry.py
|
|
DESCRIPTION: Single Source of Truth für Kanten-Typen. Parst '01_edge_vocabulary.md'.
|
|
Implementiert WP-22 Teil B (Registry & Validation).
|
|
"""
|
|
import re
|
|
import os
|
|
import json
|
|
import logging
|
|
from typing import Dict, Optional, Set
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class EdgeRegistry:
|
|
_instance = None
|
|
|
|
def __new__(cls):
|
|
if cls._instance is None:
|
|
cls._instance = super(EdgeRegistry, cls).__new__(cls)
|
|
cls._instance.initialized = False
|
|
return cls._instance
|
|
|
|
def __init__(self):
|
|
if self.initialized: return
|
|
# Pfad korrespondiert mit dem Frontmatter Pfad in 01_edge_vocabulary.md
|
|
self.vocab_path = "01_User_Manual/01_edge_vocabulary.md"
|
|
self.unknown_log_path = "data/logs/unknown_edges.jsonl"
|
|
self.canonical_map: Dict[str, str] = {} # alias -> canonical
|
|
self.valid_types: Set[str] = set()
|
|
self._load_vocabulary()
|
|
self.initialized = True
|
|
|
|
def _load_vocabulary(self):
|
|
"""Parst die Markdown-Tabelle in 01_edge_vocabulary.md"""
|
|
# Fallback Suche, falls das Skript aus Root oder app ausgeführt wird
|
|
candidates = [
|
|
self.vocab_path,
|
|
os.path.join("..", self.vocab_path),
|
|
"vault/01_User_Manual/01_edge_vocabulary.md"
|
|
]
|
|
|
|
found_path = None
|
|
for p in candidates:
|
|
if os.path.exists(p):
|
|
found_path = p
|
|
break
|
|
|
|
if not found_path:
|
|
logger.warning(f"Edge Vocabulary not found (checked: {candidates}). Registry empty.")
|
|
return
|
|
|
|
# Regex für Tabellenzeilen: | **canonical** | alias, alias | ...
|
|
# Matcht: | **caused_by** | ausgelöst_durch, wegen |
|
|
pattern = re.compile(r"\|\s*\*\*([a-z_]+)\*\*\s*\|\s*([^|]+)\|")
|
|
|
|
try:
|
|
with open(found_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
match = pattern.search(line)
|
|
if match:
|
|
canonical = match.group(1).strip()
|
|
aliases_str = match.group(2).strip()
|
|
|
|
self.valid_types.add(canonical)
|
|
self.canonical_map[canonical] = canonical # Self-ref
|
|
|
|
# Aliases parsen
|
|
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 up user inputs (e.g. remove backticks if present)
|
|
clean_alias = alias.replace("`", "")
|
|
self.canonical_map[clean_alias] = canonical
|
|
|
|
logger.info(f"EdgeRegistry loaded: {len(self.valid_types)} canonical types from {found_path}.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to parse Edge Vocabulary: {e}")
|
|
|
|
def resolve(self, edge_type: str) -> str:
|
|
"""
|
|
Normalisiert Kanten-Typen. Loggt unbekannte Typen, verwirft sie aber nicht (Learning System).
|
|
"""
|
|
if not edge_type:
|
|
return "related_to"
|
|
|
|
clean_type = edge_type.lower().strip().replace(" ", "_")
|
|
|
|
# 1. Lookup
|
|
if clean_type in self.canonical_map:
|
|
return self.canonical_map[clean_type]
|
|
|
|
# 2. Unknown Handling
|
|
self._log_unknown(clean_type)
|
|
return clean_type # Pass-through (nicht verwerfen, aber loggen)
|
|
|
|
def _log_unknown(self, edge_type: str):
|
|
"""Schreibt unbekannte Typen in ein Append-Only Log für Review."""
|
|
try:
|
|
os.makedirs(os.path.dirname(self.unknown_log_path), exist_ok=True)
|
|
# Einfaches JSONL Format
|
|
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 # Silent fail bei Logging, darf Ingestion nicht stoppen
|
|
|
|
# Singleton Accessor
|
|
registry = EdgeRegistry() |