app/core/derive_edges.py aktualisiert

app/core/derive_edges.py

@@ -1,130 +1,142 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
 """
-derive_edges.py v1.9.0
+Modul: app/core/derive_edges.py
+Version: 1.5.0
+Datum: 2025-11-08
 
-Zweck:
+Änderung
-  - Ableiten von Edges aus Note + Chunks.
+--------
-  - Rückwärtskompatibel zu bisherigen Funktionen, inkl. Parametern:
+- Integration der Type-Registry (optional): Ist im Typ die Default-Kante
-      build_edges_for_note(parsed, note_id, chunks, note_scope_refs=False, ...)
+  "references" enthalten, werden Note-Scope-References/Backlinks **additiv**
-  - Neu: Type-Registry wird optional benutzt, um fehlende Standardkanten je 'type'
+  aktiviert – auch wenn `include_note_scope_refs=False` übergeben wurde.
-    (edge_defaults) zu ergänzen, ohne vorhandene explizite Kanten zu überschreiben.
+  (Keine Breaking Changes: bestehende Parameter bleiben erhalten.)
+
+Weitere Logik (belongs_to/prev/next & chunk-scope references) bleibt unverändert.
 """
-
 from __future__ import annotations
-from typing import Any, Dict, List, Optional, Set, Tuple
 
-# Annahme: parser liefert wikilinks und frontmatter; unverändert zum Stand vor der Erweiterung
+from typing import Dict, List, Optional, Iterable
+
+# WICHTIG: benutze die Parser-Extraktion für saubere Wikilinks
+from app.core.parser import extract_wikilinks
+
+# optional: Type-Registry (Fallback: deaktiviert)
 try:
-    from app.core.parser import extract_wikilinks  # type: ignore
+    from app.core.type_registry import load_type_registry, get_type_config
-except Exception:
+    _REG = load_type_registry()  # prozessweiter Cache
-    def extract_wikilinks(text: str) -> List[str]:
+except Exception:  # pragma: no cover
-        # Minimaler Fallback – im Produktivcode steht eure echte Implementierung.
+    _REG = {"types": {"concept": {"edge_defaults": ["references"]}}}
-        import re
+    def get_type_config(_t, _r):  # type: ignore
-        return re.findall(r"\[\[([^\]]+)\]\]", text or "")
+        return {"edge_defaults": ["references"]}
 
-# 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 []
 
-# Edge-Helper (UUIDv5, etc.) werden hier nicht neu definiert; wir liefern nur die
+def _get(d: dict, *keys, default=None):
-# Edge-Strukturen zurück. Upsert passiert über eure bestehenden qdrant_points-Utilities.
+    for k in keys:
-def _edge(
+        if k in d and d[k] is not None:
-    edge_type: str,
+            return d[k]
-    src_scope: str,
+    return default
-    src_id: str,
+
-    dst_scope: str,
+def _chunk_text_for_refs(chunk: dict) -> str:
-    dst_id: str,
+    # bevorzugt 'window' → dann 'text' → 'content' → 'raw'
-    note_id: str,
+    return (
-    extra: Optional[Dict[str, Any]] = None
+        _get(chunk, "window")
-) -> Dict[str, Any]:
+        or _get(chunk, "text")
-    e = {
+        or _get(chunk, "content")
-        "type": edge_type,
+        or _get(chunk, "raw")
-        "src_scope": src_scope,
+        or ""
-        "src_id": src_id,
+    )
-        "dst_scope": dst_scope,
+
-        "dst_id": dst_id,
+def _dedupe(seq: Iterable[str]) -> List[str]:
-        "note_id": note_id,
+    seen = set()
+    out: List[str] = []
+    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,         # "chunk" | "note"
+        "source_id": source_id,
+        "target_id": target_id,
+        "note_id": note_id,     # Träger/Quelle der Kante (aktuelle Note)
     }
     if extra:
-        e.update(extra)
+        pl.update(extra)
-    return e
+    return pl
 
 def build_edges_for_note(
-    *,
     note_id: str,
-    note_type: Optional[str],
+    chunks: List[dict],
-    chunks: List[Dict[str, Any]],
+    note_level_references: Optional[List[str]] = None,
-    frontmatter: Dict[str, Any],
+    include_note_scope_refs: bool = False,
-    body_text: str,
+) -> List[dict]:
-    note_scope_refs: bool = False,
-) -> List[Dict[str, Any]]:
     """
-    Erzeugt die Standardkanten:
+    Erzeugt Kanten für eine Note.
-      - 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:
+    - belongs_to:   für jeden Chunk   (chunk -> note)
-      - Type-Registry edge_defaults ergänzt fehlende Kanten als 'type_default'
+    - next / prev:  zwischen aufeinanderfolgenden Chunks
-        (z. B. 'related_to' ohne Ziel bleibt ungenutzt – nur, wenn interpretierbar).
+    - references:   pro Chunk aus window/text
+    - optional note-scope references/backlinks: dedupliziert über alle Chunk-Funde + note_level_references
 
-    Rückgabewert:
+    Type-Registry-Erweiterung (additiv):
-      Liste von Edge-Dicts. Upsert übernimmt euer Qdrant-Utility.
+    - Wenn der *Note-Typ* 'references' in seinen edge_defaults hat, werden
+      note-scope references/backlinks zusätzlich aktiviert.
     """
-    edges: List[Dict[str, Any]] = []
+    edges: List[dict] = []
 
-    # belongs_to + prev/next
+    # Typ aus Chunk-Payloads ableiten (falls vorhanden)
-    for i, ch in enumerate(chunks):
+    note_type = None
-        cid = ch.get("chunk_id") or f"{note_id}#{i+1}"
+    for ch in chunks:
-        edges.append(_edge("belongs_to", "chunk", cid, "note", note_id, note_id))
+        nt = ch.get("type")
-        if i > 0:
+        if isinstance(nt, str) and nt.strip():
-            prev_cid = chunks[i - 1].get("chunk_id") or f"{note_id}#{i}"
+            note_type = nt.strip().lower()
-            edges.append(_edge("prev", "chunk", cid, "chunk", prev_cid, note_id))
+            break
-            edges.append(_edge("next", "chunk", prev_cid, "chunk", cid, note_id))
+    type_cfg = get_type_config(note_type, _REG)
+    edge_defaults = [e for e in (type_cfg.get("edge_defaults") or []) if isinstance(e, str)]
 
-    # references/backlink
+    want_note_scope_refs = bool(include_note_scope_refs) or ("references" in edge_defaults)
-    refs: Set[str] = set(extract_wikilinks(body_text or ""))
 
-    if note_scope_refs:
+    # belongs_to
-        # 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:
+        edges.append(_edge("belongs_to", "chunk", cid, note_id, note_id, {"chunk_id": cid}))
-        # Referenzen pro Chunk (wenn sinnvoll)
-        for ch in chunks:
-            cid = ch.get("chunk_id")
-            ctext = (ch.get("text") or "") + "\n" + (ch.get("window") or "")
-            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))
 
-    # depends_on / assigned_to aus Frontmatter
+    # next/prev
-    fm_dep = frontmatter.get("depends_on") or []
+    for i in range(len(chunks) - 1):
-    if isinstance(fm_dep, list):
+        a, b = chunks[i], chunks[i + 1]
-        for d in fm_dep:
+        a_id = _get(a, "chunk_id", "id")
-            edges.append(_edge("depends_on", "note", note_id, "note", str(d), note_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}))
 
-    fm_ass = frontmatter.get("assigned_to") or []
+    # references (chunk-scope) – Links aus window bevorzugen (Overlap-fest)
-    if isinstance(fm_ass, list):
+    refs_all: List[str] = []
-        for a in fm_ass:
+    for ch in chunks:
-            edges.append(_edge("assigned_to", "note", note_id, "person", str(a), note_id))
+        cid = _get(ch, "chunk_id", "id")
+        if not cid:
+            continue
+        txt = _chunk_text_for_refs(ch)
+        refs = extract_wikilinks(txt)  # <— Parser-Logik, kompatibel zu deinem System
+        for r in refs:
+            edges.append(_edge("references", "chunk", cid, r, note_id, {"chunk_id": cid, "ref_text": r}))
+        refs_all.extend(refs)
 
-    # Type-Defaults (nur ergänzend, wenn interpretierbar)
+    # optional: note-scope references/backlinks
-    if note_type:
+    if want_note_scope_refs:
-        defaults = get_edge_defaults_for_type(note_type)
+        refs_note = refs_all[:]
-        # Aus defaults lassen sich ohne konkretes Ziel nur symbolische Kanten ableiten.
+        if note_level_references:
-        # Wir fügen hier KEINE künstlichen Zielnoten hinzu; Registry ist vor allem
+            refs_note.extend([r for r in note_level_references if isinstance(r, str) and r])
-        # für spätere WP-04-Gewichtungen gedacht. Wenn ihr konkrete Regeln definieren
+        refs_note = _dedupe(refs_note)
-        # wollt (z. B. „task → belongs_to project aus frontmatter.project_id“),
+        for r in refs_note:
-        # kann dies hier ergänzt werden (mit echten Ziel-IDs).
+            edges.append(_edge("references", "note", note_id, r, note_id))
-        # Aktuell also: keine „leeren“ Kanten ohne Ziel hinzufügen.
+            edges.append(_edge("backlink", "note", r, note_id, note_id))
 
     return edges