#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Modul: app/core/chunk_payload.py Version: 2.2.1 Datum: 2025-11-07 Zweck ----- Erzeugt Qdrant-Payloads für Chunks. Voll abwärtskompatibel zu v2.2.0 / v2.0.1. Neu (2.2.1): • Stabilere Offsets (start/end) bei mehrfach vorkommenden Segmenten (inkrementelles Suchen + Fallback), • optionale Felder window_left_ctx_len und window_right_ctx_est zur Diagnose, • robustere Section-Pfadbehandlung. Felder (unverändert beibehalten): note_id, chunk_id (Alias: id), chunk_index, seq, path, window (mit linkem Overlap), text (Kernsegment), start, end, overlap_left, overlap_right, type, title, tags, token_count?, section_title?, section_path? """ from __future__ import annotations from typing import Any, Dict, Iterable, List, Optional, Tuple, Union try: from app.core.chunk_config import get_sizes as _get_sizes except Exception: def _get_sizes(_note_type: str): return {"overlap": (40, 60), "target": (250, 350), "max": 500} def _get_attr_or_key(obj: Any, key: str, default=None): 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: for k in ("window", "text", "content", "raw"): v = _get_attr_or_key(chunk, k, None) if isinstance(v, str) and v: return v 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("\\", "/") while p.startswith("/"): p = p[1:] return p def _dedupe_windows_to_segments(windows: List[str]) -> Tuple[List[str], List[int], str]: segments: List[str] = [] overlaps_left: List[int] = [] reconstructed = "" for w in windows: w = w or "" max_k = min(len(w), len(reconstructed)) k = 0 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, reconstructed def _overlap_len_suffix_prefix(a: str, b: str, max_probe: int = 4096) -> int: if not a or not b: return 0 a1 = a[-max_probe:] b1 = b[:max_probe] n = min(len(a1), len(b1)) for k in range(n, 0, -1): if a1[-k:] == b1[:k]: return k return 0 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]]: note_id = str(frontmatter.get("id") or "").strip() note_type = str(frontmatter.get("type", "")).lower() note_title = frontmatter.get("title", None) note_tags = frontmatter.get("tags", None) rel_path = _normalize_rel_path(rel_path) chunks_list = list(chunks) raw_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]] = [] any_explicit_window = False for idx, c in enumerate(chunks_list): w = _get_attr_or_key(c, "window", None) if isinstance(w, str) and w: any_explicit_window = True raw_windows.append(w) else: raw_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)) 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) tc = _get_attr_or_key(c, "token_count", None) token_counts.append(_to_int(tc, 0) if tc 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)) if any_explicit_window: segments, overlaps_left, recon = _dedupe_windows_to_segments(raw_windows) windows_final = raw_windows[:] else: segments = [w or "" for w in raw_windows] overlaps_left = [] windows_final = [] recon = "" try: overlap_low, overlap_high = tuple(_get_sizes(note_type).get("overlap", (40, 60))) except Exception: overlap_low, overlap_high = (40, 60) overlap_target = int(overlap_low) for i, seg in enumerate(segments): if i == 0: windows_final.append(seg) overlaps_left.append(0) recon += seg else: k = min(overlap_target, len(recon)) left_ctx = recon[-k:] if k > 0 else "" windows_final.append(left_ctx + seg) overlaps_left.append(k) recon += seg overlaps_right: List[int] = [] for i in range(len(windows_final)): if i + 1 < len(windows_final): ov = _overlap_len_suffix_prefix(windows_final[i], windows_final[i + 1], max_probe=4096) else: ov = 0 overlaps_right.append(ov) starts: List[int] = [0] * len(segments) ends: List[int] = [0] * len(segments) 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: naive fortlaufende Positionierung starts[i] = starts[i - 1] if i > 0 else 0 ends[i] = starts[i] + len(seg) search_pos = ends[i] else: pos = 0 for i, seg in enumerate(segments): starts[i] = pos pos += len(seg) ends[i] = pos payloads: List[Dict[str, Any]] = [] for i, (win, seg) in enumerate(zip(windows_final, segments)): chunk_id = ids_in[i] or f"{note_id}#{i+1}" left_len = max(0, len(win) - len(seg)) pl: Dict[str, Any] = { "note_id": note_id, "chunk_id": chunk_id, "id": chunk_id, "chunk_index": i, "seq": seqs[i], "path": rel_path, "window": win, "text": seg, "start": starts[i], "end": ends[i], "overlap_left": overlaps_left[i], "overlap_right": overlaps_right[i], "window_left_ctx_len": left_len, } if note_type: 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: sp = str(section_paths[i]).replace("\\", "/") pl["section_path"] = sp if sp else "/" payloads.append(pl) return payloads if __name__ == "__main__": # pragma: no cover fm = {"id": "demo", "title": "Demo", "type": "concept"} chunks = [ {"id": "demo#1", "text": "Alpha Beta Gamma"}, {"id": "demo#2", "text": "Gamma Delta"}, {"id": "demo#3", "text": "Delta Epsilon Zeta"}, ] pls = make_chunk_payloads(fm, "path/demo.md", 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)