""" FILE: app/core/graph/graph_utils.py DESCRIPTION: Basale Werkzeuge, ID-Generierung und Provenance-Konfiguration für den Graphen. AUDIT v4.0.0: - GOLD-STANDARD v4.0.0: Strikte 4-Parameter-ID für Kanten (kind, source, target, scope). - Eliminiert ID-Inkonsistenz zwischen Phase 1 (Autorität) und Phase 2 (Symmetrie). - rule_id und variant werden ignoriert in der ID-Generierung (nur im Payload gespeichert). - Fix für das "Steinzeitaxt"-Problem durch konsistente ID-Generierung. VERSION: 4.0.0 (WP-24c: Gold-Standard Identity) STATUS: Active """ import os import uuid import hashlib from typing import Dict, Iterable, List, Optional, Set, Any, Tuple try: import yaml except ImportError: yaml = None # WP-26 v1.0: Provenance-Literale auf valide EdgeDTO-Werte reduziert # Legacy-Prioritäten für interne Verarbeitung (werden zu source_hint gemappt) PROVENANCE_PRIORITY = { # Explizite Kanten (provenance: "explicit") "explicit:wikilink": 1.00, "inline:rel": 0.95, "callout:edge": 0.90, "explicit:callout": 0.90, "explicit:note_scope": 1.00, "explicit:note_zone": 1.00, # Regel-basierte Kanten (provenance: "rule") "derived:backlink": 0.90, "edge_defaults": 0.70, "schema_default": 0.85, # Struktur-Kanten (provenance: "structure") "structure:belongs_to": 1.00, "structure:order": 0.95, # KI-generierte Kanten (provenance: "smart") "semantic_ai": 0.90, "global_pool": 0.80, } # WP-26 v1.0: Mapping von internen Provenance-Werten zu EdgeDTO-konformen Literalen PROVENANCE_TO_DTO = { # explicit "explicit:wikilink": ("explicit", "wikilink"), "explicit:callout": ("explicit", "callout"), "explicit:note_scope": ("explicit", "note_scope"), "explicit:note_zone": ("explicit", "note_zone"), "inline:rel": ("explicit", "inline_rel"), "callout:edge": ("explicit", "callout"), "explicit": ("explicit", None), # rule "derived:backlink": ("rule", "backlink"), "edge_defaults": ("rule", "edge_defaults"), "schema_default": ("rule", "schema_default"), "inferred:schema": ("rule", "schema_default"), "rule": ("rule", None), # structure "structure:belongs_to": ("structure", "belongs_to"), "structure:order": ("structure", "order"), "structure": ("structure", None), # smart "semantic_ai": ("smart", None), "global_pool": ("smart", "global_pool"), "smart": ("smart", None), } def normalize_provenance(internal_provenance: str) -> Tuple[str, Optional[str]]: """ WP-26 v1.0: Normalisiert interne Provenance-Werte zu EdgeDTO-konformen Literalen. Args: internal_provenance: Interner Provenance-String (z.B. "explicit:callout") Returns: Tuple (provenance, source_hint) mit validen EdgeDTO-Werten """ if internal_provenance in PROVENANCE_TO_DTO: return PROVENANCE_TO_DTO[internal_provenance] # Fallback: Versuche Präfix-Matching if internal_provenance.startswith("explicit"): return ("explicit", internal_provenance.split(":")[-1] if ":" in internal_provenance else None) if internal_provenance.startswith("structure"): return ("structure", internal_provenance.split(":")[-1] if ":" in internal_provenance else None) if internal_provenance.startswith("rule") or internal_provenance.startswith("derived"): return ("rule", internal_provenance.split(":")[-1] if ":" in internal_provenance else None) # Default: explicit ohne source_hint return ("explicit", None) # --------------------------------------------------------------------------- # Pfad-Auflösung (Integration der .env Umgebungsvariablen) # --------------------------------------------------------------------------- def get_vocab_path() -> str: """Liefert den Pfad zum Edge-Vokabular aus der .env oder den Default.""" return os.getenv("MINDNET_VOCAB_PATH", "/mindnet/vault/mindnet/_system/dictionary/edge_vocabulary.md") def get_schema_path() -> str: """Liefert den Pfad zum Graph-Schema aus der .env oder den Default.""" return os.getenv("MINDNET_SCHEMA_PATH", "/mindnet/vault/mindnet/_system/dictionary/graph_schema.md") # --------------------------------------------------------------------------- # ID & String Helper # --------------------------------------------------------------------------- def _get(d: dict, *keys, default=None): """Sicherer Zugriff auf tief verschachtelte Dictionary-Keys.""" for k in keys: if isinstance(d, dict) and k in d and d[k] is not None: return d[k] return default def _dedupe_seq(seq: Iterable[str]) -> List[str]: """Dedupliziert eine Sequenz von Strings unter Beibehaltung der Reihenfolge.""" seen: Set[str] = set() out: List[str] = [] for s in seq: if s not in seen: seen.add(s) out.append(s) return out def parse_link_target(raw: str, current_note_id: Optional[str] = None) -> Tuple[str, Optional[str]]: """ Trennt einen Obsidian-Link [[Target#Section]] in seine Bestandteile Target und Section. Behandelt Self-Links (z.B. [[#Ziele]]), indem die aktuelle note_id eingesetzt wird. Returns: Tuple (target_id, target_section) """ if not raw: return "", None parts = raw.split("#", 1) target = parts[0].strip() section = parts[1].strip() if len(parts) > 1 else None # Spezialfall: Self-Link innerhalb derselben Datei if not target and section and current_note_id: target = current_note_id return target, section def _mk_edge_id(kind: str, s: str, t: str, scope: str, target_section: Optional[str] = None) -> str: """ WP-24c v4.0.0: DER GLOBALE STANDARD für Kanten-IDs. Erzeugt eine deterministische UUIDv5. Dies stellt sicher, dass manuelle Links und systemgenerierte Symmetrien dieselbe Point-ID in Qdrant erhalten. GOLD-STANDARD v4.0.0: Die ID basiert STRICT auf vier Parametern: f"edge:{kind}:{source}:{target}:{scope}" Die Parameter rule_id und variant werden IGNORIERT und fließen NICHT in die ID ein. Sie können weiterhin im Payload gespeichert werden, haben aber keinen Einfluss auf die Identität. Args: kind: Typ der Relation (z.B. 'mastered_by') s: Kanonische ID der Quell-Note t: Kanonische ID der Ziel-Note scope: Granularität (Standard: 'note') rule_id: Optionale ID der Regel (aus graph_derive_edges) - IGNORIERT in ID-Generierung variant: Optionale Variante für multiple Links zum selben Ziel - IGNORIERT in ID-Generierung """ if not all([kind, s, t]): raise ValueError(f"Incomplete data for edge ID: kind={kind}, src={s}, tgt={t}") # Der String enthält nun alle distinkten semantischen Merkmale base = f"edge:{kind}:{s}:{t}:{scope}" # Wenn ein Link auf eine spezifische Sektion zeigt, ist es eine andere Relation if target_section: base += f":{target_section}" return str(uuid.uuid5(uuid.NAMESPACE_URL, base)) def _edge(kind: str, scope: str, source_id: str, target_id: str, note_id: str, extra: Optional[dict] = None) -> dict: """ Konstruiert ein standardisiertes Kanten-Payload für Qdrant. WP-26 v1.0: Erweitert um is_internal Flag und Provenance-Normalisierung. Args: kind: Kantentyp (z.B. "derives", "caused_by") scope: Granularität ("chunk" oder "note") source_id: ID der Quelle (Chunk oder Note) target_id: ID des Ziels (Chunk oder Note) note_id: ID der Note (für Kontext) extra: Zusätzliche Payload-Felder """ pl = { "kind": kind, "relation": kind, "scope": scope, "source_id": source_id, "target_id": target_id, "note_id": note_id, "virtual": False # Standardmäßig explizit, solange nicht anders in Phase 2 gesetzt } # WP-26 v1.0: is_internal Flag berechnen # Intra-Note-Edge: Source und Target gehören zur gleichen Note source_note = source_id.split("#")[0] if "#" in source_id else source_id target_note = target_id.split("#")[0] if "#" in target_id else target_id pl["is_internal"] = (source_note == target_note) or (source_note == note_id and target_note == note_id) if extra: pl.update(extra) # WP-26 v1.0: Provenance normalisieren, falls vorhanden if "provenance" in extra: internal_prov = extra["provenance"] dto_prov, source_hint = normalize_provenance(internal_prov) pl["provenance"] = dto_prov if source_hint: pl["source_hint"] = source_hint return pl # --------------------------------------------------------------------------- # Registry Operations # --------------------------------------------------------------------------- def load_types_registry() -> dict: """ Lädt die zentrale YAML-Registry (types.yaml). Pfad wird über die Umgebungsvariable MINDNET_TYPES_FILE gesteuert. """ p = os.getenv("MINDNET_TYPES_FILE", "./config/types.yaml") if not os.path.isfile(p) or yaml is None: return {} try: with open(p, "r", encoding="utf-8") as f: data = yaml.safe_load(f) return data if data is not None else {} except Exception: return {} def get_edge_defaults_for(note_type: Optional[str], reg: dict) -> List[str]: """ Ermittelt die konfigurierten Standard-Kanten für einen Note-Typ. Greift bei Bedarf auf die globalen Defaults in der Registry zurück. """ types_map = reg.get("types", reg) if isinstance(reg, dict) else {} if note_type and isinstance(types_map, dict): t_cfg = types_map.get(note_type) if isinstance(t_cfg, dict) and isinstance(t_cfg.get("edge_defaults"), list): return [str(x) for x in t_cfg["edge_defaults"]] # Fallback auf globale Defaults for key in ("defaults", "default", "global"): v = reg.get(key) if isinstance(v, dict) and isinstance(v.get("edge_defaults"), list): return [str(x) for x in v["edge_defaults"] if isinstance(x, str)] return []