""" 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 Iterable, List, Optional, Set, Any, Tuple try: import yaml except ImportError: yaml = None # WP-15b: Prioritäten-Ranking für die De-Duplizierung von Kanten unterschiedlicher Herkunft PROVENANCE_PRIORITY = { "explicit:wikilink": 1.00, "inline:rel": 0.95, "callout:edge": 0.90, "explicit:callout": 0.90, # WP-24c v4.2.7: Callout-Kanten aus candidate_pool "semantic_ai": 0.90, # Validierte KI-Kanten "structure:belongs_to": 1.00, "structure:order": 0.95, # next/prev "explicit:note_scope": 1.00, "explicit:note_zone": 1.00, # WP-24c v4.2.0: Note-Scope Zonen (höchste Priorität) "derived:backlink": 0.90, "edge_defaults": 0.70 # Heuristik basierend auf types.yaml } # --------------------------------------------------------------------------- # 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. Wird von graph_derive_edges.py benötigt. """ 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 } if extra: pl.update(extra) 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 []