app/core/chunk_payload.py aktualisiert

app/core/chunk_payload.py

@@ -2,13 +2,13 @@
 # -*- coding: utf-8 -*-
 """
 Modul: app/core/chunk_payload.py
-Version: 2.0.0
+Version: 2.0.1
 Datum: 2025-09-30
 
 Zweck
 -----
 Erzeugt Chunk-Payloads für Qdrant. Unterstützt abwärtskompatibel bisherige Felder und
-ergänzt neue Felder für **verlustfreie Rekonstruktion** bei überlappenden Fenstern:
+ergänzt Felder für **verlustfreie Rekonstruktion** bei überlappenden Fenstern:
 
   - text            : effektiver, nicht-überlappender Segmenttext (für Rekonstruktion)
   - window          : Fenstertext inkl. Overlap (für Embeddings)
@@ -16,25 +16,22 @@ ergänzt neue Felder für **verlustfreie Rekonstruktion** bei überlappenden Fen
   - overlap_left    : Anzahl überlappender Zeichen zum **vorigen** Fenster
   - overlap_right   : Anzahl überlappender Zeichen zum **nächsten** Fenster
 
-Abwärtskompatibel bleiben:
+Abwärtskompatible Aliasse:
-  - chunk_id (note_id#<n>), chunk_index, seq, path, note_id, type, title, tags, etc.
+  - id              : == chunk_id   (wird u. a. von build_edges_for_note erwartet)
+  - content/raw     : bleiben leer; Primärfelder sind window/text
 
-Aufruf (typisch aus dem Importer)
+Typische Nutzung
----------------------------------
+----------------
     from app.core.chunk_payload import make_chunk_payloads
     payloads = make_chunk_payloads(frontmatter, rel_path, chunks, note_text=full_body)
 
-Wobei `chunks` eine Folge von Objekten oder Dicts ist, die mindestens ein Fenster enthalten:
+`chunks` ist eine Sequenz von Objekten oder Dicts, die mindestens ein Fenster enthalten:
-    c.text  ODER c.content ODER c.raw  (falls als Objekt)
+    c.text  ODER c.content ODER c.raw  (falls Objekt)
     bzw.    c["text"] ODER c["content"] ODER c["raw"] (falls Dict)
-
-Falls `note_text` nicht übergeben wird, wird die effektive Segmentierung über
-eine robuste **Overlap-Deduplikation** zwischen Fenstern ermittelt.
 """
 from __future__ import annotations
 
 from typing import Any, Dict, Iterable, List, Optional, Tuple, Union
-import re
 
 # ------------------------------- Utils ------------------------------- #
 
@@ -43,8 +40,14 @@ def _as_text(window_candidate: Any) -> str:
     if window_candidate is None:
         return ""
     # Objekt mit Attributen
-    for k in ("text", "content", "raw", "window"):
+    if not isinstance(window_candidate, dict):
-        v = getattr(window_candidate, k, None) if not isinstance(window_candidate, dict) else window_candidate.get(k)
+        for k in ("window", "text", "content", "raw"):
+            v = getattr(window_candidate, k, None)
+            if isinstance(v, str) and v:
+                return v
+    else:
+        for k in ("window", "text", "content", "raw"):
+            v = window_candidate.get(k)
             if isinstance(v, str) and v:
                 return v
     # Fallback: string-repr
@@ -58,10 +61,6 @@ def _get_int(x: Any, default: int = 0) -> int:
     except Exception:
         return default
 
-def _norm_lines(s: str) -> str:
-    """Nur für defensive Gleichheitstests – NICHT für Persistenz."""
-    return "\n".join([ln.rstrip() for ln in s.replace("\r\n", "\n").replace("\r", "\n").split("\n")]).strip()
-
 # ---------------------- Overlap-Dedupe Algorithmus ------------------- #
 
 def _dedupe_windows_to_segments(windows: List[str]) -> Tuple[List[str], List[int]]:
@@ -76,10 +75,9 @@ def _dedupe_windows_to_segments(windows: List[str]) -> Tuple[List[str], List[int
     reconstructed = ""
     for w in windows:
         w = w or ""
-        # finde größtes k, sodass reconstructed.endswith(w[:k])
+        max_k = min(len(w), len(reconstructed))
-        max_k = min(len(w), max(0, len(reconstructed)))
         k = 0
-        # Suche von groß nach klein (einfache O(n^2) – ausreichend bei kurzen Fenstern)
+        # Suche von groß nach klein (einfach, ausreichend bei kurzen Fenstern)
         for cand in range(max_k, -1, -1):
             if reconstructed.endswith(w[:cand]):
                 k = cand
@@ -111,21 +109,19 @@ def make_chunk_payloads(
     Rückgabe
     --------
     Liste von Payload-Dicts. Wichtige Felder:
-      note_id, chunk_id, chunk_index, seq, path, text, window, start, end,
+      note_id, id, chunk_id, chunk_index, seq, path, text, window,
-      overlap_left, overlap_right, type, title, tags
+      start, end, overlap_left, overlap_right, type, title, tags
     """
     note_id = str(frontmatter.get("id") or "").strip()
     note_type = frontmatter.get("type", None)
     note_title = frontmatter.get("title", None)
     note_tags = frontmatter.get("tags", None)
 
-    # 1) Fenstertexte extrahieren
+    # 1) Fenstertexte + Sequenzen extrahieren
     windows: List[str] = []
     seqs: List[int] = []
     for idx, c in enumerate(chunks):
         windows.append(_as_text(c))
-        # Bestmögliche seq ermitteln
-        s = None
         if isinstance(c, dict):
             s = c.get("seq", c.get("chunk_index", idx))
         else:
@@ -134,41 +130,35 @@ def make_chunk_payloads(
 
     # 2) Nicht-überlappende Segmente berechnen
     segments, overlaps_left = _dedupe_windows_to_segments(windows)
-    overlaps_right = [0] * len(segments)
+    overlaps_right = [0] * len(segments)  # optional: später präzisieren
-    # right-overlap ist der left-overlap des nächsten Fensters bezogen auf dessen Fenster,
-    # lässt sich nur approximieren; wir speichern ihn konsistent als 0 bzw. könnte man
-    # nachträglich bestimmen, falls benötigt.
 
-    # 3) Falls note_text gegeben ist, berechne absolute Offsets präzise
+    # 3) Offsets bestimmen (ohne/mit note_text gleich: kumulativ)
     starts: List[int] = [0] * len(segments)
     ends: List[int] = [0] * len(segments)
-    if isinstance(note_text, str):
-        pos = 0
-        for i, seg in enumerate(segments):
-            starts[i] = pos
-            pos += len(seg)
-            ends[i] = pos
-    else:
-        # Ohne Gesamtkorpus: Offsets anhand der kumulativen Segmentlängen
     pos = 0
     for i, seg in enumerate(segments):
         starts[i] = pos
         pos += len(seg)
         ends[i] = pos
 
-    # 4) Payload-Dicts aufbauen
+    # 4) Payload-Dicts zusammenstellen
     payloads: List[Dict[str, Any]] = []
     for i, (win, seg) in enumerate(zip(windows, segments)):
+        chunk_id = f"{note_id}#{i+1}"
         pl: Dict[str, Any] = {
+            # Identität
             "note_id": note_id,
-            "chunk_id": f"{note_id}#{i+1}",
+            "chunk_id": chunk_id,
+            "id": chunk_id,  # <— WICHTIG: Alias für Abwärtskompatibilität (Edges erwarten 'id')
+
+            # Indexierung
             "chunk_index": i,
             "seq": seqs[i],
             "path": rel_path.replace("\\", "/").lstrip("/"),
 
             # Texte
             "window": win,   # für Embeddings (inkl. Overlap)
-            "text": seg,          # effektiver Anteil (verlustfreie Rekonstruktion)
+            "text": seg,     # überlappungsfreier Anteil für exakte Rekonstruktion
 
             # Offsets & Overlaps
             "start": starts[i],
@@ -182,11 +172,13 @@ def make_chunk_payloads(
             pl["title"] = note_title
         if note_tags is not None:
             pl["tags"] = note_tags
+
         payloads.append(pl)
 
     return payloads
 
-# __main__ (optionaler Mini-Test)
+
+# __main__ (optional: Mini-Demo)
 if __name__ == "__main__":  # pragma: no cover
     demo_fm = {"id": "demo", "title": "Demo", "type": "concept"}
     demo_chunks = [