# note_payload.py """ Mindnet - Note Payload Builder Version: 1.4.3 Beschreibung: - Robust gegenüber alten/neuen Aufrufsignaturen (toleriert *args, **kwargs). - Liest Typ-Defaults aus ./config/config.yaml oder ./config/types.yaml. - Setzt in mindnet_notes u.a.: - retriever_weight (Frontmatter > Typ-Defaults > ENV > 1.0) - chunk_profile (Frontmatter > Typ-Defaults > ENV > "medium") - edge_defaults (Frontmatter > Typ-Defaults > []) - path, type, title, note_id, tags, created/modified/date (falls vorhanden) - Garantiert JSON-serialisierbare Payloads. """ from __future__ import annotations from typing import Any, Dict, Optional import os import json import pathlib import yaml def _as_dict(note: Any) -> Dict[str, Any]: if isinstance(note, dict): return note d: Dict[str, Any] = {} for attr in ( "id", "note_id", "title", "path", "frontmatter", "meta", "body", "text", "type", "created", "modified", "chunks", "tags", ): if hasattr(note, attr): d[attr] = getattr(note, attr) # manche Parser nutzen "metadata" statt "frontmatter" if "frontmatter" not in d and hasattr(note, "metadata"): d["frontmatter"] = getattr(note, "metadata") return d def _load_types_config(explicit: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: if isinstance(explicit, dict): return explicit for rel in ("config/config.yaml", "config/types.yaml"): p = pathlib.Path(rel) if p.exists(): with p.open("r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} # zulässig: {"types": {...}} oder direkt {...} if isinstance(data, dict) and "types" in data and isinstance(data["types"], dict): return data["types"] return data if isinstance(data, dict) else {} return {} def _get_front(n: Dict[str, Any]) -> Dict[str, Any]: fm = n.get("frontmatter") or n.get("meta") or {} return fm if isinstance(fm, dict) else {} def _coalesce(*vals): for v in vals: if v is not None: return v return None 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 list(x)] return [str(x)] def make_note_payload(note: Any, *args, **kwargs) -> Dict[str, Any]: """ Build JSON-serialisable payload for a Note. Accepts legacy extra args/kwargs (e.g. types_config, vault_root) without error. """ n = _as_dict(note) types_cfg = kwargs.get("types_config") or (args[0] if args else None) types_cfg = _load_types_config(types_cfg) fm = _get_front(n) note_type = str(fm.get("type") or n.get("type") or "note") cfg_for_type = types_cfg.get(note_type, {}) if isinstance(types_cfg, dict) else {} default_rw = _env_float("MINDNET_DEFAULT_RETRIEVER_WEIGHT", 1.0) retriever_weight = _coalesce( fm.get("retriever_weight"), cfg_for_type.get("retriever_weight"), default_rw, ) try: retriever_weight = float(retriever_weight) except Exception: retriever_weight = default_rw chunk_profile = _coalesce( fm.get("chunk_profile"), cfg_for_type.get("chunk_profile"), os.environ.get("MINDNET_DEFAULT_CHUNK_PROFILE", "medium"), ) if not isinstance(chunk_profile, str): chunk_profile = "medium" edge_defaults = _ensure_list( _coalesce(fm.get("edge_defaults"), cfg_for_type.get("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") if isinstance(path, pathlib.Path): path = str(path) payload: Dict[str, Any] = { "note_id": note_id, "title": title, "type": note_type, "path": path, "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) # Validierungs-RTT (stellt JSON-Serialisierbarkeit sicher) json.loads(json.dumps(payload, ensure_ascii=False)) return payload