# note_payload.py from __future__ import annotations from typing import Any, Dict, Optional, Tuple import os, json, pathlib, 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","metadata","type","created","modified","date","tags"): if hasattr(note, attr): d[attr] = getattr(note, attr) # Normalisiere Frontmatter fm = d.get("frontmatter") or d.get("meta") or d.get("metadata") or {} d["frontmatter"] = fm if isinstance(fm, dict) else {} return d def _pick_args(*args, **kwargs) -> Tuple[Optional[str], Optional[Dict[str,Any]]]: path = kwargs.get("path") types_cfg = kwargs.get("types_config") # legacy positional: (path, types_config) oder (types_config, ...) for a in args: if path is None and isinstance(a, (str, pathlib.Path)): path = str(a) if types_cfg is None and isinstance(a, dict): types_cfg = a return path, types_cfg 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 {} 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 _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 x] return [str(x)] def make_note_payload(note: Any, *args, **kwargs) -> Dict[str, Any]: n = _as_dict(note) path_arg, types_cfg_explicit = _pick_args(*args, **kwargs) types_cfg = _load_types_config(types_cfg_explicit) fm = n.get("frontmatter") or {} 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")) chunk_profile = chunk_profile if isinstance(chunk_profile, str) else "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") or path_arg if isinstance(path, pathlib.Path): path = str(path) payload = { "note_id": note_id, "title": title, "type": note_type, "path": path or "", # immer vorhanden "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.loads(json.dumps(payload, ensure_ascii=False)) return payload