From fd215c18e4d6d710f640f0ebbb0fd116d1aaf658 Mon Sep 17 00:00:00 2001 From: Lars Date: Mon, 17 Nov 2025 16:27:40 +0100 Subject: [PATCH] app/core/derive_edges.py aktualisiert --- app/core/derive_edges.py | 181 ++++++++++++++++++--------------------- 1 file changed, 83 insertions(+), 98 deletions(-) diff --git a/app/core/derive_edges.py b/app/core/derive_edges.py index 0465cb4..616f5d9 100644 --- a/app/core/derive_edges.py +++ b/app/core/derive_edges.py @@ -2,40 +2,38 @@ # -*- coding: utf-8 -*- """ Modul: app/core/derive_edges.py -Version: 2.2.0 (V2-superset mit "typed inline relations" + Obsidian-Callouts) +Zweck: +- Bewahrt bestehende Edgelogik (belongs_to, prev/next, references, backlink) +- Ergänzt typenbasierte Default-Kanten (edge_defaults aus config/types.yaml) +- Unterstützt "typed inline relations" ([[rel:KIND | Target]] / [[rel:KIND Target]]) +- Unterstützt Obsidian-Callouts (> [!edge] KIND: [[Target]] [[Target2]] ...) -Zweck ------ -Bewahrt die bestehende Edgelogik (belongs_to, prev/next, references, backlink) -und ergänzt: -- Typ-Default-Kanten gemäß config/types.yaml (edge_defaults je Notiztyp) -- **Explizite, getypte Inline-Relationen** direkt im Chunk-Text: - * [[rel:depends_on | Target Title]] - * [[rel:related_to Target Title]] -- **Obsidian-Callouts** zur Pflege von Kanten im Markdown: - * > [!edge] related_to: [[Vector DB Basics]] - * Mehrere Zeilen im Callout werden unterstützt (alle Zeilen beginnen mit '>'). - -Konfiguration -------------- -- ENV MINDNET_TYPES_FILE (Default: ./config/types.yaml) - -Hinweis -------- -Diese Implementierung ist rückwärtskompatibel zur bisherigen Signatur. +Kompatibilität: +- build_edges_for_note(...) Signatur unverändert +- rule_id Werte exakt wie zuvor erwartet (ohne Versionssuffix): + * structure:belongs_to + * structure:order + * explicit:wikilink + * inline:rel + * callout:edge + * edge_defaults:: + * derived:backlink """ + from __future__ import annotations import os import re -from typing import Iterable, List, Optional, Tuple, Set +from typing import Iterable, List, Optional, Tuple, Set, Dict try: import yaml # optional, nur für types.yaml except Exception: # pragma: no cover yaml = None -# ---------------------------- Utilities ------------------------------------ +# --------------------------------------------------------------------------- # +# Utilities +# --------------------------------------------------------------------------- # def _get(d: dict, *keys, default=None): for k in keys: @@ -65,11 +63,11 @@ def _dedupe_seq(seq: Iterable[str]) -> List[str]: def _edge(kind: str, scope: str, source_id: str, target_id: str, note_id: str, extra: Optional[dict] = None) -> dict: pl = { "kind": kind, - "relation": kind, # v2 Feld (alias) + "relation": kind, # Alias (v2) "scope": scope, # "chunk" | "note" "source_id": source_id, "target_id": target_id, - "note_id": note_id, # Träger/Quelle der Kante (aktuelle Note) + "note_id": note_id, # Träger-Note der Kante } if extra: pl.update(extra) @@ -85,11 +83,13 @@ def _mk_edge_id(kind: str, s: str, t: str, scope: str, rule_id: Optional[str] = except Exception: # pragma: no cover return base -# ---------------------- Typen-Registry (types.yaml) ------------------------ +# --------------------------------------------------------------------------- # +# Typen-Registry (types.yaml) +# --------------------------------------------------------------------------- # def _env(n: str, default: Optional[str] = None) -> str: v = os.getenv(n) - return v if v is not None else (default or "" ) + return v if v is not None else (default or "") def _load_types_registry() -> dict: """Lädt die YAML-Registry aus MINDNET_TYPES_FILE oder ./config/types.yaml""" @@ -117,20 +117,19 @@ def _edge_defaults_for(note_type: Optional[str], reg: dict) -> List[str]: 3) [] """ types_map = _get_types_map(reg) - # 1) exakter Typ 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)] - # 2) Fallback 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)] - # 3) leer return [] -# ------------------------ Parser für Links --------------------------------- +# --------------------------------------------------------------------------- # +# Parser für Links / Relationen +# --------------------------------------------------------------------------- # # Normale Wikilinks (Fallback) _WIKILINK_RE = re.compile(r"\[\[(?:[^\|\]]+\|)?([a-zA-Z0-9_\-#:. ]+)\]\]") @@ -152,28 +151,21 @@ def _extract_typed_relations(text: str) -> Tuple[List[Tuple[str,str]], str]: t = (m.group("target") or "").strip() if k and t: pairs.append((k, t)) - return "" # löschen + return "" # Link entfernen text = _REL_PIPE.sub(_collect, text) text = _REL_SPACE.sub(_collect, text) return pairs, text -# ---- Obsidian Callout Parser ---------------------------------------------- -# Callout-Start erkennt Zeilen wie: > [!edge] ... (case-insensitive) +# Obsidian Callout Parser _CALLOUT_START = re.compile(r"^\s*>\s*\[!edge\]\s*(.*)$", re.IGNORECASE) - -# Innerhalb von Callouts erwarten wir je Zeile Muster wie: -# related_to: [[Vector DB Basics]] -# depends_on: [[A]], [[B]] -# similar_to: Qdrant Vektordatenbank -_REL_LINE = re.compile(r"^(?P[a-z_]+)\s*:\s*(?P.+?)\s*$", re.IGNORECASE) - +_REL_LINE = re.compile(r"^(?P[a-z_]+)\s*:\s*(?P.+?)\s*$", re.IGNORECASE) _WIKILINKS_IN_LINE = re.compile(r"\[\[([^\]]+)\]\]") def _extract_callout_relations(text: str) -> Tuple[List[Tuple[str,str]], str]: """ - Findet Obsidian-Callouts vom Typ [!edge] und extrahiert (kind, target). - Entfernt den gesamten Callout-Block aus dem Text, damit Wikilinks daraus - nicht zusätzlich als "references" gezählt werden. + Findet [!edge]-Callouts und extrahiert (kind, target). Entfernt den gesamten + Callout-Block aus dem Text (damit Wikilinks daraus nicht zusätzlich als + "references" gezählt werden). """ if not text: return [], text @@ -181,8 +173,8 @@ def _extract_callout_relations(text: str) -> Tuple[List[Tuple[str,str]], str]: lines = text.splitlines() out_pairs: List[Tuple[str,str]] = [] keep_lines: List[str] = [] - i = 0 + while i < len(lines): m = _CALLOUT_START.match(lines[i]) if not m: @@ -190,28 +182,22 @@ def _extract_callout_relations(text: str) -> Tuple[List[Tuple[str,str]], str]: i += 1 continue - # Wir sind in einem Callout-Block; erste Zeile nach dem Marker: - # Rest dieser Zeile nach [!edge] mitnehmen block_lines: List[str] = [] first_rest = m.group(1) or "" if first_rest.strip(): block_lines.append(first_rest) - # Folgezeilen sind Teil des Callouts, solange sie weiterhin mit '>' beginnen i += 1 while i < len(lines) and lines[i].lstrip().startswith('>'): - # Entferne führendes '>' und evtl. Leerzeichen block_lines.append(lines[i].lstrip()[1:].lstrip()) i += 1 - # Parse jede Blockzeile eigenständig for bl in block_lines: mrel = _REL_LINE.match(bl) if not mrel: continue kind = (mrel.group("kind") or "").strip().lower() targets = mrel.group("targets") or "" - # Wikilinks bevorzugt found = _WIKILINKS_IN_LINE.findall(targets) if found: for t in found: @@ -219,12 +205,12 @@ def _extract_callout_relations(text: str) -> Tuple[List[Tuple[str,str]], str]: if t: out_pairs.append((kind, t)) else: - # Fallback: Split per ',' oder ';' for raw in re.split(r"[,;]", targets): t = raw.strip() if t: out_pairs.append((kind, t)) - # Wichtig: Callout wird NICHT in keep_lines übernommen (entfernt) + + # Callout wird NICHT in keep_lines übernommen continue remainder = "\n".join(keep_lines) @@ -236,7 +222,9 @@ def _extract_wikilinks(text: str) -> List[str]: ids.append(m.group(1).strip()) return ids -# --------------------------- Hauptfunktion --------------------------------- +# --------------------------------------------------------------------------- # +# Hauptfunktion +# --------------------------------------------------------------------------- # def build_edges_for_note( note_id: str, @@ -251,31 +239,31 @@ def build_edges_for_note( - next / prev: zwischen aufeinanderfolgenden Chunks - references: pro Chunk aus window/text (via Wikilinks) - typed inline relations: [[rel:KIND | Target]] oder [[rel:KIND Target]] - - Obsidian Callouts: > [!edge] KIND: [[Target]] + - Obsidian Callouts: > [!edge] KIND: [[Target]] [[Target2]] - optional note-scope references/backlinks: dedupliziert über alle Chunk-Funde + note_level_references - typenbasierte Default-Kanten (edge_defaults) je gefundener Referenz """ edges: List[dict] = [] - # --- 0) Note-Typ ermitteln (aus erstem Chunk erwartet) --- + # Note-Typ (aus erstem Chunk erwartet) note_type = None if chunks: note_type = _get(chunks[0], "type") - # --- 1) belongs_to --- + # 1) belongs_to for ch in chunks: cid = _get(ch, "chunk_id", "id") if not cid: continue edges.append(_edge("belongs_to", "chunk", cid, note_id, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id("belongs_to", cid, note_id, "chunk", "structure:belongs_to:v1"), + "edge_id": _mk_edge_id("belongs_to", cid, note_id, "chunk", "structure:belongs_to"), "provenance": "rule", - "rule_id": "structure:belongs_to:v1", + "rule_id": "structure:belongs_to", "confidence": 1.0, })) - # --- 2) next/prev --- + # 2) next / prev for i in range(len(chunks) - 1): a, b = chunks[i], chunks[i + 1] a_id = _get(a, "chunk_id", "id") @@ -284,20 +272,20 @@ def build_edges_for_note( continue edges.append(_edge("next", "chunk", a_id, b_id, note_id, { "chunk_id": a_id, - "edge_id": _mk_edge_id("next", a_id, b_id, "chunk", "structure:order:v1"), + "edge_id": _mk_edge_id("next", a_id, b_id, "chunk", "structure:order"), "provenance": "rule", - "rule_id": "structure:order:v1", + "rule_id": "structure:order", "confidence": 0.95, })) edges.append(_edge("prev", "chunk", b_id, a_id, note_id, { "chunk_id": b_id, - "edge_id": _mk_edge_id("prev", b_id, a_id, "chunk", "structure:order:v1"), + "edge_id": _mk_edge_id("prev", b_id, a_id, "chunk", "structure:order"), "provenance": "rule", - "rule_id": "structure:order:v1", + "rule_id": "structure:order", "confidence": 0.95, })) - # --- 3) references (chunk-scope) + inline relations + callouts + abgeleitete Relationen je Ref --- + # 3) references + typed inline + callouts + defaults (chunk-scope) reg = _load_types_registry() defaults = _edge_defaults_for(note_type, reg) refs_all: List[str] = [] @@ -308,27 +296,29 @@ def build_edges_for_note( continue raw = _chunk_text_for_refs(ch) - # a) typed inline relations zuerst extrahieren + # 3a) typed inline relations typed, remainder = _extract_typed_relations(raw) for kind, target in typed: + kind = kind.strip().lower() + if not kind or not target: + continue edges.append(_edge(kind, "chunk", cid, target, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id(kind, cid, target, "chunk", "inline:rel:v1"), + "edge_id": _mk_edge_id(kind, cid, target, "chunk", "inline:rel"), "provenance": "explicit", - "rule_id": "inline:rel:v1", + "rule_id": "inline:rel", "confidence": 0.95, })) - # symmetrische Relationen zusätzlich rückwärts if kind in {"related_to", "similar_to"}: edges.append(_edge(kind, "chunk", target, cid, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id(kind, target, cid, "chunk", "inline:rel:v1"), + "edge_id": _mk_edge_id(kind, target, cid, "chunk", "inline:rel"), "provenance": "explicit", - "rule_id": "inline:rel:v1", + "rule_id": "inline:rel", "confidence": 0.95, })) - # b) Obsidian Callouts extrahieren (und aus remainder entfernen) + # 3b) callouts call_pairs, remainder2 = _extract_callout_relations(remainder) for kind, target in call_pairs: k = (kind or "").strip().lower() @@ -336,93 +326,88 @@ def build_edges_for_note( continue edges.append(_edge(k, "chunk", cid, target, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id(k, cid, target, "chunk", "callout:edge:v1"), + "edge_id": _mk_edge_id(k, cid, target, "chunk", "callout:edge"), "provenance": "explicit", - "rule_id": "callout:edge:v1", + "rule_id": "callout:edge", "confidence": 0.95, })) if k in {"related_to", "similar_to"}: edges.append(_edge(k, "chunk", target, cid, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id(k, target, cid, "chunk", "callout:edge:v1"), + "edge_id": _mk_edge_id(k, target, cid, "chunk", "callout:edge"), "provenance": "explicit", - "rule_id": "callout:edge:v1", + "rule_id": "callout:edge", "confidence": 0.95, })) - # c) generische Wikilinks (remainder2) → "references" + # 3c) generische Wikilinks → references (+ defaults je Ref) refs = _extract_wikilinks(remainder2) for r in refs: - # reale Referenz (wie bisher) edges.append(_edge("references", "chunk", cid, r, note_id, { "chunk_id": cid, "ref_text": r, - "edge_id": _mk_edge_id("references", cid, r, "chunk", "explicit:wikilink:v1"), + "edge_id": _mk_edge_id("references", cid, r, "chunk", "explicit:wikilink"), "provenance": "explicit", - "rule_id": "explicit:wikilink:v1", + "rule_id": "explicit:wikilink", "confidence": 1.0, })) - # abgeleitete Kanten je default-Relation for rel in defaults: if rel == "references": - continue # doppelt vermeiden + continue edges.append(_edge(rel, "chunk", cid, r, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id(rel, cid, r, "chunk", f"edge_defaults:{note_type}:{rel}:v1"), + "edge_id": _mk_edge_id(rel, cid, r, "chunk", f"edge_defaults:{note_type}:{rel}"), "provenance": "rule", - "rule_id": f"edge_defaults:{note_type}:{rel}:v1", + "rule_id": f"edge_defaults:{note_type}:{rel}", "confidence": 0.7, })) - # symmetrisch? if rel in {"related_to", "similar_to"}: edges.append(_edge(rel, "chunk", r, cid, note_id, { "chunk_id": cid, - "edge_id": _mk_edge_id(rel, r, cid, "chunk", f"edge_defaults:{note_type}:{rel}:v1"), + "edge_id": _mk_edge_id(rel, r, cid, "chunk", f"edge_defaults:{note_type}:{rel}"), "provenance": "rule", - "rule_id": f"edge_defaults:{note_type}:{rel}:v1", + "rule_id": f"edge_defaults:{note_type}:{rel}", "confidence": 0.7, })) refs_all.extend(refs) - # --- 4) optional: note-scope references/backlinks (+ defaults) --- + # 4) optional note-scope refs/backlinks (+ defaults) if include_note_scope_refs: refs_note = list(refs_all or []) if note_level_references: refs_note.extend([r for r in note_level_references if isinstance(r, str) and r]) refs_note = _dedupe_seq(refs_note) for r in refs_note: - # echte note-scope Referenz & Backlink (wie bisher) edges.append(_edge("references", "note", note_id, r, note_id, { - "edge_id": _mk_edge_id("references", note_id, r, "note", "explicit:note_scope:v1"), + "edge_id": _mk_edge_id("references", note_id, r, "note", "explicit:note_scope"), "provenance": "explicit", - "rule_id": "explicit:note_scope:v1", + "rule_id": "explicit:note_scope", "confidence": 1.0, })) edges.append(_edge("backlink", "note", r, note_id, note_id, { - "edge_id": _mk_edge_id("backlink", r, note_id, "note", "derived:backlink:v1"), + "edge_id": _mk_edge_id("backlink", r, note_id, "note", "derived:backlink"), "provenance": "rule", - "rule_id": "derived:backlink:v1", + "rule_id": "derived:backlink", "confidence": 0.9, })) - # und zusätzlich default-Relationen (note-scope) for rel in defaults: if rel == "references": continue edges.append(_edge(rel, "note", note_id, r, note_id, { - "edge_id": _mk_edge_id(rel, note_id, r, "note", f"edge_defaults:{note_type}:{rel}:v1"), + "edge_id": _mk_edge_id(rel, note_id, r, "note", f"edge_defaults:{note_type}:{rel}"), "provenance": "rule", - "rule_id": f"edge_defaults:{note_type}:{rel}:v1", + "rule_id": f"edge_defaults:{note_type}:{rel}", "confidence": 0.7, })) if rel in {"related_to", "similar_to"}: edges.append(_edge(rel, "note", r, note_id, note_id, { - "edge_id": _mk_edge_id(rel, r, note_id, "note", f"edge_defaults:{note_type}:{rel}:v1"), + "edge_id": _mk_edge_id(rel, r, note_id, "note", f"edge_defaults:{note_type}:{rel}"), "provenance": "rule", - "rule_id": f"edge_defaults:{note_type}:{rel}:v1", + "rule_id": f"edge_defaults:{note_type}:{rel}", "confidence": 0.7, })) - # --- 5) Dedupe (Schlüssel: source_id, target_id, relation, rule_id) --- + # 5) De-Dupe (source_id, target_id, relation, rule_id) seen: Set[Tuple[str,str,str,str]] = set() out: List[dict] = [] for e in edges: