diff --git a/app/core/edges.py b/app/core/edges.py index 9009ee7..4359cea 100644 --- a/app/core/edges.py +++ b/app/core/edges.py @@ -1,45 +1,120 @@ -from __future__ import annotations -from typing import List, Dict +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +Modul: app/core/edges.py +Version: 1.0.0 +Datum: 2025-09-09 -def deriv_edges_for_note(note_meta: Dict, chunk_payloads: List[Dict]) -> List[Dict]: +Zweck +----- +Zentrale, konsistente Erzeugung von Edge-Payloads im **neuen Schema**: + - kind : "belongs_to" | "next" | "prev" | "references" | "backlink" + - source_id : ID des Quellknotens (Chunk- oder Note-ID) + - target_id : ID des Zielknotens + - scope : "chunk" | "note" + - note_id : Owner-Note (für performantes Filtern/Löschen) + - seq : optional (z. B. Reihenfolge von Vorkommen) + +Hinweise +-------- +- Edges werden dedupliziert (key=(kind,source_id,target_id,scope)). +- Für Chunk-Edges wird `note_id` aus dem Chunk-Payload entnommen. +- Für Note-Scope-Edges ist `note_id` die Quell-Note-ID. +""" + +from __future__ import annotations +from typing import Dict, List + +def build_edges_for_note( + note_id: str, + chunk_payloads: List[Dict], + note_level_refs: List[str] | None, + *, + include_note_scope_refs: bool = False, +) -> List[Dict]: edges: List[Dict] = [] - # Chunk → Note (belongs_to) + prev/next - for idx, ch in enumerate(chunk_payloads): + # Chunk-Scope: belongs_to / prev / next / references + for ch in chunk_payloads: + cid = ch["id"] + owner = ch.get("note_id") or note_id + # belongs_to edges.append({ - "src_id": ch["id"], "dst_id": note_meta["id"], - "edge_type": "belongs_to", "scope": "chunk" + "kind": "belongs_to", + "source_id": cid, + "target_id": note_id, + "scope": "chunk", + "note_id": owner, }) - prev_id = ch.get("neighbors",{}).get("prev") - next_id = ch.get("neighbors",{}).get("next") + # Nachbarn + nb = ch.get("neighbors") or {} + prev_id = nb.get("prev") + next_id = nb.get("next") if prev_id: - edges.append({"src_id": ch["id"], "dst_id": prev_id, "edge_type": "next", "scope": "chunk"}) - edges.append({"src_id": prev_id, "dst_id": ch["id"], "edge_type": "prev", "scope": "chunk"}) + edges.append({ + "kind": "prev", + "source_id": cid, + "target_id": prev_id, + "scope": "chunk", + "note_id": owner, + }) + edges.append({ + "kind": "next", + "source_id": prev_id, + "target_id": cid, + "scope": "chunk", + "note_id": owner, + }) if next_id: - edges.append({"src_id": ch["id"], "dst_id": next_id, "edge_type": "next", "scope": "chunk"}) - edges.append({"src_id": next_id, "dst_id": ch["id"], "edge_type": "prev", "scope": "chunk"}) - - # references aus wikilinks (Chunk-Scope) - for ref in ch.get("references", []): + edges.append({ + "kind": "next", + "source_id": cid, + "target_id": next_id, + "scope": "chunk", + "note_id": owner, + }) + edges.append({ + "kind": "prev", + "source_id": next_id, + "target_id": cid, + "scope": "chunk", + "note_id": owner, + }) + # references aus Chunk + for ref in (ch.get("references") or []): tid = ref.get("target_id") - if not tid: continue - edges.append({"src_id": ch["id"], "dst_id": tid, "edge_type": "references", "scope": "chunk"}) - edges.append({"src_id": tid, "dst_id": ch["id"], "edge_type": "backlink", "scope": "chunk"}) + if not tid: + continue + edges.append({ + "kind": "references", + "source_id": cid, + "target_id": tid, + "scope": "chunk", + "note_id": owner, + }) - # depends_on / assigned_to (Note-Scope, aus Frontmatter) - for dep in note_meta.get("depends_on", []) or []: - edges.append({"src_id": note_meta["id"], "dst_id": dep, "edge_type": "depends_on", "scope": "note"}) - for ass in note_meta.get("assigned_to", []) or []: - edges.append({"src_id": note_meta["id"], "dst_id": ass, "edge_type": "assigned_to", "scope": "note"}) - - # Note-Level references (optional: falls du im Note-Payload `references` sammelst) - for tid in note_meta.get("references", []) or []: - edges.append({"src_id": note_meta["id"], "dst_id": tid, "edge_type": "references", "scope": "note"}) - edges.append({"src_id": tid, "dst_id": note_meta["id"], "edge_type": "backlink", "scope": "note"}) + # Note-Scope: backlink (immer); references (optional) + unique_refs = list(dict.fromkeys(note_level_refs or [])) + for tid in unique_refs: + if include_note_scope_refs: + edges.append({ + "kind": "references", + "source_id": note_id, + "target_id": tid, + "scope": "note", + "note_id": note_id, + }) + edges.append({ + "kind": "backlink", + "source_id": tid, + "target_id": note_id, + "scope": "note", + "note_id": note_id, + }) # Dedupe - uniq = {} + dedup = {} for e in edges: - k = (e["src_id"], e["edge_type"], e["dst_id"], e.get("scope","note")) - uniq[k] = e - return list(uniq.values()) \ No newline at end of file + k = (e["kind"], e["source_id"], e["target_id"], e.get("scope", "")) + dedup[k] = e + return list(dedup.values())