""" note_payload.py Version: 1.4.2 Description: Builds the payload for a *note* document destined for the Qdrant "notes" collection. - Defensive against both dict-like and attribute-like "ParsedNote" inputs. - Accepts extra/legacy arguments via *args / **kwargs (e.g., vault_root, type_defaults). - Ensures "retriever_weight" is always present in the payload (float), resolved as: kwarg retriever_weight > frontmatter.retriever_weight > frontmatter.retriever.weight > type_defaults[type].retriever_weight > 1.0 - Preserves common metadata fields expected downstream. Public API: make_note_payload(parsed_note, *_, retriever_weight=None, vault_root=None, type_defaults=None, **__) """ from typing import Any, Dict, Optional def _as_dict(obj: Any) -> Dict[str, Any]: if isinstance(obj, dict): return obj # Try common attribute names to build a dict view out = {} for key in ( "frontmatter", "fm", "meta", "note_id", "id", "title", "type", "tags", "aliases", "created", "updated", "date", "abs_path", "path", "rel_path", ): if hasattr(obj, key): out[key] = getattr(obj, key) return out def _get(obj: Any, *keys: str, default: Any=None): """Get first existing key/attribute from obj.""" if isinstance(obj, dict): for k in keys: if k in obj: return obj[k] return default # attribute access for k in keys: if hasattr(obj, k): return getattr(obj, k) return default def _as_list(val): if val is None: return None if isinstance(val, (list, tuple)): return list(val) if isinstance(val, str): return [val] try: return list(val) # best-effort except Exception: return [val] def _coerce_float(val: Any, default: float) -> float: if val is None: return float(default) try: return float(val) except Exception: return float(default) def _clean(d: Dict[str, Any]) -> Dict[str, Any]: return {k: v for k, v in d.items() if v is not None} def make_note_payload( parsed_note: Any, *_, # ignore legacy extra positional args for backward compatibility retriever_weight: Optional[float] = None, vault_root: Optional[str] = None, type_defaults: Optional[Dict[str, Dict[str, Any]]] = None, **__, # ignore any unexpected kwargs ) -> Dict[str, Any]: nd = _as_dict(parsed_note) fm = _get(nd, "frontmatter", "fm", "meta", default={}) or {} note_id = _get(nd, "note_id", "id") or fm.get("id") title = _get(nd, "title") or fm.get("title") ntype = _get(nd, "type") or fm.get("type") or "concept" # Path handling abs_path = _get(nd, "abs_path", "path") rel_path = _get(nd, "rel_path") if vault_root and abs_path and not rel_path: try: from pathlib import Path rel_path = str(Path(abs_path).resolve().relative_to(Path(vault_root).resolve())) except Exception: rel_path = _get(nd, "path") or abs_path # Tags / aliases tags = _as_list(_get(nd, "tags") or fm.get("tags")) aliases = _as_list(_get(nd, "aliases") or fm.get("aliases")) # Created/Updated created = _get(nd, "created", "date") or fm.get("created") or fm.get("date") updated = _get(nd, "updated") or fm.get("updated") # Chunk profile (effective) chunk_profile = fm.get("chunk_profile") if not chunk_profile and type_defaults and ntype in type_defaults: chunk_profile = type_defaults[ntype].get("chunk_profile") # Retriever weight resolution (ensures it is present) if retriever_weight is None: retriever_weight = ( fm.get("retriever_weight") or (fm.get("retriever", {}) or {}).get("weight") ) if retriever_weight is None and type_defaults and ntype in type_defaults: retriever_weight = type_defaults[ntype].get("retriever_weight") retriever_weight = _coerce_float(retriever_weight, default=1.0) payload = _clean({ "note_id": note_id, "id": note_id, # keep both, many downstream tools expect 'id' "title": title, "type": ntype, "tags": tags, "aliases": aliases, "created": created, "updated": updated, "path": rel_path or abs_path, "chunk_profile": chunk_profile, "retriever_weight": retriever_weight, }) return payload