# app/core/note_payload.py # ----------------------------------------------------------------------------- # Name: note_payload.py # Version: 1.2.1 (2025-09-08) # Zweck: Erzeugt den Qdrant-Payload für Notes, inkl. deterministischer # Hash-Bildung zur Idempotenz-Erkennung. # # Änderungen: # 1.2.1: Akzeptiert jetzt sowohl dict-Input als auch Objekt-Input (z. B. ParsedNote) # mit Attributen .frontmatter, .body, .path. Dadurch kein AttributeError mehr. # 1.2.0: Konfigurierbare Hash-Strategie via ENV MINDNET_HASH_MODE # ('body' | 'body+frontmatter' | 'frontmatter'); kanonische FM-Serialisierung. # # Steuerung Hash-Strategie (unverändert): # export MINDNET_HASH_MODE=body+frontmatter # MINDNET_HASH_MODE=frontmatter python3 -m scripts.import_markdown --vault ./vault --apply # # Hinweis: # - Datei-Zeitstempel (mtime/ctime) werden NICHT verwendet. # - Default-Strategie bleibt 'body' (rückwärtskompatibel). # ----------------------------------------------------------------------------- from __future__ import annotations import hashlib import json import os from typing import Any, Dict, Optional # -----------------------------------------------------------------------------# # Dienstfunktionen # -----------------------------------------------------------------------------# def sha256_text(s: str) -> str: """Bildet SHA-256 über den gegebenen Unicode-String (UTF-8).""" return hashlib.sha256(s.encode("utf-8")).hexdigest() def canonicalize_frontmatter(fm: Dict[str, Any]) -> str: """ Serialisiert das Frontmatter deterministisch: - JSON mit sortierten Keys - kompakte Separatoren - UTF-8, keine ASCII-Escapes Achtung: Datumswerte müssen Strings sein (siehe Schema). """ return json.dumps( fm or {}, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) def get_hash_mode_from_env() -> str: """ Liest die Hash-Strategie aus ENV MINDNET_HASH_MODE. Zulässig: 'body' (Default), 'body+frontmatter', 'frontmatter' """ val = (os.environ.get("MINDNET_HASH_MODE") or "").strip().lower() if val in ("body", "body+frontmatter", "frontmatter"): return val return "body" def compute_hash(body: str, frontmatter: Dict[str, Any], mode: Optional[str] = None) -> str: """ Berechnet den Hash gemäß Strategie. - 'body': nur Body - 'body+frontmatter': Body + FM (kanonisch) - 'frontmatter': nur FM (kanonisch) """ strategy = (mode or get_hash_mode_from_env()).lower() body_str = (body or "").strip() fm_str = canonicalize_frontmatter(frontmatter or {}) if strategy == "frontmatter": return sha256_text(fm_str) if strategy == "body+frontmatter": combo = body_str + "\n\n---\n\n" + fm_str return sha256_text(combo) # Default / 'body' return sha256_text(body_str) # -----------------------------------------------------------------------------# # Helfer: parsed -> (frontmatter, body, path) # -----------------------------------------------------------------------------# def _extract_parsed(parsed: Any) -> tuple[Dict[str, Any], str, Optional[str]]: """ Erlaubt sowohl dict- als auch objektbasierte Parser-Ergebnisse. Erwartet mindestens 'frontmatter' + 'body'. 'path' ist optional. """ # dict-Eingang if isinstance(parsed, dict): fm = dict(parsed.get("frontmatter") or {}) body = parsed.get("body") or "" path = parsed.get("path") return fm, body, path # objektbasierter Eingang (z. B. ParsedNote) # Erwartete Attribute: .frontmatter (dict), .body (str), optional .path fm = {} if hasattr(parsed, "frontmatter"): fm_val = getattr(parsed, "frontmatter") if isinstance(fm_val, dict): fm = dict(fm_val) else: # Notfalls in ein dict konvertieren, falls FM ein pydantic/BaseModel ist try: fm = dict(fm_val) # type: ignore[arg-type] except Exception: # finaler Fallback: JSON roundtrip fm = json.loads(json.dumps(fm_val, default=getattr(fm_val, "dict", None))) body = getattr(parsed, "body", "") or "" path = getattr(parsed, "path", None) return fm, body, path # -----------------------------------------------------------------------------# # Hauptfunktion für Note-Payload # -----------------------------------------------------------------------------# def make_note_payload(parsed: Any, vault_root: Optional[str] = None) -> Dict[str, Any]: """ Baut den Payload für eine Note auf Basis der geparsten Markdown-Datei. parsed: dict ODER Objekt mit Attributen .frontmatter, .body, optional .path Rückgabe-Payload (kompatibel mit mindnet_notes Schema): { "note_id": "...", "title": "...", "type": "...", "status": "...", "created": "...", "updated": "...", "path": "...", # falls vorhanden "tags": [...], # optional "hash_fulltext": "sha256...", ... } """ fm, body, path = _extract_parsed(parsed) # Hash nach konfigurierter Strategie berechnen hash_fulltext = compute_hash(body=body, frontmatter=fm, mode=None) payload: Dict[str, Any] = { "note_id": fm.get("id") or fm.get("note_id"), "title": fm.get("title"), "type": fm.get("type"), "status": fm.get("status"), "created": fm.get("created"), "updated": fm.get("updated"), "path": path or fm.get("path"), "tags": fm.get("tags"), "hash_fulltext": hash_fulltext, } # Bekannte optionale FM-Felder transparent durchreichen (ohne Hash-Einfluss) passthrough_keys = [ "area", "project", "source", "lang", "slug", ] for k in passthrough_keys: if k in fm: payload[k] = fm[k] return payload # -----------------------------------------------------------------------------# # Optional: Self-Test # -----------------------------------------------------------------------------# if __name__ == "__main__": class _PN: def __init__(self): self.frontmatter = { "id": "demo-123", "title": "Demo", "type": "note", "status": "active", "created": "2025-09-08T10:00:00+00:00", "updated": "2025-09-08T10:00:00+00:00", "tags": ["demo", "test"], } body = "# Überschrift\n\nText." path = "demo.md" parsed_dict = { "frontmatter": { "id": "demo-456", "title": "Demo2", "type": "note", "status": "active", "created": "2025-09-08T10:00:00+00:00", "updated": "2025-09-08T10:00:00+00:00", }, "body": "Text2", "path": "demo2.md", } for mode in ("body", "body+frontmatter", "frontmatter"): os.environ["MINDNET_HASH_MODE"] = mode print(f"\n-- MODE={mode}") print(make_note_payload(_PN())) print(make_note_payload(parsed_dict))