81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""
|
|
FILE: app/core/graph/graph_utils.py
|
|
DESCRIPTION: Basale Werkzeuge, ID-Generierung und Provenance-Konfiguration für den Graphen.
|
|
"""
|
|
import os
|
|
import hashlib
|
|
from typing import Iterable, List, Optional, Set, Any
|
|
|
|
try:
|
|
import yaml
|
|
except ImportError:
|
|
yaml = None
|
|
|
|
# WP-15b: Prioritäten-Ranking für die De-Duplizierung
|
|
PROVENANCE_PRIORITY = {
|
|
"explicit:wikilink": 1.00,
|
|
"inline:rel": 0.95,
|
|
"callout:edge": 0.90,
|
|
"semantic_ai": 0.90, # Validierte KI-Kanten
|
|
"structure:belongs_to": 1.00,
|
|
"structure:order": 0.95, # next/prev
|
|
"explicit:note_scope": 1.00,
|
|
"derived:backlink": 0.90,
|
|
"edge_defaults": 0.70 # Heuristik (types.yaml)
|
|
}
|
|
|
|
def _get(d: dict, *keys, default=None):
|
|
"""Sicherer Zugriff auf verschachtelte 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 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 _mk_edge_id(kind: str, s: str, t: str, scope: str, rule_id: Optional[str] = None) -> str:
|
|
"""Erzeugt eine deterministische 12-Byte ID mittels BLAKE2s."""
|
|
base = f"{kind}:{s}->{t}#{scope}"
|
|
if rule_id: base += f"|{rule_id}"
|
|
return hashlib.blake2s(base.encode("utf-8"), digest_size=12).hexdigest()
|
|
|
|
def _edge(kind: str, scope: str, source_id: str, target_id: str, note_id: str, extra: Optional[dict] = None) -> dict:
|
|
"""Konstruiert ein Kanten-Payload für Qdrant."""
|
|
pl = {
|
|
"kind": kind,
|
|
"relation": kind,
|
|
"scope": scope,
|
|
"source_id": source_id,
|
|
"target_id": target_id,
|
|
"note_id": note_id,
|
|
}
|
|
if extra: pl.update(extra)
|
|
return pl
|
|
|
|
def load_types_registry() -> dict:
|
|
"""Lädt die YAML-Registry."""
|
|
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: return yaml.safe_load(f) or {}
|
|
except Exception: return {}
|
|
|
|
def get_edge_defaults_for(note_type: Optional[str], reg: dict) -> List[str]:
|
|
"""Ermittelt Standard-Kanten für einen Typ."""
|
|
types_map = reg.get("types", reg) if isinstance(reg, dict) else {}
|
|
if note_type and isinstance(types_map, dict):
|
|
t = types_map.get(note_type)
|
|
if isinstance(t, dict) and isinstance(t.get("edge_defaults"), list):
|
|
return [str(x) for x in t["edge_defaults"] if isinstance(x, str)]
|
|
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 [] |