mindnet/app/core/chunk_payload.py

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

Kurzbeschreibung
----------------
Erzeugt Qdrant-Payloads für Text-Chunks einer Note. Jeder Chunk enthält
den tatsächlichen Text unter dem Schlüssel ``text``. Dadurch kann der
Exporter den vollständigen Body verlässlich aus Chunks rekonstruieren,
falls ``notes.payload.fulltext`` fehlt.

Wesentliche Features
--------------------
- Stabile, idempotente Payload-Struktur für Chunks
- Persistenter Chunk-Text (``text``)
- Extraktion von Wikilinks pro Chunk (``wikilinks`` & ``references``)
- Pfadübernahme (relativ zum Vault, wird vom Aufrufer geliefert)
- Bereinigung leerer Felder (keine ``None``/leere Collections im Payload)

Abhängigkeiten
--------------
- ``app.core.chunker.Chunk`` und ``assemble_chunks`` (nur für CLI-Demo)
- ``app.core.parser.extract_wikilinks`` und ``read_markdown`` (nur CLI-Demo)

Beispiele (CLI – Sichtprüfung)
------------------------------
    python3 -m app.core.chunk_payload --from-file ./vault/demo.md --print
"""

from __future__ import annotations
from typing import Dict, List, Any
import argparse
import json
import os

# WICHTIG: Keine direkten Selbst- oder Kreuz-Imports, die Zyklen erzeugen!
# Für die Kernfunktion reicht extract_wikilinks als Typ-unabhängige Hilfsfunktion.
try:
    from app.core.parser import extract_wikilinks  # nur Funktion, erzeugt keine Zyklen
except Exception:  # pragma: no cover
    from .parser import extract_wikilinks  # type: ignore

# Typ-Hinweis für Chunk (zur Laufzeit nicht notwendig)
try:
    from app.core.chunker import Chunk  # nur Typ-Annotation / Nutzung im CLI
except Exception:  # pragma: no cover
    class Chunk:  # type: ignore
        pass


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

def _drop_empty(d: Dict[str, Any]) -> Dict[str, Any]:
    """Entfernt leere/None-Felder aus einem Dict (für saubere Payloads)."""
    return {k: v for k, v in d.items() if v not in (None, [], {}, "")}


# ---------------------------------------------------------------------------
# Kernfunktion
# ---------------------------------------------------------------------------

def make_chunk_payloads(note_meta: Dict[str, Any], path: str, chunks: List[Chunk]) -> List[Dict[str, Any]]:
    """
    Baut Payloads für alle ``chunks`` der gegebenen Note.

    Parameters
    ----------
    note_meta : Dict[str, Any]
        Minimale Metadaten der Note (mind. ``id``, ``title``; optional ``type``,
        ``area``, ``project``, ``tags``, ``lang``).
    path : str
        Relativer Pfad der Note innerhalb des Vaults (z. B. "area/topic/file.md").
    chunks : List[Chunk]
        Liste vorsegmentierter Chunks (vgl. app.core.chunker.Chunk).

    Returns
    -------
    List[Dict[str, Any]]
        Payload-Objekte, bereit für Qdrant-Upserts.
    """
    res: List[Dict[str, Any]] = []
    for ch in chunks:
        # 'text' kann bei leerem Abschnitt None/'' sein → robust behandeln
        text_val = getattr(ch, "text", "") or ""
        wikilinks = extract_wikilinks(text_val)
        payload = {
            "id": getattr(ch, "id", None),
            "note_id": note_meta.get("id"),
            "note_title": note_meta.get("title"),
            "chunk_index": getattr(ch, "index", None),
            "char_start": getattr(ch, "char_start", None),
            "char_end": getattr(ch, "char_end", None),
            "token_count": getattr(ch, "token_count", None),
            "section_title": getattr(ch, "section_title", None),
            "section_path": getattr(ch, "section_path", None),
            "lang": note_meta.get("lang"),
            "wikilinks": wikilinks,
            "external_links": [],
            "references": [{"target_id": w, "kind": "wikilink"} for w in wikilinks],
            "neighbors": {
                "prev": getattr(ch, "neighbors_prev", None),
                "next": getattr(ch, "neighbors_next", None),
            },
            "path": path,          # vom Aufrufer relativ geliefert
            "text": text_val,      # WICHTIG für Export/Rekonstruktion
        }
        res.append(_drop_empty(payload))
    return res


# ---------------------------------------------------------------------------
# CLI – nur zur Sichtprüfung / Debug
# ---------------------------------------------------------------------------

def _cli() -> None:
    # CLI lädt hier *lazy* die schweren Abhängigkeiten,
    # damit die Modul-Initialisierung zirkularfrei bleibt.
    try:
        from app.core.parser import read_markdown
        from app.core.chunker import assemble_chunks
    except Exception:  # pragma: no cover
        from .parser import read_markdown  # type: ignore
        from .chunker import assemble_chunks  # type: ignore

    ap = argparse.ArgumentParser(description="Vorschau: Chunk-Payloads erzeugen und anzeigen")
    ap.add_argument("--from-file", dest="src", required=True, help="Pfad zu einer Markdown-Datei")
    ap.add_argument("--print", dest="do_print", action="store_true", help="Payload auf stdout ausgeben")
    args = ap.parse_args()

    note = read_markdown(args.src)
    note_meta = {
        "id": note.frontmatter.get("id") or note.frontmatter.get("note_id"),
        "title": note.frontmatter.get("title"),
        "type": note.frontmatter.get("type"),
        "area": note.frontmatter.get("area"),
        "project": note.frontmatter.get("project"),
        "tags": note.frontmatter.get("tags", []),
        "lang": note.frontmatter.get("lang"),
    }

    chunks = assemble_chunks(note.frontmatter.get("id"), note.body or "", note.frontmatter.get("type", "concept"))

    # Vault-Root heuristisch relativieren (nur für Demo)
    vault_root = os.path.dirname(os.path.dirname(args.src))  # heuristisch
    rel_path = os.path.relpath(args.src, vault_root).replace("\\", "/").lstrip("/")

    payloads = make_chunk_payloads(note_meta, rel_path, chunks)

    if args.do_print:
        print(json.dumps(payloads, ensure_ascii=False, indent=2))


if __name__ == "__main__":  # pragma: no cover
    _cli()