mindnet/app/core/chunk_payload.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Modul: app/core/chunk_payload.py
Version: 2.0.0
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:

  - text            : effektiver, nicht-überlappender Segmenttext (für Rekonstruktion)
  - window          : Fenstertext inkl. Overlap (für Embeddings)
  - start, end      : absolute Offsets (0-basiert) des effektiven Segments im Gesamtkorpus
  - overlap_left    : Anzahl überlappender Zeichen zum **vorigen** Fenster
  - overlap_right   : Anzahl überlappender Zeichen zum **nächsten** Fenster

Abwärtskompatibel bleiben:
  - chunk_id (note_id#<n>), chunk_index, seq, path, note_id, type, title, tags, etc.

Aufruf (typisch aus dem Importer)
---------------------------------
    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:
    c.text  ODER c.content ODER c.raw  (falls als 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 ------------------------------- #

def _as_text(window_candidate: Any) -> str:
    """Extrahiert Fenstertext aus beliebigem Chunk-Objekt/Dikt."""
    if window_candidate is None:
        return ""
    # Objekt mit Attributen
    for k in ("text", "content", "raw", "window"):
        v = getattr(window_candidate, k, None) if not isinstance(window_candidate, dict) else window_candidate.get(k)
        if isinstance(v, str) and v:
            return v
    # Fallback: string-repr
    if isinstance(window_candidate, str):
        return window_candidate
    return ""

def _get_int(x: Any, default: int = 0) -> int:
    try:
        return int(x)
    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]]:
    """
    Ermittelt nicht-überlappende Segmente zu einer geordneten Folge von Fenster-Strings.
    Gibt (segments, overlaps_left) zurück, wobei:
      - segments[i]      = Fenster[i] ohne das vorangestellte Overlap
      - overlaps_left[i] = Länge des Overlaps von Fenster[i] zum bisher rekonstruierten Text
    """
    segments: List[str] = []
    overlaps_left: List[int] = []
    reconstructed = ""
    for w in windows:
        w = w or ""
        # finde größtes k, sodass reconstructed.endswith(w[:k])
        max_k = min(len(w), max(0, len(reconstructed)))
        k = 0
        # Suche von groß nach klein (einfache O(n^2) – ausreichend bei kurzen Fenstern)
        for cand in range(max_k, -1, -1):
            if reconstructed.endswith(w[:cand]):
                k = cand
                break
        seg = w[k:]
        segments.append(seg)
        overlaps_left.append(k)
        reconstructed += seg
    return segments, overlaps_left

# ----------------------------- Public API ---------------------------- #

def make_chunk_payloads(
    frontmatter: Dict[str, Any],
    rel_path: str,
    chunks: Iterable[Union[Dict[str, Any], Any]],
    note_text: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """
    Baut Payload-Dicts pro Chunk.

    Parameter
    ---------
    frontmatter : dict   – erwartete Keys: id (note_id), title, type, tags (optional)
    rel_path    : str    – relativer Pfad der Note im Vault
    chunks      : iter   – Sequenz von Chunk-Objekten/-Dicts mit Fenstertext
    note_text   : str?   – optionaler Gesamtkorpus (Body) für exakte Offsets

    Rückgabe
    --------
    Liste von Payload-Dicts. Wichtige Felder:
      note_id, chunk_id, chunk_index, seq, path, text, window, 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
    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:
            s = getattr(c, "seq", getattr(c, "chunk_index", idx))
        seqs.append(_get_int(s, idx))

    # 2) Nicht-überlappende Segmente berechnen
    segments, overlaps_left = _dedupe_windows_to_segments(windows)
    overlaps_right = [0] * len(segments)
    # 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
    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
    payloads: List[Dict[str, Any]] = []
    for i, (win, seg) in enumerate(zip(windows, segments)):
        pl: Dict[str, Any] = {
            "note_id": note_id,
            "chunk_id": f"{note_id}#{i+1}",
            "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)

            # Offsets & Overlaps
            "start": starts[i],
            "end": ends[i],
            "overlap_left": overlaps_left[i],
            "overlap_right": overlaps_right[i],
        }
        if note_type is not None:
            pl["type"] = note_type
        if note_title is not None:
            pl["title"] = note_title
        if note_tags is not None:
            pl["tags"] = note_tags
        payloads.append(pl)

    return payloads

# __main__ (optionaler Mini-Test)
if __name__ == "__main__":  # pragma: no cover
    demo_fm = {"id": "demo", "title": "Demo", "type": "concept"}
    demo_chunks = [
        {"text": "Alpha Beta Gamma"},
        {"text": "Gamma Delta"},
        {"text": "Delta Epsilon Zeta"},
    ]
    pls = make_chunk_payloads(demo_fm, "x/demo.md", demo_chunks, note_text="Alpha Beta Gamma Delta Epsilon Zeta")
    from pprint import pprint
    pprint(pls)
    recon = "".join(p["text"] for p in pls)
    print("RECON:", recon)