mindnet/app/core/note_payload.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Modul: app/core/note_payload.py
Version: 2.0.0

Zweck
-----
Erzeugt ein robustes Note-Payload. Werte wie `retriever_weight`, `chunk_profile`
und `edge_defaults` werden in folgender Priorität bestimmt:
1) Frontmatter (Note)
2) Typ-Registry (config/types.yaml: types.<type>.*)
3) Registry-Defaults (config/types.yaml: defaults.*)
4) ENV-Defaults (MINDNET_DEFAULT_RETRIEVER_WEIGHT / MINDNET_DEFAULT_CHUNK_PROFILE)
"""
from __future__ import annotations

from typing import Any, Dict, Tuple, Optional
import os, json, pathlib

try:
    import yaml
except Exception:
    yaml = None

def _as_dict(x) -> Dict[str, Any]:
    if isinstance(x, dict):
        return dict(x)
    try:
        return dict(x or {})
    except Exception:
        return {"raw": str(x)}

def _pick_args(*args, **kwargs) -> Tuple[Optional[str], Optional[dict]]:
    path = kwargs.get("path") or (args[0] if args else None)
    types_cfg = kwargs.get("types_cfg") or kwargs.get("types") or None
    return path, types_cfg

def _env_float(name: str, default: float) -> float:
    try:
        return float(os.environ.get(name, default))
    except Exception:
        return default

def _ensure_list(x) -> list:
    if x is None: return []
    if isinstance(x, list): return [str(i) for i in x]
    if isinstance(x, (set, tuple)): return [str(i) for i in x]
    return [str(x)]

# ---- Registry laden --------------------------------------------------------

def _load_types_config(explicit_cfg: Optional[dict] = None) -> dict:
    if explicit_cfg and isinstance(explicit_cfg, dict):
        return explicit_cfg
    path = os.getenv("MINDNET_TYPES_FILE") or "./config/types.yaml"
    if not os.path.isfile(path) or yaml is None:
        return {}
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}

def _cfg_for_type(note_type: str, reg: dict) -> dict:
    if not isinstance(reg, dict):
        return {}
    types = reg.get("types") if isinstance(reg.get("types"), dict) else reg
    return types.get(note_type, {}) if isinstance(types, dict) else {}

def _cfg_defaults(reg: dict) -> dict:
    if not isinstance(reg, dict):
        return {}
    for key in ("defaults", "default", "global"):
        v = reg.get(key)
        if isinstance(v, dict):
            return v
    return {}

# ---- Haupt-API -------------------------------------------------------------

def make_note_payload(note: Any, *args, **kwargs) -> Dict[str, Any]:
    n = _as_dict(note)
    path_arg, types_cfg_explicit = _pick_args(*args, **kwargs)
    reg = _load_types_config(types_cfg_explicit)

    fm = n.get("frontmatter") or {}
    fm_type = fm.get("type") or n.get("type") or "concept"
    note_type = str(fm_type)

    cfg_type = _cfg_for_type(note_type, reg)
    cfg_def  = _cfg_defaults(reg)

    default_rw = _env_float("MINDNET_DEFAULT_RETRIEVER_WEIGHT", 1.0)
    retriever_weight = fm.get("retriever_weight")
    if retriever_weight is None:
        retriever_weight = cfg_type.get("retriever_weight", cfg_def.get("retriever_weight", default_rw))
    try:
        retriever_weight = float(retriever_weight)
    except Exception:
        retriever_weight = default_rw

    chunk_profile = fm.get("chunk_profile")
    if chunk_profile is None:
        chunk_profile = cfg_type.get("chunk_profile", cfg_def.get("chunk_profile", os.environ.get("MINDNET_DEFAULT_CHUNK_PROFILE", "medium")))
    if not isinstance(chunk_profile, str):
        chunk_profile = "medium"

    edge_defaults = fm.get("edge_defaults")
    if edge_defaults is None:
        edge_defaults = cfg_type.get("edge_defaults", cfg_def.get("edge_defaults", []))
    edge_defaults = _ensure_list(edge_defaults)

    note_id = n.get("note_id") or n.get("id") or fm.get("id")
    title = n.get("title") or fm.get("title") or ""
    path = n.get("path") or path_arg
    if isinstance(path, pathlib.Path):
        path = str(path)

    payload = {
        "note_id": note_id,
        "title": title,
        "type": note_type,
        "path": path or "",
        "retriever_weight": retriever_weight,
        "chunk_profile": chunk_profile,
        "edge_defaults": edge_defaults,
    }

    tags = fm.get("tags") or fm.get("keywords") or n.get("tags")
    if tags: payload["tags"] = _ensure_list(tags)
    for k in ("created","modified","date"):
        v = fm.get(k) or n.get(k)
        if v: payload[k] = str(v)

    # JSON-Roundtrip zur harten Validierung (ASCII beibehalten)
    json.loads(json.dumps(payload, ensure_ascii=False))
    return payload