mindnet/app/core/chunk_payload.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Modul: app/core/chunk_payload.py
Version: 2.1.0
Datum: 2025-10-01

Zweck
-----
Erzeugt Chunk-Payloads für Qdrant. Voll abwärtskompatibel zu v2.0.1 und erweitert um:
- Übernahme vorhandener deterministischer Chunk-IDs (falls vom Chunker geliefert).
- Durchreichen von token_count, section_title, section_path (falls vorhanden).
- Exaktere start/end-Offsets, wenn der vollständige Note-Body (note_text) vorliegt.
- Berechnung von overlap_right (nicht mehr pauschal 0).

Felder (wie 2.0.1, beibehalten):
  - window        : Fenstertext inkl. Overlap (für Embeddings & Link-Erkennung)
  - text          : effektiver Segmenttext ohne linkes Overlap (für Rekonstruktion)
  - start, end    : Offsets des Segments im Gesamtkorpus (0-basiert, [start, end))
  - overlap_left  : Zeichen-Overlap zum VORHERIGEN Fenster
  - overlap_right : Zeichen-Overlap zum NÄCHSTEN Fenster
  - note_id, chunk_id, id (Alias), chunk_index, seq, path, type, title, tags

Abwärtskompatible Aliasse / Verhalten:
  - 'id' == 'chunk_id' (Edges nutzen häufig 'id').
  - Fensterquelle wird robust aus ('window'|'text'|'content'|'raw') gelesen.
  - Pfad bleibt relativ und nutzt forward slashes.

Nutzung
-------
    from app.core.chunk_payload import make_chunk_payloads
    pls = make_chunk_payloads(frontmatter, rel_path, chunks, note_text=full_body)

Parameter
---------
- frontmatter : dict   – erwartet mind. 'id' (Note-ID); optional 'title','type','tags'
- rel_path    : str    – relativer Pfad der Note im Vault (forward slashes, ohne führenden '/')
- chunks      : Seq[object|dict]
- note_text   : str|None – optional kompletter Body für exakte Offsets

Hinweis
------
Dein Chunker liefert bereits Fenster mit vorangestelltem Overlap. Wir ermitteln den
linken Overlap gegen den bisher rekonstruierten Text und entfernen ihn in 'text'.

Lizenz: MIT (projektintern)
"""
from __future__ import annotations

from typing import Any, Dict, Iterable, List, Optional, Tuple, Union


# ------------------------------- Utils ------------------------------- #

def _get_attr_or_key(obj: Any, key: str, default=None):
    """Liest Feld 'key' aus Objekt ODER Dict; sonst default."""
    if obj is None:
        return default
    if isinstance(obj, dict):
        return obj.get(key, default)
    return getattr(obj, key, default)

def _as_window_text(chunk: Any) -> str:
    """
    Extrahiert den Fenstertext robust aus Chunk (Objekt oder Dict).
    Bevorzugt 'window', dann 'text', 'content', 'raw'.
    """
    for k in ("window", "text", "content", "raw"):
        v = _get_attr_or_key(chunk, k, None)
        if isinstance(v, str) and v:
            return v
    # Fallback: falls der Chunk selbst ein String ist (exotisch)
    if isinstance(chunk, str):
        return chunk
    return ""

def _to_int(x: Any, default: int = 0) -> int:
    try:
        return int(x)
    except Exception:
        return default

def _normalize_rel_path(p: str) -> str:
    p = (p or "").replace("\\", "/")
    # kein führender Slash (relativ bleiben)
    while p.startswith("/"):
        p = p[1:]
    return p


# ---------------------- Overlap & Offsets ---------------------------- #

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 ""
        max_k = min(len(w), len(reconstructed))
        k = 0
        # Suche von groß nach klein (einfach, robust 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

def _overlap_len_suffix_prefix(a_suffix: str, b_prefix: str, max_probe: int = 4096) -> int:
    """
    Länge des längsten Suffixes von a_suffix, der Prefix von b_prefix ist (bruteforce, begrenzt).
    """
    if not a_suffix or not b_prefix:
        return 0
    a = a_suffix[-max_probe:]
    b = b_prefix[:max_probe]
    n = min(len(a), len(b))
    for k in range(n, 0, -1):
        if a[-k:] == b[:k]:
            return k
    return 0


# ----------------------------- 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 (kompatibel zu v2.0.1, erweitert).

    Rückgabe-Felder (wichtigste):
      note_id, chunk_id, id, chunk_index, seq, path,
      window, text, start, end, overlap_left, overlap_right,
      token_count?, section_title?, section_path?, 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)
    rel_path = _normalize_rel_path(rel_path)

    # 1) Fenstertexte & Sequenzen & (falls vorhanden) vorgegebene IDs sammeln
    windows: List[str] = []
    seqs: List[int] = []
    ids_in: List[Optional[str]] = []
    token_counts: List[Optional[int]] = []
    section_titles: List[Optional[str]] = []
    section_paths: List[Optional[str]] = []

    chunks_list = list(chunks)  # falls Iterator
    for idx, c in enumerate(chunks_list):
        windows.append(_as_window_text(c))
        seqs.append(_to_int(_get_attr_or_key(c, "seq", _get_attr_or_key(c, "chunk_index", idx)), idx))
        # übernehme deterministische ID (falls vom Chunker geliefert)
        cid = _get_attr_or_key(c, "chunk_id", _get_attr_or_key(c, "id", None))
        ids_in.append(str(cid) if isinstance(cid, str) and cid else None)
        token_counts.append(_to_int(_get_attr_or_key(c, "token_count", None), 0) if _get_attr_or_key(c, "token_count", None) is not None else None)
        section_titles.append(_get_attr_or_key(c, "section_title", None))
        section_paths.append(_get_attr_or_key(c, "section_path", None))

    # 2) Nicht-überlappende Segmente berechnen
    segments, overlaps_left = _dedupe_windows_to_segments(windows)

    # 3) overlap_right berechnen (Blick auf nächstes Fenster)
    overlaps_right: List[int] = []
    for i in range(len(windows)):
        if i + 1 < len(windows):
            ov = _overlap_len_suffix_prefix(windows[i], windows[i + 1], max_probe=4096)
        else:
            ov = 0
        overlaps_right.append(ov)

    # 4) start/end Offsets bestimmen
    #    - wenn note_text vorhanden: vorwärtsgerichtetes Matching ab der letzten Endposition
    #    - sonst kumulativ (wie in v2.0.1)
    starts: List[int] = [0] * len(segments)
    ends: List[int] = [0] * len(segments)
    pos = 0
    if isinstance(note_text, str) and note_text:
        search_pos = 0
        for i, seg in enumerate(segments):
            if not seg:
                starts[i] = ends[i] = search_pos
                continue
            j = note_text.find(seg, search_pos)
            if j >= 0:
                starts[i] = j
                ends[i] = j + len(seg)
                search_pos = ends[i]
            else:
                # Fallback: kumulativ
                starts[i] = pos
                pos += len(seg)
                ends[i] = pos
    else:
        for i, seg in enumerate(segments):
            starts[i] = pos
            pos += len(seg)
            ends[i] = pos

    # 5) Payload-Dicts zusammenstellen
    payloads: List[Dict[str, Any]] = []
    for i, (win, seg) in enumerate(zip(windows, segments)):
        # finale chunk_id: bevorzugt deterministische Vorgabe, sonst Fallback
        chunk_id = ids_in[i] or f"{note_id}#{i+1}"

        pl: Dict[str, Any] = {
            # Identität
            "note_id": note_id,
            "chunk_id": chunk_id,
            "id": chunk_id,  # Alias für Abwärtskompatibilität

            # Ordnung
            "chunk_index": i,
            "seq": seqs[i],

            # Pfad
            "path": rel_path,

            # Texte
            "window": win,   # mit Overlap
            "text": seg,     # Overlap entfernt

            # Offsets & Overlaps
            "start": starts[i],
            "end": ends[i],
            "overlap_left": overlaps_left[i],
            "overlap_right": overlaps_right[i],
        }

        # optionale Metafelder aus Note / Chunk
        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

        if token_counts[i] is not None:
            pl["token_count"] = int(token_counts[i])
        if section_titles[i] is not None:
            pl["section_title"] = section_titles[i]
        if section_paths[i] is not None:
            # normalisiere section_path minimal
            sp = str(section_paths[i]).replace("\\", "/")
            pl["section_path"] = sp if sp else "/"

        payloads.append(pl)

    return payloads


# __main__ (optionale Mini-Demo)
if __name__ == "__main__":  # pragma: no cover
    demo_fm = {"id": "demo", "title": "Demo", "type": "concept"}
    demo_chunks = [
        {"id": "demo#1", "text": "Alpha Beta Gamma", "token_count": 3, "section_title": "Intro", "section_path": "/intro"},
        {"id": "demo#2", "text": "Gamma Delta", "token_count": 2, "section_title": "Teil 1", "section_path": "/teil-1"},
        {"id": "demo#3", "text": "Delta Epsilon Zeta", "token_count": 3, "section_title": "Teil 2", "section_path": "/teil-2"},
    ]
    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)