from __future__ import annotations import json import os import uuid from dataclasses import dataclass, field from typing import List, Optional, Tuple import frontmatter from flask import current_app from app.services.vault import Vault from app.utils.paths import atomic_write, ensure_unique_path from app.utils.slugs import slugify from app.utils.time import now_iso_utc, is_iso_utc REQUIRED_FIELDS = ("id", "title", "created", "updated") OPTIONAL_LIST_FIELDS = ("aliases", "tags") @dataclass class Note: id: str title: str created: str updated: str slug: str rel_path: str # relative path from vault root, e.g. notes/foo.md tags: List[str] = field(default_factory=list) aliases: List[str] = field(default_factory=list) status: Optional[str] = None summary: Optional[str] = None body: str = "" def _normalize_metadata(meta: dict) -> dict: meta = dict(meta or {}) # Lists for key in OPTIONAL_LIST_FIELDS: val = meta.get(key, []) if val is None: val = [] if isinstance(val, str): # Allow comma-separated string as input val = [x.strip() for x in val.split(",") if x.strip()] if not isinstance(val, list): val = [str(val)] meta[key] = val # Strings for key in ("title", "status", "summary"): if key in meta and meta[key] is not None: meta[key] = str(meta[key]) # UUID if not meta.get("id"): meta["id"] = str(uuid.uuid4()) else: try: meta["id"] = str(uuid.UUID(str(meta["id"]))) except Exception: meta["id"] = str(uuid.uuid4()) # Timestamps created = meta.get("created") updated = meta.get("updated") if not created or not is_iso_utc(str(created)): created = now_iso_utc() if not updated or not is_iso_utc(str(updated)): updated = created meta["created"] = str(created) meta["updated"] = str(updated) # Title fallback if not meta.get("title"): meta["title"] = "Untitled" return meta def _frontmatter_from_note(note: Note) -> dict: data = { "id": note.id, "title": note.title, "created": note.created, "updated": note.updated, } if note.tags: data["tags"] = list(note.tags) if note.aliases: data["aliases"] = list(note.aliases) if note.status: data["status"] = note.status if note.summary: data["summary"] = note.summary return data def get_vault() -> Vault: vault_path = current_app.config.get("KB_VAULT_PATH") if not vault_path: raise RuntimeError("Vault path is not configured (KB_VAULT_PATH).") v = Vault(vault_path) v.ensure_structure() return v def note_path_for_slug(vault: Vault, slug: str) -> Tuple[str, str]: rel = os.path.join("notes", f"{slug}.md") abs_path = os.path.join(vault.paths.root, rel) return rel, abs_path def load_note_from_file(vault: Vault, abs_path: str) -> Note: with open(abs_path, "rb") as f: post = frontmatter.load(f) meta = _normalize_metadata(post.metadata) title = meta["title"] slug = slugify(title) rel, _ = note_path_for_slug(vault, slug) # Use actual rel path of the file (could differ from current slug if renamed later) rel = vault.relpath(abs_path) note = Note( id=meta["id"], title=title, created=meta["created"], updated=meta["updated"], slug=slug, rel_path=rel, tags=meta.get("tags", []), aliases=meta.get("aliases", []), status=meta.get("status"), summary=meta.get("summary"), body=post.content or "", ) return note def create_note( title: str, body: str = "", tags: Optional[List[str]] = None, aliases: Optional[List[str]] = None, status: Optional[str] = None, summary: Optional[str] = None, ) -> Note: vault = get_vault() meta = _normalize_metadata( { "title": title, "tags": tags or [], "aliases": aliases or [], "status": status, "summary": summary, } ) slug = slugify(meta["title"]) rel, abs_path = note_path_for_slug(vault, slug) abs_path = ensure_unique_path(abs_path) # If collision caused a suffix, recompute rel path accordingly rel = vault.relpath(abs_path) note = Note( id=meta["id"], title=meta["title"], created=meta["created"], updated=meta["updated"], slug=slug, rel_path=rel, tags=meta.get("tags", []), aliases=meta.get("aliases", []), status=meta.get("status"), summary=meta.get("summary"), body=body or "", ) save_note(vault, note) return note def save_note(vault: Vault, note: Note) -> None: data = _frontmatter_from_note(note) post = frontmatter.Post(note.body, **data) text = frontmatter.dumps(post) atomic_write(vault.abspath(note.rel_path), text.encode("utf-8")) def list_notes() -> list[Note]: vault = get_vault() notes: list[Note] = [] for path in vault.iter_markdown_files(): try: notes.append(load_note_from_file(vault, path)) except Exception: # Ignore malformed files for now (could log) continue # Sort by updated desc notes.sort(key=lambda n: n.updated, reverse=True) return notes def load_note_by_id(note_id: str) -> Optional[Note]: vault = get_vault() for path in vault.iter_markdown_files(): try: with open(path, "rb") as f: post = frontmatter.load(f) meta = _normalize_metadata(post.metadata) if str(meta.get("id")) == str(note_id): # Load fully to construct Note with body and rel path return load_note_from_file(vault, path) except Exception: continue return None def update_note_body(note_id: str, new_body: str) -> Optional[Note]: note = load_note_by_id(note_id) if not note: return None note.body = new_body # Update timestamp note.updated = now_iso_utc() vault = get_vault() save_note(vault, note) return note