33ec9b5bc5
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
143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
"""Vault scanning and index-note introspection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from . import frontmatter
|
|
from .models import Note
|
|
|
|
# Directories that are never vault content.
|
|
SKIP_DIRS = {
|
|
".git", ".trash", ".venv", ".obsidian", "templates",
|
|
"graphify-out", "graphify-out-temp", "__pycache__", "node_modules",
|
|
}
|
|
|
|
# Files that are never categorised.
|
|
SKIP_FILES = {"index.md"}
|
|
|
|
|
|
class Vault:
|
|
"""Read-only view of an Obsidian vault."""
|
|
|
|
def __init__(self, root: str | os.PathLike):
|
|
self.root = Path(root).expanduser().resolve()
|
|
if not self.root.is_dir():
|
|
raise FileNotFoundError(f"vault not found: {self.root}")
|
|
|
|
# ---------------------------------------------------------------- scanning
|
|
|
|
def _iter_md(self):
|
|
for dirpath, dirnames, filenames in os.walk(self.root):
|
|
dirnames[:] = sorted(d for d in dirnames if d not in SKIP_DIRS)
|
|
for fn in sorted(filenames):
|
|
if not fn.endswith(".md"):
|
|
continue
|
|
if fn in SKIP_FILES:
|
|
continue
|
|
yield Path(dirpath) / fn
|
|
|
|
def read_note(self, path: Path) -> Note | None:
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeDecodeError):
|
|
return None
|
|
|
|
fm, body, had = frontmatter.parse(text)
|
|
rel = str(path.relative_to(self.root))
|
|
|
|
def as_list(key: str) -> list[str]:
|
|
val = fm.get(key)
|
|
if isinstance(val, list):
|
|
return [str(x) for x in val if x is not None]
|
|
if val:
|
|
return [str(val)]
|
|
return []
|
|
|
|
return Note(
|
|
path=str(path),
|
|
rel_path=rel,
|
|
filename=path.name,
|
|
stem=path.stem,
|
|
title=fm.get("title"),
|
|
note_type=fm.get("type"),
|
|
status=fm.get("status"),
|
|
tags=as_list("tags"),
|
|
areas=as_list("areas"),
|
|
projects=as_list("projects"),
|
|
fm=fm,
|
|
has_frontmatter=had,
|
|
body=body,
|
|
)
|
|
|
|
def scan(self) -> list[Note]:
|
|
notes = []
|
|
for path in self._iter_md():
|
|
note = self.read_note(path)
|
|
if note is not None:
|
|
notes.append(note)
|
|
return notes
|
|
|
|
# ------------------------------------------------------------------ shape
|
|
|
|
def folders(self) -> list[str]:
|
|
"""All vault-relative folders that contain at least one note."""
|
|
found = set()
|
|
for path in self._iter_md():
|
|
rel = path.relative_to(self.root)
|
|
if len(rel.parts) > 1:
|
|
found.add(str(rel.parent))
|
|
return sorted(found)
|
|
|
|
def index_notes(self) -> list[Note]:
|
|
"""Notes whose type is `index` — the Table of Contents family."""
|
|
out = []
|
|
for path in self._iter_md():
|
|
note = self.read_note(path)
|
|
if note and (note.note_type or "").strip().lower() == "index":
|
|
out.append(note)
|
|
return out
|
|
|
|
def toc_sections(self) -> dict[str, list[str]]:
|
|
"""
|
|
Map each index note (by rel_path) to its `##` section headings.
|
|
|
|
Used to decide where a new entry should be inserted.
|
|
"""
|
|
result: dict[str, list[str]] = {}
|
|
for note in self.index_notes():
|
|
sections = [
|
|
m.group(1).strip()
|
|
for m in re.finditer(r"^##\s+(.+)$", note.body, re.MULTILINE)
|
|
]
|
|
result[note.rel_path] = sections
|
|
return result
|
|
|
|
def known_titles(self) -> dict[str, str]:
|
|
"""Map lowercased title/stem → rel_path, for duplicate detection."""
|
|
out: dict[str, str] = {}
|
|
for path in self._iter_md():
|
|
note = self.read_note(path)
|
|
if not note:
|
|
continue
|
|
out.setdefault(note.stem.lower(), note.rel_path)
|
|
if note.title:
|
|
out.setdefault(note.title.lower(), note.rel_path)
|
|
return out
|
|
|
|
def inbox_notes(self, inbox: str = "inbox") -> list[Note]:
|
|
"""Notes sitting in the inbox, awaiting sorting. Skips `_` helper files."""
|
|
folder = self.root / inbox
|
|
if not folder.is_dir():
|
|
return []
|
|
out = []
|
|
for path in sorted(folder.glob("*.md")):
|
|
if path.name.startswith("_"):
|
|
continue
|
|
note = self.read_note(path)
|
|
if note is not None:
|
|
out.append(note)
|
|
return out
|