Lars
8800160564
All checks were successful
Deploy mindnet to llm-node / deploy (push) Successful in 3s
105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
# 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
|