From 4eb5e34ea7256f9929c4dca1ac2c91c2654d55b2 Mon Sep 17 00:00:00 2001 From: Lars Date: Sat, 8 Nov 2025 15:21:17 +0100 Subject: [PATCH] app/core/derive_edges.py aktualisiert --- app/core/derive_edges.py | 226 +++++++++++++++++++++------------------ 1 file changed, 119 insertions(+), 107 deletions(-) diff --git a/app/core/derive_edges.py b/app/core/derive_edges.py index 5fd29e0..24ee1c7 100644 --- a/app/core/derive_edges.py +++ b/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: - - Ableiten von Edges aus Note + Chunks. - - Rückwärtskompatibel zu bisherigen Funktionen, inkl. Parametern: - build_edges_for_note(parsed, note_id, chunks, note_scope_refs=False, ...) - - Neu: Type-Registry wird optional benutzt, um fehlende Standardkanten je 'type' - (edge_defaults) zu ergänzen, ohne vorhandene explizite Kanten zu überschreiben. +Änderung +-------- +- Integration der Type-Registry (optional): Ist im Typ die Default-Kante + "references" enthalten, werden Note-Scope-References/Backlinks **additiv** + aktiviert – auch wenn `include_note_scope_refs=False` übergeben wurde. + (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 -except Exception: - def extract_wikilinks(text: str) -> List[str]: - # Minimaler Fallback – im Produktivcode steht eure echte Implementierung. - import re - return re.findall(r"\[\[([^\]]+)\]\]", text or "") + from app.core.type_registry import load_type_registry, get_type_config + _REG = load_type_registry() # prozessweiter Cache +except Exception: # pragma: no cover + _REG = {"types": {"concept": {"edge_defaults": ["references"]}}} + def get_type_config(_t, _r): # type: ignore + 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 -# Edge-Strukturen zurück. Upsert passiert über eure bestehenden qdrant_points-Utilities. -def _edge( - edge_type: str, - src_scope: str, - src_id: str, - dst_scope: str, - dst_id: str, - note_id: str, - extra: Optional[Dict[str, Any]] = None -) -> Dict[str, Any]: - e = { - "type": edge_type, - "src_scope": src_scope, - "src_id": src_id, - "dst_scope": dst_scope, - "dst_id": dst_id, - "note_id": note_id, +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: + # bevorzugt 'window' → dann 'text' → 'content' → 'raw' + 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, # "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) - return e + pl.update(extra) + return pl def build_edges_for_note( - *, note_id: str, - note_type: Optional[str], - chunks: List[Dict[str, Any]], - frontmatter: Dict[str, Any], - body_text: str, - note_scope_refs: bool = False, -) -> List[Dict[str, Any]]: + chunks: List[dict], + note_level_references: Optional[List[str]] = None, + include_note_scope_refs: bool = False, +) -> List[dict]: """ - 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. + Erzeugt Kanten für eine Note. + + - belongs_to: für jeden Chunk (chunk -> note) + - next / prev: zwischen aufeinanderfolgenden Chunks + - references: pro Chunk aus window/text + - optional note-scope references/backlinks: dedupliziert über alle Chunk-Funde + note_level_references + + Type-Registry-Erweiterung (additiv): + - 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 - for i, ch in enumerate(chunks): - cid = ch.get("chunk_id") or f"{note_id}#{i+1}" - edges.append(_edge("belongs_to", "chunk", cid, "note", note_id, note_id)) - if i > 0: - 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)) + # Typ aus Chunk-Payloads ableiten (falls vorhanden) + note_type = None + for ch in chunks: + nt = ch.get("type") + if isinstance(nt, str) and nt.strip(): + note_type = nt.strip().lower() + break + 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 - refs: Set[str] = set(extract_wikilinks(body_text or "")) + want_note_scope_refs = bool(include_note_scope_refs) or ("references" in edge_defaults) - if note_scope_refs: - # Referenzen auf Note-Ebene - for ref in sorted(refs): - edges.append(_edge("references", "note", note_id, "note", ref, note_id)) - edges.append(_edge("backlink", "note", ref, "note", note_id, note_id)) - else: - # 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)) + # 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})) - # depends_on / assigned_to aus Frontmatter - fm_dep = frontmatter.get("depends_on") or [] - if isinstance(fm_dep, list): - for d in fm_dep: - edges.append(_edge("depends_on", "note", note_id, "note", str(d), note_id)) + # 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})) - fm_ass = frontmatter.get("assigned_to") or [] - if isinstance(fm_ass, list): - for a in fm_ass: - edges.append(_edge("assigned_to", "note", note_id, "person", str(a), note_id)) + # references (chunk-scope) – Links aus window bevorzugen (Overlap-fest) + 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) # <— 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) - if note_type: - defaults = get_edge_defaults_for_type(note_type) - # Aus defaults lassen sich ohne konkretes Ziel nur symbolische Kanten ableiten. - # Wir fügen hier KEINE künstlichen Zielnoten hinzu; Registry ist vor allem - # für spätere WP-04-Gewichtungen gedacht. Wenn ihr konkrete Regeln definieren - # wollt (z. B. „task → belongs_to project aus frontmatter.project_id“), - # kann dies hier ergänzt werden (mit echten Ziel-IDs). - # Aktuell also: keine „leeren“ Kanten ohne Ziel hinzufügen. + # optional: note-scope references/backlinks + if want_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)) return edges