mindnet/app/core/chunking/chunking_strategies.py

"""
FILE: app/core/chunking/chunking_strategies.py
DESCRIPTION: Mathematische Splitting-Strategien.
             AUDIT v3.3.2: 100% Konformität zur 'by_heading' Spezifikation.
             - Implementiert Hybrid-Safety-Net (Sliding Window für Übergrößen).
             - Breadcrumb-Kontext im Window (H1 > H2).
             - Sliding Window mit H1-Kontext (Gold-Standard v3.1.0).
"""
from typing import List, Dict, Any, Optional
from .chunking_models import RawBlock, Chunk
from .chunking_utils import estimate_tokens
from .chunking_parser import split_sentences

def _create_context_win(doc_title: str, sec_title: Optional[str], text: str) -> str:
    """Baut den Breadcrumb-Kontext für das Embedding-Fenster."""
    parts = []
    if doc_title: parts.append(doc_title)
    if sec_title and sec_title != doc_title: parts.append(sec_title)
    prefix = " > ".join(parts)
    return f"{prefix}\n{text}".strip() if prefix else text

def strategy_sliding_window(blocks: List[RawBlock], 
                            config: Dict[str, Any], 
                            note_id: str, 
                            context_prefix: str = "") -> List[Chunk]:
    """
    Fasst Blöcke zusammen und schneidet bei 'target' Tokens.
    Ignoriert H2-Überschriften beim Splitting, um Kontext zu wahren.
    """
    target = config.get("target", 400)
    max_tokens = config.get("max", 600)
    overlap_val = config.get("overlap", (50, 80))
    overlap = sum(overlap_val) // 2 if isinstance(overlap_val, tuple) else overlap_val
    
    chunks: List[Chunk] = []
    buf: List[RawBlock] = []

    def _add(txt, sec, path):
        idx = len(chunks)
        # H1-Kontext Präfix für das Window-Feld
        win = f"{context_prefix}\n{txt}".strip() if context_prefix else txt
        chunks.append(Chunk(
            id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx,
            text=txt, window=win, token_count=estimate_tokens(txt),
            section_title=sec, section_path=path,
            neighbors_prev=None, neighbors_next=None
        ))

    def flush():
        nonlocal buf
        if not buf: return
        text_body = "\n\n".join([b.text for b in buf])
        sec_title = buf[-1].section_title; sec_path = buf[-1].section_path
        
        if estimate_tokens(text_body) <= max_tokens:
            _add(text_body, sec_title, sec_path)
        else:
            sents = split_sentences(text_body); cur_sents = []; cur_len = 0
            for s in sents:
                slen = estimate_tokens(s)
                if cur_len + slen > target and cur_sents:
                    _add(" ".join(cur_sents), sec_title, sec_path)
                    ov_s = []; ov_l = 0
                    for os in reversed(cur_sents):
                        if ov_l + estimate_tokens(os) < overlap:
                            ov_s.insert(0, os); ov_l += estimate_tokens(os)
                        else: break
                    cur_sents = list(ov_s); cur_sents.append(s); cur_len = ov_l + slen
                else:
                    cur_sents.append(s); cur_len += slen
            if cur_sents:
                _add(" ".join(cur_sents), sec_title, sec_path)
        buf = []

    for b in blocks:
        # H2-Überschriften werden ignoriert, um den Zusammenhang zu wahren
        if b.kind == "heading": continue 
        if estimate_tokens("\n\n".join([x.text for x in buf])) + estimate_tokens(b.text) >= target:
            flush()
        buf.append(b)
    flush()
    return chunks

def strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]:
    """
    Splittet Text basierend auf Markdown-Überschriften mit Hybrid-Safety-Net.
    """
    strict = config.get("strict_heading_split", False)
    target = config.get("target", 400)
    max_tokens = config.get("max", 600)
    split_level = config.get("split_level", 2)
    overlap = sum(config.get("overlap", (50, 80))) // 2
    
    chunks: List[Chunk] = []
    buf: List[str] = []
    cur_tokens = 0

    def _add_to_chunks(txt, title, path):
        idx = len(chunks)
        win = _create_context_win(doc_title, title, txt)
        chunks.append(Chunk(
            id=f"{note_id}#c{idx:02d}", note_id=note_id, index=idx,
            text=txt, window=win, token_count=estimate_tokens(txt),
            section_title=title, section_path=path,
            neighbors_prev=None, neighbors_next=None
        ))

    def _flush(title, path):
        nonlocal buf, cur_tokens
        if not buf: return
        full_text = "\n\n".join(buf)
        if estimate_tokens(full_text) <= max_tokens:
            _add_to_chunks(full_text, title, path)
        else:
            sents = split_sentences(full_text); cur_sents = []; sub_len = 0
            for s in sents:
                slen = estimate_tokens(s)
                if sub_len + slen > target and cur_sents:
                    _add_to_chunks(" ".join(cur_sents), title, path)
                    ov_s = []; ov_l = 0
                    for os in reversed(cur_sents):
                        if ov_l + estimate_tokens(os) < overlap:
                            ov_s.insert(0, os); ov_l += estimate_tokens(os)
                        else: break
                    cur_sents = list(ov_s); cur_sents.append(s); sub_len = ov_l + slen
                else: cur_sents.append(s); sub_len += slen
            if cur_sents: _add_to_chunks(" ".join(cur_sents), title, path)
        buf = []; cur_tokens = 0

    for b in blocks:
        if b.kind == "heading":
            if b.level < split_level: _flush(b.section_title, b.section_path)
            elif b.level == split_level:
                if strict or cur_tokens >= target: _flush(b.section_title, b.section_path)
            continue
        bt = estimate_tokens(b.text)
        if cur_tokens + bt > max_tokens and buf: _flush(b.section_title, b.section_path)
        buf.append(b.text); cur_tokens += bt
    if buf:
        last_b = blocks[-1] if blocks else None
        _flush(last_b.section_title if last_b else None, last_b.section_path if last_b else "/")
    return chunks