obsidian-sorter: automated note filing for an Obsidian vault
Classifies notes from an inbox folder and files them, with a human approval step. Zero third-party dependencies. - type mirrors folder (200 projects / 300 areas / 400 Personal Family) - TypeSafe JEV for semantic classification (~$0.0001 per note) - watcher writes a proposal INTO the note; the owner ticks a checkbox - systemd user service on .27, 10s poll - 15/15 functional tests
This commit is contained in:
213
sorter/frontmatter.py
Normal file
213
sorter/frontmatter.py
Normal file
@@ -0,0 +1,213 @@
|
||||
"""
|
||||
Minimal YAML frontmatter reader/writer.
|
||||
|
||||
Deliberately dependency-free: the vault uses a simple, uniform YAML subset
|
||||
(scalars, dash-lists, inline lists, quoted strings). NixOS hosts have no pip,
|
||||
so avoiding PyYAML makes deployment trivial.
|
||||
|
||||
Supports:
|
||||
key: value
|
||||
key: "quoted value"
|
||||
key: [a, b, c]
|
||||
key:
|
||||
- item
|
||||
- item
|
||||
|
||||
Does NOT support (not present in the vault): block scalars (| >), anchors,
|
||||
multi-document streams, nested maps beyond one level.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_DELIM = re.compile(r"^---\s*$", re.MULTILINE)
|
||||
|
||||
|
||||
def _strip_comment(value: str) -> str:
|
||||
"""Remove a trailing ' # comment', but not inside quotes."""
|
||||
if not value:
|
||||
return value
|
||||
out = []
|
||||
quote = None
|
||||
for i, ch in enumerate(value):
|
||||
if quote:
|
||||
out.append(ch)
|
||||
if ch == quote and (i == 0 or value[i - 1] != "\\"):
|
||||
quote = None
|
||||
continue
|
||||
if ch in ('"', "'"):
|
||||
quote = ch
|
||||
out.append(ch)
|
||||
continue
|
||||
if ch == "#" and (i == 0 or value[i - 1] in " \t"):
|
||||
break
|
||||
out.append(ch)
|
||||
return "".join(out).rstrip()
|
||||
|
||||
|
||||
def parse_scalar(raw: str) -> Any:
|
||||
"""Parse a single YAML scalar into a Python value."""
|
||||
if raw is None:
|
||||
return None
|
||||
v = _strip_comment(raw.strip())
|
||||
if v == "":
|
||||
return None
|
||||
|
||||
# Fully quoted string
|
||||
if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"):
|
||||
return v[1:-1]
|
||||
|
||||
# Inline list: [a, b, c]
|
||||
if v.startswith("[") and v.endswith("]"):
|
||||
inner = v[1:-1].strip()
|
||||
if not inner:
|
||||
return []
|
||||
return [parse_scalar(part) for part in _split_inline(inner)]
|
||||
|
||||
low = v.lower()
|
||||
if low in ("true", "yes"):
|
||||
return True
|
||||
if low in ("false", "no"):
|
||||
return False
|
||||
if low in ("null", "~"):
|
||||
return None
|
||||
|
||||
return v
|
||||
|
||||
|
||||
def _split_inline(inner: str) -> list[str]:
|
||||
"""Split an inline list body on commas, respecting quotes."""
|
||||
parts, buf, quote = [], [], None
|
||||
for ch in inner:
|
||||
if quote:
|
||||
buf.append(ch)
|
||||
if ch == quote:
|
||||
quote = None
|
||||
continue
|
||||
if ch in ('"', "'"):
|
||||
quote = ch
|
||||
buf.append(ch)
|
||||
continue
|
||||
if ch == ",":
|
||||
parts.append("".join(buf).strip())
|
||||
buf = []
|
||||
continue
|
||||
buf.append(ch)
|
||||
if buf:
|
||||
parts.append("".join(buf).strip())
|
||||
return [p for p in parts if p != ""]
|
||||
|
||||
|
||||
def parse(text: str) -> tuple[dict[str, Any], str, bool]:
|
||||
"""
|
||||
Split a note into (frontmatter_dict, body, had_frontmatter).
|
||||
|
||||
If the note has no valid frontmatter block, returns ({}, text, False).
|
||||
"""
|
||||
if not text.startswith("---"):
|
||||
return {}, text, False
|
||||
|
||||
m = re.match(r"^---\s*\n(.*?)\n---\s*\n?", text, re.DOTALL)
|
||||
if not m:
|
||||
return {}, text, False
|
||||
|
||||
fm_text = m.group(1)
|
||||
body = text[m.end():]
|
||||
return parse_block(fm_text), body, True
|
||||
|
||||
|
||||
def parse_block(text: str) -> dict[str, Any]:
|
||||
"""Parse the body of a frontmatter block into a dict."""
|
||||
result: dict[str, Any] = {}
|
||||
key: str | None = None
|
||||
|
||||
for raw in text.split("\n"):
|
||||
if not raw.strip() or raw.lstrip().startswith("#"):
|
||||
continue
|
||||
|
||||
# List item belonging to the current key
|
||||
item = re.match(r"^\s+-\s*(.*)$", raw) or re.match(r"^-\s*(.*)$", raw)
|
||||
if item and key is not None:
|
||||
if not isinstance(result.get(key), list):
|
||||
result[key] = []
|
||||
result[key].append(parse_scalar(item.group(1)))
|
||||
continue
|
||||
|
||||
# Indented continuation of a scalar value
|
||||
if re.match(r"^\s+\S", raw) and key is not None:
|
||||
prev = result.get(key)
|
||||
extra = parse_scalar(raw.strip())
|
||||
if isinstance(prev, list):
|
||||
prev.append(extra)
|
||||
elif prev in (None, ""):
|
||||
result[key] = extra
|
||||
else:
|
||||
result[key] = f"{prev} {extra}"
|
||||
continue
|
||||
|
||||
# key: value
|
||||
kv = re.match(r"^([A-Za-z0-9_\-]+):\s?(.*)$", raw)
|
||||
if kv:
|
||||
key = kv.group(1)
|
||||
value = kv.group(2)
|
||||
result[key] = None if value.strip() == "" else parse_scalar(value)
|
||||
continue
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def dump_block(fm: dict[str, Any], order: list[str] | None = None) -> str:
|
||||
"""Serialise a frontmatter dict back to YAML text (no --- delimiters)."""
|
||||
lines: list[str] = []
|
||||
keys = list(order or [])
|
||||
keys += [k for k in fm if k not in keys]
|
||||
|
||||
for k in keys:
|
||||
if k not in fm:
|
||||
continue
|
||||
v = fm[k]
|
||||
lines.extend(_dump_key(k, v))
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _dump_key(key: str, value: Any) -> list[str]:
|
||||
if value is None:
|
||||
return [f"{key}:"]
|
||||
if isinstance(value, list):
|
||||
if not value:
|
||||
return [f"{key}: []"]
|
||||
out = [f"{key}:"]
|
||||
for item in value:
|
||||
out.append(f" - {_quote(item)}")
|
||||
return out
|
||||
if isinstance(value, bool):
|
||||
return [f"{key}: {'true' if value else 'false'}"]
|
||||
return [f"{key}: {_quote(value)}"]
|
||||
|
||||
|
||||
def _quote(value: Any) -> str:
|
||||
"""Quote a scalar when required for valid YAML."""
|
||||
if isinstance(value, (int, float)):
|
||||
return str(value)
|
||||
s = str(value)
|
||||
if s == "":
|
||||
return '""'
|
||||
needs = (
|
||||
s != s.strip()
|
||||
or s.lower() in ("true", "false", "yes", "no", "null", "~")
|
||||
or any(c in s for c in ":#[]{}&*!|>%@`\"'")
|
||||
or re.match(r"^\d{4}-\d{2}-\d{2}( \d{2}:\d{2})?$", s)
|
||||
)
|
||||
if needs:
|
||||
escaped = s.replace("\\", "\\\\").replace('"', '\\"')
|
||||
return f'"{escaped}"'
|
||||
return s
|
||||
|
||||
|
||||
def render(fm: dict[str, Any], body: str, order: list[str] | None = None) -> str:
|
||||
"""Render a complete note (frontmatter block + body)."""
|
||||
block = dump_block(fm, order=order)
|
||||
body = body.lstrip("\n")
|
||||
return f"---\n{block}\n---\n\n{body}"
|
||||
Reference in New Issue
Block a user