""" FILE: app/core/chunker.py DESCRIPTION: Zerlegt Texte in Chunks (Sliding Window oder nach Headings). Orchestriert die Smart-Edge-Allocation via SemanticAnalyzer. VERSION: 2.9.0 (Feat: Hybrid Strict Splitting with Size Safety) STATUS: Active DEPENDENCIES: app.services.semantic_analyzer, app.core.derive_edges, markdown_it, yaml, asyncio EXTERNAL_CONFIG: config/types.yaml LAST_ANALYSIS: 2025-12-16 """ from __future__ import annotations from dataclasses import dataclass from typing import List, Dict, Optional, Tuple, Any, Set import re import math import yaml from pathlib import Path import asyncio import logging # Services from app.services.semantic_analyzer import get_semantic_analyzer # Core Imports try: from app.core.derive_edges import build_edges_for_note except ImportError: # Mock für Tests def build_edges_for_note(note_id, chunks, note_level_references=None, include_note_scope_refs=False): return [] logger = logging.getLogger(__name__) # ========================================== # 1. HELPER & CONFIG # ========================================== BASE_DIR = Path(__file__).resolve().parent.parent.parent CONFIG_PATH = BASE_DIR / "config" / "types.yaml" DEFAULT_PROFILE = {"strategy": "sliding_window", "target": 400, "max": 600, "overlap": (50, 80)} _CONFIG_CACHE = None def _load_yaml_config() -> Dict[str, Any]: global _CONFIG_CACHE if _CONFIG_CACHE is not None: return _CONFIG_CACHE if not CONFIG_PATH.exists(): return {} try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: data = yaml.safe_load(f) _CONFIG_CACHE = data return data except Exception: return {} def get_chunk_config(note_type: str) -> Dict[str, Any]: full_config = _load_yaml_config() profiles = full_config.get("chunking_profiles", {}) type_def = full_config.get("types", {}).get(note_type.lower(), {}) profile_name = type_def.get("chunking_profile") if not profile_name: profile_name = full_config.get("defaults", {}).get("chunking_profile", "sliding_standard") config = profiles.get(profile_name, DEFAULT_PROFILE).copy() if "overlap" in config and isinstance(config["overlap"], list): config["overlap"] = tuple(config["overlap"]) return config def extract_frontmatter_from_text(md_text: str) -> Tuple[Dict[str, Any], str]: fm_match = re.match(r'^\s*---\s*\n(.*?)\n---', md_text, re.DOTALL) if not fm_match: return {}, md_text try: frontmatter = yaml.safe_load(fm_match.group(1)) if not isinstance(frontmatter, dict): frontmatter = {} except yaml.YAMLError: frontmatter = {} text_without_fm = re.sub(r'^\s*---\s*\n(.*?)\n---', '', md_text, flags=re.DOTALL) return frontmatter, text_without_fm.strip() # ========================================== # 2. DATA CLASSES # ========================================== _SENT_SPLIT = re.compile(r'(?<=[.!?])\s+(?=[A-ZÄÖÜ0-9„(])'); _WS = re.compile(r'\s+') def estimate_tokens(text: str) -> int: return max(1, math.ceil(len(text.strip()) / 4)) def split_sentences(text: str) -> list[str]: text = _WS.sub(' ', text.strip()) if not text: return [] parts = _SENT_SPLIT.split(text) return [p.strip() for p in parts if p.strip()] @dataclass class RawBlock: kind: str; text: str; level: Optional[int]; section_path: str; section_title: Optional[str] @dataclass class Chunk: id: str; note_id: str; index: int; text: str; window: str; token_count: int section_title: Optional[str]; section_path: str neighbors_prev: Optional[str]; neighbors_next: Optional[str] suggested_edges: Optional[List[str]] = None # ========================================== # 3. PARSING & STRATEGIES # ========================================== def parse_blocks(md_text: str) -> Tuple[List[RawBlock], str]: """Zerlegt Text in logische Blöcke (Absätze, Header).""" blocks = [] h1_title = "Dokument" section_path = "/" current_h2 = None fm, text_without_fm = extract_frontmatter_from_text(md_text) h1_match = re.search(r'^#\s+(.*)', text_without_fm, re.MULTILINE) if h1_match: h1_title = h1_match.group(1).strip() lines = text_without_fm.split('\n') buffer = [] for line in lines: stripped = line.strip() if stripped.startswith('# '): if buffer: content = "\n".join(buffer).strip() if content: blocks.append(RawBlock("paragraph", content, None, section_path, current_h2)) buffer = [] blocks.append(RawBlock("heading", stripped, 1, section_path, current_h2)) elif stripped.startswith('## '): if buffer: content = "\n".join(buffer).strip() if content: blocks.append(RawBlock("paragraph", content, None, section_path, current_h2)) buffer = [] current_h2 = stripped[3:].strip() section_path = f"/{current_h2}" blocks.append(RawBlock("heading", stripped, 2, section_path, current_h2)) elif stripped.startswith('### '): if buffer: content = "\n".join(buffer).strip() if content: blocks.append(RawBlock("paragraph", content, None, section_path, current_h2)) buffer = [] blocks.append(RawBlock("heading", stripped, 3, section_path, current_h2)) elif not stripped: if buffer: content = "\n".join(buffer).strip() if content: blocks.append(RawBlock("paragraph", content, None, section_path, current_h2)) buffer = [] else: buffer.append(line) if buffer: content = "\n".join(buffer).strip() if content: blocks.append(RawBlock("paragraph", content, None, section_path, current_h2)) return blocks, h1_title def _create_chunk_obj(chunks_list: List[Chunk], note_id: str, txt: str, win: str, sec: Optional[str], path: str): idx = len(chunks_list) chunks_list.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, suggested_edges=[] )) def _strategy_sliding_window(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "", context_prefix: str = "") -> List[Chunk]: 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 = []; buf = [] def flush_buffer(): nonlocal buf if not buf: return text_body = "\n\n".join([b.text for b in buf]) win_body = f"{context_prefix}\n{text_body}".strip() if context_prefix else text_body # Basis-Info vom ersten Block im Buffer sec = buf[0].section_title if buf else None path = buf[0].section_path if buf else "/" if estimate_tokens(text_body) <= max_tokens: _create_chunk_obj(chunks, note_id, text_body, win_body, sec, path) else: # Fallback: Wenn Block zu groß, intern splitten (Sentence-Level) sentences = split_sentences(text_body) current_chunk_sents = [] current_len = 0 for sent in sentences: sent_len = estimate_tokens(sent) if current_len + sent_len > target and current_chunk_sents: c_txt = " ".join(current_chunk_sents) c_win = f"{context_prefix}\n{c_txt}".strip() if context_prefix else c_txt _create_chunk_obj(chunks, note_id, c_txt, c_win, sec, path) overlap_sents = [] ov_len = 0 for s in reversed(current_chunk_sents): if ov_len + estimate_tokens(s) < overlap: overlap_sents.insert(0, s) ov_len += estimate_tokens(s) else: break current_chunk_sents = list(overlap_sents) current_chunk_sents.append(sent) current_len = ov_len + sent_len else: current_chunk_sents.append(sent) current_len += sent_len if current_chunk_sents: c_txt = " ".join(current_chunk_sents) c_win = f"{context_prefix}\n{c_txt}".strip() if context_prefix else c_txt _create_chunk_obj(chunks, note_id, c_txt, c_win, sec, path) buf = [] for b in blocks: if b.kind == "heading": flush_buffer() current_buf_text = "\n\n".join([x.text for x in buf]) if buf and (estimate_tokens(current_buf_text) + estimate_tokens(b.text) >= target): flush_buffer() buf.append(b) flush_buffer() return chunks def _strategy_by_heading(blocks: List[RawBlock], config: Dict[str, Any], note_id: str, doc_title: str = "") -> List[Chunk]: """ MODUS: Structured / Heading Split - split_level: Ebene für logische Trennung (z.B. H2). - strict_heading_split: True: Trennt an jedem Header <= split_level. NEU v2.9: Wenn Inhalt > max_tokens, wird trotzdem gesplittet (Safety Split). False: Fasst zusammen bis 'target' erreicht ist. """ split_level = config.get("split_level", 2) target = config.get("target", 400) max_limit = config.get("max", 600) strict_mode = config.get("strict_heading_split", False) chunks = [] current_chunk_blocks = [] context_prefix = f"# {doc_title}" def has_content(blk_list): return any(b.kind != "heading" for b in blk_list) def flush_current_chunk(): nonlocal current_chunk_blocks if not current_chunk_blocks: return text_body = "\n\n".join([b.text for b in current_chunk_blocks]) win_body = f"{context_prefix}\n{text_body}".strip() first_b = current_chunk_blocks[0] sec = first_b.section_title path = first_b.section_path _create_chunk_obj(chunks, note_id, text_body, win_body, sec, path) current_chunk_blocks = [] def get_current_size(): txt = "\n\n".join([b.text for b in current_chunk_blocks]) return estimate_tokens(txt) for b in blocks: # 1. Header Logic (Struktur-Trigger) is_splitter = (b.kind == "heading" and b.level is not None and b.level <= split_level) if is_splitter: is_higher_hierarchy = (b.level < split_level) if strict_mode: # STRICT: # Wir splitten immer, außer der Vor-Chunk ist leer. if current_chunk_blocks and has_content(current_chunk_blocks): flush_current_chunk() current_chunk_blocks.append(b) else: # SOFT: # Split bei Hierarchie-Wechsel ODER wenn voll. if is_higher_hierarchy: flush_current_chunk() current_chunk_blocks.append(b) elif current_chunk_blocks and get_current_size() >= target: flush_current_chunk() current_chunk_blocks.append(b) else: current_chunk_blocks.append(b) else: # 2. Content Logic (Safety Trigger für Monster-Abschnitte) # Bevor wir den Block anhängen: Würde er das Fass zum Überlaufen bringen? # Wir nutzen hier 'max' als harte Grenze für den Safety-Split. current_size = get_current_size() block_size = estimate_tokens(b.text) if current_chunk_blocks and (current_size + block_size > max_limit): # NOTBREMSE: Chunk wird zu groß. # Wir splitten hier, auch wenn kein Header da ist. # Der Kontext (Section Title) bleibt erhalten, da er aus `current_h2` kommt (siehe parse_blocks). flush_current_chunk() current_chunk_blocks.append(b) else: current_chunk_blocks.append(b) # Letzten Rest flushen flush_current_chunk() return chunks # ========================================== # 4. ORCHESTRATION (ASYNC) # ========================================== async def assemble_chunks(note_id: str, md_text: str, note_type: str, config: Optional[Dict] = None) -> List[Chunk]: if config is None: config = get_chunk_config(note_type) fm, body_text = extract_frontmatter_from_text(md_text) note_status = fm.get("status", "").lower() primary_strategy = config.get("strategy", "sliding_window") enable_smart_edges = config.get("enable_smart_edge_allocation", False) if enable_smart_edges and note_status in ["draft", "initial_gen"]: logger.info(f"Chunker: Skipping Smart Edges for draft '{note_id}'.") enable_smart_edges = False blocks, doc_title = parse_blocks(md_text) if primary_strategy == "by_heading": chunks = await asyncio.to_thread(_strategy_by_heading, blocks, config, note_id, doc_title) else: chunks = await asyncio.to_thread(_strategy_sliding_window, blocks, config, note_id, doc_title) if not chunks: return [] if enable_smart_edges: chunks = await _run_smart_edge_allocation(chunks, md_text, note_id, note_type) for i, ch in enumerate(chunks): ch.neighbors_prev = chunks[i-1].id if i > 0 else None ch.neighbors_next = chunks[i+1].id if i < len(chunks)-1 else None return chunks def _extract_all_edges_from_md(md_text: str, note_id: str, note_type: str) -> List[str]: dummy_chunk = { "chunk_id": f"{note_id}#full", "text": md_text, "content": md_text, "window": md_text, "type": note_type } raw_edges = build_edges_for_note( note_id, [dummy_chunk], note_level_references=None, include_note_scope_refs=False ) all_candidates = set() for e in raw_edges: kind = e.get("kind") target = e.get("target_id") if target and kind not in ["belongs_to", "next", "prev", "backlink"]: all_candidates.add(f"{kind}:{target}") return list(all_candidates) async def _run_smart_edge_allocation(chunks: List[Chunk], full_text: str, note_id: str, note_type: str) -> List[Chunk]: analyzer = get_semantic_analyzer() candidate_list = _extract_all_edges_from_md(full_text, note_id, note_type) if not candidate_list: return chunks tasks = [] for chunk in chunks: tasks.append(analyzer.assign_edges_to_chunk(chunk.text, candidate_list, note_type)) results_per_chunk = await asyncio.gather(*tasks) assigned_edges_global = set() for i, confirmed_edges in enumerate(results_per_chunk): chunk = chunks[i] chunk.suggested_edges = confirmed_edges assigned_edges_global.update(confirmed_edges) if confirmed_edges: injection_str = "\n" + " ".join([f"[[rel:{e.split(':')[0]}|{e.split(':')[1]}]]" for e in confirmed_edges if ':' in e]) chunk.text += injection_str chunk.window += injection_str unassigned = set(candidate_list) - assigned_edges_global if unassigned: fallback_str = "\n" + " ".join([f"[[rel:{e.split(':')[0]}|{e.split(':')[1]}]]" for e in unassigned if ':' in e]) for chunk in chunks: chunk.text += fallback_str chunk.window += fallback_str if chunk.suggested_edges is None: chunk.suggested_edges = [] chunk.suggested_edges.extend(list(unassigned)) return chunks