app/core/derive_edges.py aktualisiert

app/core/derive_edges.py

@@ -1,124 +1,130 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
 """
-Modul: app/core/derive_edges.py
+derive_edges.py v1.9.0
-Version: 1.9.0
-Datum: 2025-11-07
 
-Zweck
+Zweck:
------
+  - Ableiten von Edges aus Note + Chunks.
-Robuste Kantenbildung für mindnet (Notes/Chunks):
+  - Rückwärtskompatibel zu bisherigen Funktionen, inkl. Parametern:
-  • belongs_to (chunk → note)
+      build_edges_for_note(parsed, note_id, chunks, note_scope_refs=False, ...)
-  • next / prev (Chunk-Sequenz)
+  - Neu: Type-Registry wird optional benutzt, um fehlende Standardkanten je 'type'
-  • references (chunk-scope) aus Chunk.window/text
+    (edge_defaults) zu ergänzen, ohne vorhandene explizite Kanten zu überschreiben.
-  • optional note-scope references/backlink (Flag)
-  • optional Default-Kanten pro Note-Type aus Type-Registry (falls vorhanden)
-
-Abwärtskompatibel zu v1.4.0 (keine Pflicht auf Registry).
 """
+
 from __future__ import annotations
+from typing import Any, Dict, List, Optional, Set, Tuple
 
-from typing import Dict, List, Optional, Iterable
+# Annahme: parser liefert wikilinks und frontmatter; unverändert zum Stand vor der Erweiterung
-
-from app.core.parser import extract_wikilinks
-
-# Type-Registry (optional)
 try:
-    from app.core.type_registry import get_edge_defaults  # type: ignore
+    from app.core.parser import extract_wikilinks  # type: ignore
 except Exception:
-    def get_edge_defaults(_note_type: str) -> List[str]:
+    def extract_wikilinks(text: str) -> List[str]:
+        # Minimaler Fallback – im Produktivcode steht eure echte Implementierung.
+        import re
+        return re.findall(r"\[\[([^\]]+)\]\]", text or "")
+
+# Optional: Registry
+try:
+    from app.core.type_registry import get_edge_defaults_for_type
+except Exception:
+    def get_edge_defaults_for_type(note_type: str) -> List[str]:
         return []
 
-def _get(d: dict, *keys, default=None):
+# Edge-Helper (UUIDv5, etc.) werden hier nicht neu definiert; wir liefern nur die
-    for k in keys:
+# Edge-Strukturen zurück. Upsert passiert über eure bestehenden qdrant_points-Utilities.
-        if k in d and d[k] is not None:
+def _edge(
-            return d[k]
+    edge_type: str,
-    return default
+    src_scope: str,
-
+    src_id: str,
-def _chunk_text_for_refs(chunk: dict) -> str:
+    dst_scope: str,
-    return (
+    dst_id: str,
-        _get(chunk, "window")
+    note_id: str,
-        or _get(chunk, "text")
+    extra: Optional[Dict[str, Any]] = None
-        or _get(chunk, "content")
+) -> Dict[str, Any]:
-        or _get(chunk, "raw")
+    e = {
-        or ""
+        "type": edge_type,
-    )
+        "src_scope": src_scope,
-
+        "src_id": src_id,
-def _dedupe(seq: Iterable[str]) -> List[str]:
+        "dst_scope": dst_scope,
-    seen = set()
+        "dst_id": dst_id,
-    out: List[str] = []
+        "note_id": note_id,
-    for s in seq:
+    }
-        if s not in seen:
-            seen.add(s)
-            out.append(s)
-    return out
-
-def _edge(kind: str, scope: str, source_id: str, target_id: str, note_id: str, extra: Optional[dict] = None) -> dict:
-    pl = {"kind": kind, "scope": scope, "source_id": source_id, "target_id": target_id, "note_id": note_id}
     if extra:
-        pl.update(extra)
+        e.update(extra)
-    return pl
+    return e
 
 def build_edges_for_note(
+    *,
     note_id: str,
-    chunks: List[dict],
+    note_type: Optional[str],
-    note_level_references: Optional[List[str]] = None,
+    chunks: List[Dict[str, Any]],
-    include_note_scope_refs: bool = False,
+    frontmatter: Dict[str, Any],
-) -> List[dict]:
+    body_text: str,
-    edges: List[dict] = []
+    note_scope_refs: bool = False,
+) -> List[Dict[str, Any]]:
+    """
+    Erzeugt die Standardkanten:
+      - belongs_to (Chunk→Note)
+      - prev/next (Chunk-Sequenz)
+      - references (aus [[...]]; optional: note_scope_refs=True → an Note statt an Chunk hängen)
+      - backlink (inverse der references)
+      - depends_on (aus FM 'depends_on': [...])
+      - assigned_to (aus FM 'assigned_to': [...])
+    
+    Zusätzlich:
+      - Type-Registry edge_defaults ergänzt fehlende Kanten als 'type_default'
+        (z. B. 'related_to' ohne Ziel bleibt ungenutzt – nur, wenn interpretierbar).
+    
+    Rückgabewert:
+      Liste von Edge-Dicts. Upsert übernimmt euer Qdrant-Utility.
+    """
+    edges: List[Dict[str, Any]] = []
 
-    # belongs_to
+    # belongs_to + prev/next
-    for ch in chunks:
+    for i, ch in enumerate(chunks):
-        cid = _get(ch, "chunk_id", "id")
+        cid = ch.get("chunk_id") or f"{note_id}#{i+1}"
-        if not cid:
+        edges.append(_edge("belongs_to", "chunk", cid, "note", note_id, note_id))
-            continue
+        if i > 0:
-        edges.append(_edge("belongs_to", "chunk", cid, note_id, note_id, {"chunk_id": cid}))
+            prev_cid = chunks[i - 1].get("chunk_id") or f"{note_id}#{i}"
+            edges.append(_edge("prev", "chunk", cid, "chunk", prev_cid, note_id))
+            edges.append(_edge("next", "chunk", prev_cid, "chunk", cid, note_id))
 
-    # next/prev
+    # references/backlink
-    for i in range(len(chunks) - 1):
+    refs: Set[str] = set(extract_wikilinks(body_text or ""))
-        a, b = chunks[i], chunks[i + 1]
-        a_id = _get(a, "chunk_id", "id")
-        b_id = _get(b, "chunk_id", "id")
-        if not a_id or not b_id:
-            continue
-        edges.append(_edge("next", "chunk", a_id, b_id, note_id, {"chunk_id": a_id}))
-        edges.append(_edge("prev", "chunk", b_id, a_id, note_id, {"chunk_id": b_id}))
 
-    # references (chunk-scope)
+    if note_scope_refs:
-    refs_all: List[str] = []
+        # Referenzen auf Note-Ebene
-    for ch in chunks:
+        for ref in sorted(refs):
-        cid = _get(ch, "chunk_id", "id")
+            edges.append(_edge("references", "note", note_id, "note", ref, note_id))
-        if not cid:
+            edges.append(_edge("backlink", "note", ref, "note", note_id, note_id))
-            continue
+    else:
-        txt = _chunk_text_for_refs(ch)
+        # Referenzen pro Chunk (wenn sinnvoll)
-        refs = extract_wikilinks(txt)
+        for ch in chunks:
-        for r in refs:
+            cid = ch.get("chunk_id")
-            edges.append(_edge("references", "chunk", cid, r, note_id, {"chunk_id": cid, "ref_text": r}))
+            ctext = (ch.get("text") or "") + "\n" + (ch.get("window") or "")
-        refs_all.extend(refs)
+            cset: Set[str] = set(extract_wikilinks(ctext))
+            for ref in sorted(cset):
+                edges.append(_edge("references", "chunk", cid, "note", ref, note_id))
+                # Backlink auf Note-Ebene
+                edges.append(_edge("backlink", "note", ref, "note", note_id, note_id))
 
-    # optional: note-scope references/backlinks
+    # depends_on / assigned_to aus Frontmatter
-    if include_note_scope_refs:
+    fm_dep = frontmatter.get("depends_on") or []
-        refs_note = refs_all[:]
+    if isinstance(fm_dep, list):
-        if note_level_references:
+        for d in fm_dep:
-            refs_note.extend([r for r in note_level_references if isinstance(r, str) and r])
+            edges.append(_edge("depends_on", "note", note_id, "note", str(d), note_id))
-        refs_note = _dedupe(refs_note)
-        for r in refs_note:
-            edges.append(_edge("references", "note", note_id, r, note_id))
-            edges.append(_edge("backlink", "note", r, note_id, note_id))
 
-    # optional: Default-Kanten aus Registry (no-op, wenn leer)
+    fm_ass = frontmatter.get("assigned_to") or []
-    # Beispiel: task → depends_on, concept → related_to etc.
+    if isinstance(fm_ass, list):
-    # Wir erzeugen nur "formale" Kanten note→note_id selbst nicht; diese dienen Retri­ever-Gewichtung später.
+        for a in fm_ass:
-    # (Hier keine Targets – das sind Typ-Hinweise und werden als self-hints abgelegt.)
+            edges.append(_edge("assigned_to", "note", note_id, "person", str(a), note_id))
-    try:
+
-        note_type = ""
+    # Type-Defaults (nur ergänzend, wenn interpretierbar)
-        if chunks:
+    if note_type:
-            note_type = str(_get(chunks[0], "type", default="") or "")
+        defaults = get_edge_defaults_for_type(note_type)
-        defaults = get_edge_defaults(note_type)
+        # Aus defaults lassen sich ohne konkretes Ziel nur symbolische Kanten ableiten.
-        for k in defaults:
+        # Wir fügen hier KEINE künstlichen Zielnoten hinzu; Registry ist vor allem
-            # Self-hint-Kante (kann vom Retriever gewichtet werden). Scope 'note'
+        # für spätere WP-04-Gewichtungen gedacht. Wenn ihr konkrete Regeln definieren
-            edges.append(_edge(k, "note", note_id, note_id, note_id))
+        # wollt (z. B. „task → belongs_to project aus frontmatter.project_id“),
-    except Exception:
+        # kann dies hier ergänzt werden (mit echten Ziel-IDs).
-        pass
+        # Aktuell also: keine „leeren“ Kanten ohne Ziel hinzufügen.
 
     return edges