store/chunking.py

"""Chunking — természetes vágási pontok.

A `prototype-agentic` mintát követjük (chunking.py:27-76):
  * SINGLE_CALL_THRESHOLD = 30000 char → nem darabolunk (egy LLM hívás elég)
  * default 15K char chunk + 500 char overlap
  * vágási preferencia: \\n\\n (bekezdés) > '. ' (mondat) > '\\n' > szóköz

A chunk-ok metadata-jában tárolódik a forrás dokumentum neve és a chunk_index.
"""

from __future__ import annotations

from typing import Any

from config import settings


def needs_chunking(text: str) -> bool:
    """Eldönti, hogy a szöveg hosszabb-e mint a SINGLE_CALL_THRESHOLD."""
    return len(text or "") > settings.single_call_threshold


def chunk_text(
    text: str,
    max_chars: int | None = None,
    overlap: int | None = None,
) -> list[str]:
    """Egy szöveget chunk-okra darabol természetes vágási pontoknál.

    Default: settings.chunk_max_chars (15_000) + settings.chunk_overlap_chars (500).
    """
    max_chars = max_chars or settings.chunk_max_chars
    overlap = overlap or settings.chunk_overlap_chars

    if not text:
        return []

    if len(text) <= max_chars:
        return [text]

    chunks: list[str] = []
    pos = 0
    n = len(text)

    while pos < n:
        end = pos + max_chars
        if end >= n:
            chunks.append(text[pos:n])
            break

        # Természetes vágási pont keresése a [pos + max_chars - 500, pos + max_chars] tartományban
        cut = _find_natural_cut(text, pos + max_chars - overlap, end)
        chunks.append(text[pos:cut])
        # Overlap: a következő chunk a cut - overlap-tól kezdődik (ha értelmes)
        pos = max(cut - overlap, pos + 1)
        if pos >= n - 1:
            break

    return chunks


def _find_natural_cut(text: str, min_pos: int, max_pos: int) -> int:
    """A [min_pos, max_pos] tartományban keres egy természetes vágási pontot.

    Preferencia sorrend: bekezdés-vég, mondat-vég, sortörés, szóköz.
    Ha egyik sem talál → max_pos (kemény vágás).
    """
    window = text[min_pos:max_pos]
    if not window:
        return max_pos

    # 1. Bekezdés-vég: \n\n
    idx = window.rfind("\n\n")
    if idx >= 0:
        return min_pos + idx + 2

    # 2. Mondat-vég: '. ', '! ', '? '
    for marker in (". ", "! ", "? ", ".\n", "!\n", "?\n"):
        idx = window.rfind(marker)
        if idx >= 0:
            return min_pos + idx + len(marker)

    # 3. Sortörés
    idx = window.rfind("\n")
    if idx >= 0:
        return min_pos + idx + 1

    # 4. Szóköz
    idx = window.rfind(" ")
    if idx >= 0:
        return min_pos + idx + 1

    # 5. Kemény vágás
    return max_pos


def chunk_document(
    file_name: str,
    full_text: str,
    doc_type: str | None = None,
) -> list[dict[str, Any]]:
    """Dokumentumot chunk-listára bont, metadata-val a vector store-hoz.

    Returns:
        [{"text": str, "metadata": {"source": ..., "doc_type": ..., "chunk_index": int}}, ...]

    Ha a `full_text` < SINGLE_CALL_THRESHOLD, egyetlen chunk lesz.
    """
    chunks_text = chunk_text(full_text)
    return [
        {
            "text": chunk,
            "metadata": {
                "source": file_name,
                "doc_type": doc_type or "egyeb",
                "chunk_index": idx,
            },
        }
        for idx, chunk in enumerate(chunks_text)
    ]