mindnet/app/core/derive_edges.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Modul: app/core/derive_edges.py
Version: 1.9.0
Datum: 2025-11-07

Zweck
-----
Robuste Kantenbildung für mindnet (Notes/Chunks):
  • belongs_to (chunk → note)
  • next / prev (Chunk-Sequenz)
  • references (chunk-scope) aus Chunk.window/text
  • 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 Dict, List, Optional, Iterable

from app.core.parser import extract_wikilinks

# Type-Registry (optional)
try:
    from app.core.type_registry import get_edge_defaults  # type: ignore
except Exception:
    def get_edge_defaults(_note_type: str) -> List[str]:
        return []

def _get(d: dict, *keys, default=None):
    for k in keys:
        if k in d and d[k] is not None:
            return d[k]
    return default

def _chunk_text_for_refs(chunk: dict) -> str:
    return (
        _get(chunk, "window")
        or _get(chunk, "text")
        or _get(chunk, "content")
        or _get(chunk, "raw")
        or ""
    )

def _dedupe(seq: Iterable[str]) -> List[str]:
    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, "source_id": source_id, "target_id": target_id, "note_id": note_id}
    if extra:
        pl.update(extra)
    return pl

def build_edges_for_note(
    note_id: str,
    chunks: List[dict],
    note_level_references: Optional[List[str]] = None,
    include_note_scope_refs: bool = False,
) -> List[dict]:
    edges: List[dict] = []

    # 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}))

    # next/prev
    for i in range(len(chunks) - 1):
        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)
    refs_all: List[str] = []
    for ch in chunks:
        cid = _get(ch, "chunk_id", "id")
        if not cid:
            continue
        txt = _chunk_text_for_refs(ch)
        refs = extract_wikilinks(txt)
        for r in refs:
            edges.append(_edge("references", "chunk", cid, r, note_id, {"chunk_id": cid, "ref_text": r}))
        refs_all.extend(refs)

    # optional: note-scope references/backlinks
    if include_note_scope_refs:
        refs_note = refs_all[:]
        if note_level_references:
            refs_note.extend([r for r in note_level_references if isinstance(r, str) and r])
        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)
    # Beispiel: task → depends_on, concept → related_to etc.
    # Wir erzeugen nur "formale" Kanten note→note_id selbst nicht; diese dienen Retri­ever-Gewichtung später.
    # (Hier keine Targets – das sind Typ-Hinweise und werden als self-hints abgelegt.)
    try:
        note_type = ""
        if chunks:
            note_type = str(_get(chunks[0], "type", default="") or "")
        defaults = get_edge_defaults(note_type)
        for k in defaults:
            # Self-hint-Kante (kann vom Retriever gewichtet werden). Scope 'note'
            edges.append(_edge(k, "note", note_id, note_id, note_id))
    except Exception:
        pass

    return edges