# app/core/derive_edges.py from __future__ import annotations import re import unicodedata from typing import Dict, List, Tuple WIKILINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#([^\]|]+))?(?:\|([^\]]+))?\]\]") def _slug(s: str) -> str: s = s.strip() if s.endswith(".md"): s = s[:-3] s = unicodedata.normalize("NFKD", s) s = "".join(ch for ch in s if not unicodedata.combining(ch)) s = s.replace("\\", "/") s = s.split("/")[-1] # nur letzter Pfadteil s = s.lower() s = s.replace(" ", "-") s = re.sub(r"[^a-z0-9\-]+", "", s) s = re.sub(r"-{2,}", "-", s).strip("-") return s def build_note_index(notes_payloads: List[dict]) -> Tuple[Dict[str, dict], Dict[str, dict], Dict[str, dict]]: by_id: Dict[str, dict] = {} by_slug: Dict[str, dict] = {} by_file_slug: Dict[str, dict] = {} for n in notes_payloads: nid = n.get("note_id") or n.get("id") if not nid: continue by_id[nid] = n title = n.get("title", "") path = n.get("path", "") # z.B. "Ordner/Datei.md" file_slug = _slug(path.split("/")[-1]) if path else "" if title: by_slug[_slug(title)] = n if file_slug: by_file_slug[file_slug] = n return by_id, by_slug, by_file_slug def resolve_target(note_like: str, idx: Tuple[Dict[str,dict],Dict[str,dict],Dict[str,dict]]) -> Tuple[str|None, str]: """ Liefert (target_note_id | None, resolution_hint) """ by_id, by_slug, by_file_slug = idx key = note_like.strip() if key in by_id: return key, "by_id" s = _slug(key) if s in by_slug: return by_slug[s]["note_id"], "by_slug" if s in by_file_slug: return by_file_slug[s]["note_id"], "by_file_slug" return None, "unresolved" def derive_wikilink_edges(note_payload: dict, chunks_payloads: List[dict], note_index) -> List[dict]: """ Erzeugt Edges aus [[...]] im Note-Text und in Chunk-Text: - references: source_note -> target_note - backlink: target_note -> source_note - optional: references_at: source_chunk -> target_note (wo steht der Link?) """ edges: List[dict] = [] source_note_id = note_payload["note_id"] def _make_edge(kind: str, src: str, tgt: str, seq: int|None=None, extra: dict|None=None) -> dict: e = { "edge_id": None, # später vergeben (qdrant_points macht robusten Fallback + UUID) "kind": kind, "source_id": src, "target_id": tgt, } if seq is not None: e["seq"] = seq if extra: e.update(extra) return e # 1) Links im Note-Gesamttext (falls vorhanden) fulltext = note_payload.get("fulltext") or note_payload.get("body") or "" if fulltext: for m in WIKILINK_RE.finditer(fulltext): raw_target, heading, alias = m.groups() target_id, how = resolve_target(raw_target, note_index) extra = {"raw": raw_target, "alias": alias, "heading": heading, "resolution": how} if target_id: edges.append(_make_edge("references", source_note_id, target_id, extra=extra)) edges.append(_make_edge("backlink", target_id, source_note_id, extra=extra)) else: # Unresolved: referenziere Slug/Raw, markiere Status extra["status"] = "unresolved" extra["target_label"] = raw_target edges.append(_make_edge("references", source_note_id, raw_target, extra=extra)) # 2) Links in den einzelnen Chunks for i, ch in enumerate(chunks_payloads, start=1): txt = ch.get("text") or ch.get("content") or "" if not txt: continue for m in WIKILINK_RE.finditer(txt): raw_target, heading, alias = m.groups() target_id, how = resolve_target(raw_target, note_index) extra = {"raw": raw_target, "alias": alias, "heading": heading, "resolution": how} # optional: chunk→note Referenz (wo steht der Link?) if target_id: edges.append(_make_edge("references_at", ch["chunk_id"], target_id, seq=i, extra=extra)) # die Note→Note-Kanten sind oben schon generiert; doppelt vermeiden: else: extra["status"] = "unresolved" extra["target_label"] = raw_target edges.append(_make_edge("references_at", ch["chunk_id"], raw_target, seq=i, extra=extra)) return edges