""" note_payload.py — Mindnet payload helpers Version: 0.5.2 (generated 2025-11-08 21:03:48) Purpose: - Build a NOTE payload without dropping existing fields. - Resolve and inject: * retriever_weight * chunk_profile * edge_defaults Resolution order: 1) Frontmatter fields 2) Type defaults from a provided registry ('types' kwarg) OR YAML file (types_file kwarg). YAML formats supported: - root['types'][note_type]{{retriever_weight, chunk_profile, edge_defaults}} - root[note_type] is the type block directly 3) ENV MINDNET_DEFAULT_RETRIEVER_WEIGHT 4) Fallback 1.0 Notes: - Function signature tolerant: accepts **kwargs (e.g. vault_root, types_file, types, types_registry). - Does NOT attempt to create edges; it only exposes 'edge_defaults' in the NOTE payload for later stages. """ from __future__ import annotations from typing import Any, Dict, Optional, Mapping, Union import os from pathlib import Path try: import yaml # type: ignore except Exception: # pragma: no cover yaml = None # will skip YAML loading if unavailable # -------- helpers -------- def _coerce_mapping(obj: Any) -> Dict[str, Any]: if obj is None: return {{}} if isinstance(obj, dict): return dict(obj) # try common attributes out: Dict[str, Any] = {{}} for k in ("__dict__",): if hasattr(obj, k): out.update(getattr(obj, k)) # named attributes we often see for k in ("id","note_id","title","type","path","source_path","frontmatter"): if hasattr(obj, k) and k not in out: out[k] = getattr(obj, k) return out def _get_frontmatter(parsed: Mapping[str, Any]) -> Dict[str, Any]: fm = parsed.get("frontmatter") if isinstance(fm, dict): return dict(fm) return {{}} # tolerate notes without frontmatter def _load_types_from_yaml(types_file: Optional[Union[str, Path]]) -> Dict[str, Any]: if types_file is None: # try common defaults candidates = [ Path("config/types.yaml"), Path("config/types.yml"), Path("config.yaml"), Path("config.yml"), ] for p in candidates: if p.exists(): types_file = p break if types_file is None: return {{}} p = Path(types_file) if not p.exists() or yaml is None: return {{}} try: data = yaml.safe_load(p.read_text(encoding="utf-8")) if not isinstance(data, dict): return {{}} # support both shapes: {{types: {{concept: ...}}}} OR {{concept: ...}} if "types" in data and isinstance(data["types"], dict): return dict(data["types"]) return data except Exception: return {{}} def _resolve_type_defaults(note_type: Optional[str], types: Optional[Dict[str,Any]]) -> Dict[str, Any]: defaults = {{}} if not note_type or not types or not isinstance(types, dict): return defaults block = types.get(note_type) if isinstance(block, dict): defaults.update(block) return defaults def _to_float(val: Any, fallback: float) -> float: if val is None: return fallback try: return float(val) except Exception: return fallback def _first_nonempty(*vals): for v in vals: if v is not None: if isinstance(v, str) and v.strip() == "": continue return v return None # -------- main API -------- def make_note_payload(parsed_note: Any, **kwargs) -> Dict[str, Any]: parsed = _coerce_mapping(parsed_note) fm = _get_frontmatter(parsed) # external sources types_registry = kwargs.get("types") or kwargs.get("types_registry") types_from_yaml = _load_types_from_yaml(kwargs.get("types_file")) # registry wins over YAML if provided types_all: Dict[str, Any] = types_registry if isinstance(types_registry, dict) else types_from_yaml note_type: Optional[str] = _first_nonempty(parsed.get("type"), fm.get("type")) title: Optional[str] = _first_nonempty(parsed.get("title"), fm.get("title")) note_id: Optional[str] = _first_nonempty(parsed.get("note_id"), parsed.get("id"), fm.get("id")) type_defaults = _resolve_type_defaults(note_type, types_all) # --- resolve retriever_weight --- env_default = os.getenv("MINDNET_DEFAULT_RETRIEVER_WEIGHT") env_default_val = _to_float(env_default, 1.0) if env_default is not None else 1.0 effective_retriever_weight = _to_float( _first_nonempty( fm.get("retriever_weight"), type_defaults.get("retriever_weight"), env_default_val, 1.0, ), 1.0, ) # --- resolve chunk_profile --- effective_chunk_profile = _first_nonempty( fm.get("chunk_profile"), fm.get("profile"), type_defaults.get("chunk_profile"), os.getenv("MINDNET_DEFAULT_CHUNK_PROFILE"), ) # --- resolve edge_defaults (list[str]) --- edge_defaults = _first_nonempty( fm.get("edge_defaults"), type_defaults.get("edge_defaults"), ) if edge_defaults is None: edge_defaults = [] if isinstance(edge_defaults, str): # allow "a,b,c" edge_defaults = [s.strip() for s in edge_defaults.split(",") if s.strip()] elif not isinstance(edge_defaults, list): edge_defaults = [] # Start payload by preserving existing parsed keys (shallow copy); DO NOT drop fields payload: Dict[str, Any] = dict(parsed) # Ensure canonical top-level fields if note_id is not None: payload["id"] = note_id payload["note_id"] = note_id if title is not None: payload["title"] = title if note_type is not None: payload["type"] = note_type payload["retriever_weight"] = effective_retriever_weight if effective_chunk_profile is not None: payload["chunk_profile"] = effective_chunk_profile if edge_defaults: payload["edge_defaults"] = edge_defaults # keep frontmatter merged (without duplication) if "frontmatter" in payload and isinstance(payload["frontmatter"], dict): fm_out = dict(payload["frontmatter"]) fm_out.setdefault("type", note_type) fm_out["retriever_weight"] = effective_retriever_weight if effective_chunk_profile is not None: fm_out["chunk_profile"] = effective_chunk_profile if edge_defaults: fm_out["edge_defaults"] = edge_defaults payload["frontmatter"] = fm_out return payload