""" FILE: app/core/graph/graph_derive_edges.py DESCRIPTION: Hauptlogik zur Kanten-Aggregation und De-Duplizierung. WP-15b/c Audit: - Präzises Sektions-Splitting via parse_link_target. - Eindeutige ID-Generierung pro Sektions-Variante (Multigraph). - Ermöglicht dem Retriever die Super-Edge-Aggregation. """ from typing import List, Optional, Dict, Tuple from .graph_utils import ( _get, _edge, _mk_edge_id, _dedupe_seq, parse_link_target, PROVENANCE_PRIORITY, load_types_registry, get_edge_defaults_for ) from .graph_extractors import ( extract_typed_relations, extract_callout_relations, extract_wikilinks ) 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]: """ Erzeugt und aggregiert alle Kanten für eine Note. Sorgt für die physische Trennung von Sektions-Links via Edge-ID. """ edges: List[dict] = [] # note_type für die Ermittlung der edge_defaults (types.yaml) note_type = _get(chunks[0], "type") if chunks else "concept" # 1) Struktur-Kanten (Internal: belongs_to, next/prev) # Diese erhalten die Provenienz 'structure' und sind in der Registry geschützt. for idx, ch in enumerate(chunks): cid = _get(ch, "chunk_id", "id") if not cid: continue # Verbindung Chunk -> Note # WP-24c v4.0.0: rule_id wird nur im Payload gespeichert, fließt nicht in die ID ein edges.append(_edge("belongs_to", "chunk", cid, note_id, note_id, { "chunk_id": cid, "edge_id": _mk_edge_id("belongs_to", cid, note_id, "chunk"), "provenance": "structure", "rule_id": "structure:belongs_to", "confidence": PROVENANCE_PRIORITY["structure:belongs_to"] })) # Horizontale Verkettung (Ordnung) if idx < len(chunks) - 1: next_id = _get(chunks[idx+1], "chunk_id", "id") if next_id: # WP-24c v4.0.0: rule_id wird nur im Payload gespeichert, fließt nicht in die ID ein edges.append(_edge("next", "chunk", cid, next_id, note_id, { "chunk_id": cid, "edge_id": _mk_edge_id("next", cid, next_id, "chunk"), "provenance": "structure", "rule_id": "structure:order", "confidence": PROVENANCE_PRIORITY["structure:order"] })) edges.append(_edge("prev", "chunk", next_id, cid, note_id, { "chunk_id": next_id, "edge_id": _mk_edge_id("prev", next_id, cid, "chunk"), "provenance": "structure", "rule_id": "structure:order", "confidence": PROVENANCE_PRIORITY["structure:order"] })) # 2) Inhaltliche Kanten (Explicit & Candidate Pool) reg = load_types_registry() defaults = get_edge_defaults_for(note_type, reg) refs_all: List[str] = [] for ch in chunks: cid = _get(ch, "chunk_id", "id") if not cid: continue raw = _get(ch, "window") or _get(ch, "text") or "" # A. Typed Relations (Inline [[rel:kind|target]]) typed, rem = extract_typed_relations(raw) for k, raw_t in typed: t, sec = parse_link_target(raw_t, note_id) if not t: continue payload = { "chunk_id": cid, # WP-24c v4.0.0: variant wird nur im Payload gespeichert (target_section), fließt nicht in die ID ein "edge_id": _mk_edge_id(k, cid, t, "chunk"), "provenance": "explicit", "rule_id": "inline:rel", "confidence": PROVENANCE_PRIORITY["inline:rel"] } if sec: payload["target_section"] = sec edges.append(_edge(k, "chunk", cid, t, note_id, payload)) # B. Candidate Pool (WP-15b Validierte KI-Kanten) pool = ch.get("candidate_pool") or ch.get("candidate_edges") or [] for cand in pool: raw_t, k, p = cand.get("to"), cand.get("kind", "related_to"), cand.get("provenance", "semantic_ai") t, sec = parse_link_target(raw_t, note_id) if t: # WP-24c v4.0.0: rule_id und variant werden nur im Payload gespeichert, fließen nicht in die ID ein payload = { "chunk_id": cid, "edge_id": _mk_edge_id(k, cid, t, "chunk"), "provenance": p, "rule_id": f"candidate:{p}", "confidence": PROVENANCE_PRIORITY.get(p, 0.90) } if sec: payload["target_section"] = sec edges.append(_edge(k, "chunk", cid, t, note_id, payload)) # C. Callouts (> [!edge]) call_pairs, rem2 = extract_callout_relations(rem) for k, raw_t in call_pairs: t, sec = parse_link_target(raw_t, note_id) if not t: continue # WP-24c v4.0.0: rule_id und variant werden nur im Payload gespeichert, fließen nicht in die ID ein payload = { "chunk_id": cid, "edge_id": _mk_edge_id(k, cid, t, "chunk"), "provenance": "explicit", "rule_id": "callout:edge", "confidence": PROVENANCE_PRIORITY["callout:edge"] } if sec: payload["target_section"] = sec edges.append(_edge(k, "chunk", cid, t, note_id, payload)) # D. Standard Wikilinks & Typ-Defaults refs = extract_wikilinks(rem2) for raw_r in refs: r, sec = parse_link_target(raw_r, note_id) if not r: continue # WP-24c v4.0.0: rule_id und variant werden nur im Payload gespeichert, fließen nicht in die ID ein payload = { "chunk_id": cid, "ref_text": raw_r, "edge_id": _mk_edge_id("references", cid, r, "chunk"), "provenance": "explicit", "rule_id": "explicit:wikilink", "confidence": PROVENANCE_PRIORITY["explicit:wikilink"] } if sec: payload["target_section"] = sec edges.append(_edge("references", "chunk", cid, r, note_id, payload)) # Automatische Kanten-Vererbung aus types.yaml for rel in defaults: if rel != "references": # WP-24c v4.0.0: rule_id und variant werden nur im Payload gespeichert, fließen nicht in die ID ein def_payload = { "chunk_id": cid, "edge_id": _mk_edge_id(rel, cid, r, "chunk"), "provenance": "rule", "rule_id": f"edge_defaults:{rel}", "confidence": PROVENANCE_PRIORITY["edge_defaults"] } if sec: def_payload["target_section"] = sec edges.append(_edge(rel, "chunk", cid, r, note_id, def_payload)) refs_all.extend([parse_link_target(r, note_id)[0] for r in refs]) # 3) Note-Scope (Grobe Struktur-Verbindungen) if include_note_scope_refs: cleaned_note_refs = [parse_link_target(r, note_id)[0] for r in (note_level_references or [])] refs_note = _dedupe_seq((refs_all or []) + cleaned_note_refs) for r in refs_note: if not r: continue # WP-24c v4.0.0: rule_id wird nur im Payload gespeichert, fließt nicht in die ID ein edges.append(_edge("references", "note", note_id, r, note_id, { "edge_id": _mk_edge_id("references", note_id, r, "note"), "provenance": "explicit", "rule_id": "explicit:note_scope", "confidence": PROVENANCE_PRIORITY["explicit:note_scope"] })) # Backlinks zur Stärkung der Bidirektionalität edges.append(_edge("backlink", "note", r, note_id, note_id, { "edge_id": _mk_edge_id("backlink", r, note_id, "note"), "provenance": "rule", "rule_id": "derived:backlink", "confidence": PROVENANCE_PRIORITY["derived:backlink"] })) # 4) De-Duplizierung (In-Place) # WP-24c v4.0.0: Da die EDGE-ID nur auf 4 Parametern basiert (kind, source, target, scope), # werden Links auf unterschiedliche Abschnitte derselben Note durch die De-Duplizierung # konsolidiert. Die Sektion-Information bleibt im Payload (target_section) erhalten. unique_map: Dict[str, dict] = {} for e in edges: eid = e["edge_id"] # Höhere Confidence gewinnt bei identischer ID if eid not in unique_map or e.get("confidence", 0) > unique_map[eid].get("confidence", 0): unique_map[eid] = e return list(unique_map.values())