245 lines
6.4 KiB
Python
245 lines
6.4 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
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
|
|
|
|
|
|
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):
|
|
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_frontmatter_text(abs_path: str):
|
|
# Always read as text for python-frontmatter
|
|
with open(abs_path, "r", encoding="utf-8", errors="replace") as f:
|
|
return frontmatter.loads(f.read())
|
|
|
|
|
|
def load_note_from_file(vault: Vault, abs_path: str) -> Note:
|
|
post = _load_frontmatter_text(abs_path)
|
|
meta = _normalize_metadata(post.metadata)
|
|
title = meta["title"]
|
|
slug = slugify(title)
|
|
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)
|
|
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 p in vault.iter_markdown_files():
|
|
try:
|
|
abs_path = p if os.path.isabs(p) else vault.abspath(p)
|
|
notes.append(load_note_from_file(vault, abs_path))
|
|
except Exception:
|
|
continue
|
|
notes.sort(key=lambda n: n.updated, reverse=True)
|
|
return notes
|
|
|
|
|
|
_hex_re = re.compile(r"[0-9a-fA-F]")
|
|
def _uuid_key(value: str) -> Optional[str]:
|
|
if value is None:
|
|
return None
|
|
s = "".join(_hex_re.findall(str(value)))
|
|
if len(s) == 32:
|
|
return s.lower()
|
|
try:
|
|
return uuid.UUID(str(value)).hex
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def load_note_by_id(note_id: str) -> Optional[Note]:
|
|
vault = get_vault()
|
|
target_key = _uuid_key(note_id)
|
|
if not target_key:
|
|
return None
|
|
|
|
for p in vault.iter_markdown_files():
|
|
try:
|
|
abs_path = p if os.path.isabs(p) else vault.abspath(p)
|
|
post = _load_frontmatter_text(abs_path)
|
|
raw_id = (post.metadata or {}).get("id")
|
|
cand_key = _uuid_key(raw_id)
|
|
if cand_key and cand_key == target_key:
|
|
return load_note_from_file(vault, abs_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
|
|
note.updated = now_iso_utc()
|
|
vault = get_vault()
|
|
save_note(vault, note)
|
|
return note |