286 lines
10 KiB
Python
286 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
vault_lib.py — shared library for Ryan's R/J/F/I Obsidian vault Librarian tooling.
|
|
|
|
Single source of truth for:
|
|
- tree walking + ID parsing (SYS.AC.ID)
|
|
- wikilink extraction / broken-link detection
|
|
- the topology manifest (machine-readable JSON block in a 00-area note)
|
|
- cross-tree (/vault <-> /data) topology diff
|
|
|
|
Used by:
|
|
- vault (the CLI: search / rename / create / move IDs across both trees)
|
|
- vault-librarian.py (the nightly deterministic report generator)
|
|
|
|
Conventions (see ryan-vault-rjfi-migration + obsidian-vault-navigation skills):
|
|
VAULT = /vault -> the JDex (markdown map). Obsidian-synced. Authoritative for structure.
|
|
DATA = /data -> blob/artifact territory. SMB-served. Mirrors the SYS.AC skeleton,
|
|
MINUS `0 System area`, PLUS `_unsorted/` (allowed, no vault counterpart).
|
|
Every level is prefixed: `R Ryan / R.1 Life admin / R.11 Identity... / R.11.84 ... / R.11.84 ....md`
|
|
Folder-note = a .md whose name == its parent folder name (Folder Notes plugin needs this).
|
|
"""
|
|
import os
|
|
import re
|
|
import json
|
|
|
|
VAULT = "/vault"
|
|
DATA = "/data"
|
|
|
|
# Directories never treated as vault structure.
|
|
SKIP_DIRS = {".git", ".obsidian"}
|
|
# Top-level vault dir that is meta, not a system (unprefixed).
|
|
SYSTEM_META = "0 System area"
|
|
# /data dirs allowed to exist without a /vault counterpart.
|
|
DATA_EXEMPT = {"_unsorted"}
|
|
|
|
# --- regexes on folder/leaf basenames ---
|
|
SYS_RE = re.compile(r'^([RJFI]) ') # "R Ryan"
|
|
AREA_RE = re.compile(r'^([RJFI])\.(\d) ') # "R.1 Life admin"
|
|
CAT_RE = re.compile(r'^([RJFI])\.(\d\d) ') # "R.11 Identity & vital records"
|
|
ID_RE = re.compile(r'^([RJFI])\.(\d\d)\.(\d\d) ') # "R.11.84 My journal & notes by day"
|
|
# extend-notation leaf that never got a real ID, e.g. "25+ Smoker Brisket.md"
|
|
EXTEND_RE = re.compile(r'^(\d\d)\+ ')
|
|
|
|
WIKILINK_RE = re.compile(r'!?\[\[([^\]]+)\]\]')
|
|
|
|
|
|
def parse_id(basename):
|
|
"""Return (sys, area_digit, cat, id) or None. basename is a folder/leaf name."""
|
|
m = ID_RE.match(basename)
|
|
if not m:
|
|
return None
|
|
return {"sys": m.group(1), "area": m.group(2), "cat": m.group(2) + m.group(3)[:0], "raw": basename}
|
|
|
|
|
|
def id_code(basename):
|
|
"""Return the 'R.11.84' code for a leaf name, or None."""
|
|
m = ID_RE.match(basename)
|
|
if not m:
|
|
return None
|
|
return f"{m.group(1)}.{m.group(2)}.{m.group(3)}"
|
|
|
|
|
|
def cat_code(basename):
|
|
m = CAT_RE.match(basename)
|
|
return f"{m.group(1)}.{m.group(2)}" if m else None
|
|
|
|
|
|
def area_code(basename):
|
|
m = AREA_RE.match(basename)
|
|
return f"{m.group(1)}.{m.group(2)}" if m else None
|
|
|
|
|
|
def walk_structure(root, include_meta=False):
|
|
"""
|
|
Walk a vault-shaped tree. Return dict:
|
|
systems: {name}
|
|
areas: {code -> relpath}
|
|
cats: {code -> relpath}
|
|
ids: {code -> relpath} (folder relpaths)
|
|
Skips .git/.obsidian. If include_meta False, skips the `0 System area` subtree.
|
|
"""
|
|
systems, areas, cats, ids = {}, {}, {}, {}
|
|
for dirpath, dirnames, filenames in os.walk(root):
|
|
dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
|
|
rel = os.path.relpath(dirpath, root)
|
|
top = rel.split(os.sep)[0]
|
|
if not include_meta and top == SYSTEM_META:
|
|
dirnames[:] = []
|
|
continue
|
|
base = os.path.basename(dirpath)
|
|
if ID_RE.match(base):
|
|
ids[id_code(base)] = rel
|
|
elif CAT_RE.match(base):
|
|
cats[cat_code(base)] = rel
|
|
elif AREA_RE.match(base):
|
|
areas[area_code(base)] = rel
|
|
elif SYS_RE.match(base) and os.sep not in rel:
|
|
systems[base[0]] = rel
|
|
return {"systems": systems, "areas": areas, "cats": cats, "ids": ids}
|
|
|
|
|
|
ARCHIVE_SUBPATH = os.path.join(SYSTEM_META, "09 System Archive")
|
|
|
|
|
|
def is_archived(relpath):
|
|
"""True if a vault-relative path is inside the frozen pre-RJFI archive."""
|
|
return relpath.startswith(ARCHIVE_SUBPATH)
|
|
|
|
|
|
def all_notes(root, include_meta=True, include_archive=True):
|
|
"""Yield absolute paths of every .md note (skips .git/.obsidian).
|
|
include_archive=False skips the frozen `09 Archive of pre-RJFI tree` subtree."""
|
|
for dirpath, dirnames, filenames in os.walk(root):
|
|
dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
|
|
rel = os.path.relpath(dirpath, root)
|
|
if not include_archive and is_archived(rel):
|
|
dirnames[:] = []
|
|
continue
|
|
for f in filenames:
|
|
if f.endswith(".md"):
|
|
yield os.path.join(dirpath, f)
|
|
|
|
|
|
def is_folder_note(path):
|
|
"""A note whose filename stem == its parent directory name."""
|
|
stem = os.path.splitext(os.path.basename(path))[0]
|
|
parent = os.path.basename(os.path.dirname(path))
|
|
return stem == parent
|
|
|
|
|
|
def build_title_index(root):
|
|
"""
|
|
Map every resolvable wikilink target -> abs path.
|
|
Obsidian resolves by basename (title) AND by full relative path.
|
|
"""
|
|
idx = {}
|
|
for p in all_notes(root):
|
|
stem = os.path.splitext(os.path.basename(p))[0]
|
|
rel = os.path.relpath(p, root)
|
|
relnoext = os.path.splitext(rel)[0]
|
|
idx.setdefault(stem, p)
|
|
idx[relnoext] = p # full-path form always wins if explicit
|
|
return idx
|
|
|
|
|
|
def extract_links(text):
|
|
"""Return list of (raw_target, cleaned_target). Strips #heading and |alias."""
|
|
out = []
|
|
for m in WIKILINK_RE.finditer(text):
|
|
raw = m.group(1)
|
|
cleaned = raw.split("|")[0].split("#")[0].strip()
|
|
out.append((raw, cleaned))
|
|
return out
|
|
|
|
|
|
# link targets we deliberately skip (blobs, code, doc examples)
|
|
_BLOB_EXT = re.compile(r'\.(jpg|jpeg|png|gif|webp|pdf|svg|opus|mp3|mp4|mov|3gp|zip|xlsx?|docx?|pptx?|csv|heic)$', re.I)
|
|
_NOISE = ("this.file", "{{", "...", '"')
|
|
|
|
|
|
def link_is_checkable(target):
|
|
if not target:
|
|
return False
|
|
if _BLOB_EXT.search(target):
|
|
return False
|
|
if any(n in target for n in _NOISE):
|
|
return False
|
|
return True
|
|
|
|
|
|
def find_broken_links(root, include_archive=False):
|
|
"""
|
|
Return list of dicts: {file, target, guess}
|
|
guess = an existing note title that likely IS the intended target, else None.
|
|
Guess strategies (in order): case-insensitive title match; then match on the
|
|
NN.NN ID-number tail across systems (catches migration carry-forwards like
|
|
[[11.30 Mental health]] -> R.12.30, and stale un-prefixed self-links [[13.05 ..]] -> F.13.05).
|
|
Skips the frozen 09-archive subtree by default.
|
|
"""
|
|
idx = build_title_index(root)
|
|
low = {}
|
|
for k in idx:
|
|
low.setdefault(k.lower(), k)
|
|
# index existing notes by their NN.NN id-number tail -> [stems]
|
|
idnum_re = re.compile(r'(\d\d\.\d\d)')
|
|
by_idnum = {}
|
|
for k in idx:
|
|
m = re.match(r'^[RJFI]?\.?(\d\d\.\d\d)', k)
|
|
if m:
|
|
by_idnum.setdefault(m.group(1), []).append(k)
|
|
broken = []
|
|
for p in all_notes(root, include_archive=include_archive):
|
|
try:
|
|
text = open(p, encoding="utf-8").read()
|
|
except Exception:
|
|
continue
|
|
for raw, cleaned in extract_links(text):
|
|
if not link_is_checkable(cleaned):
|
|
continue
|
|
if cleaned in idx:
|
|
continue
|
|
base = cleaned.split("/")[-1]
|
|
if base in idx:
|
|
continue
|
|
guess = low.get(cleaned.lower()) or low.get(base.lower())
|
|
if not guess:
|
|
m = re.match(r'^[RJFI]?\.?(\d\d\.\d\d)', cleaned)
|
|
if m and m.group(1) in by_idnum:
|
|
cands = by_idnum[m.group(1)]
|
|
if len(cands) == 1:
|
|
guess = cands[0]
|
|
broken.append({
|
|
"file": os.path.relpath(p, root),
|
|
"target": cleaned,
|
|
"guess": guess,
|
|
})
|
|
return broken
|
|
|
|
|
|
def topology_diff(vault_root=VAULT, data_root=DATA):
|
|
"""
|
|
Compare the SYS.AC.ID skeleton of /vault (authoritative) vs /data.
|
|
/data skips `0 System area` and is allowed `_unsorted`.
|
|
Returns dict of missing/orphan at each level.
|
|
"""
|
|
v = walk_structure(vault_root, include_meta=False)
|
|
d = walk_structure(data_root, include_meta=False)
|
|
out = {}
|
|
for level in ("systems", "areas", "cats", "ids"):
|
|
vk = set(v[level].keys())
|
|
dk = set(d[level].keys())
|
|
out[level] = {
|
|
"missing_in_data": sorted(vk - dk), # vault has it, /data lacks a home
|
|
"orphan_in_data": sorted(dk - vk), # /data has it, no vault note
|
|
}
|
|
return out
|
|
|
|
|
|
# ---- topology manifest (the 00-area source-of-truth note) ----
|
|
MANIFEST_REL = "0 System area/05 AI Agent/05 AI Agent.md"
|
|
MANIFEST_BEGIN = "<!-- VAULT-TOPOLOGY-JSON:BEGIN -->"
|
|
MANIFEST_END = "<!-- VAULT-TOPOLOGY-JSON:END -->"
|
|
|
|
|
|
def build_manifest(vault_root=VAULT):
|
|
"""Build the topology manifest dict from the live vault (authoritative)."""
|
|
s = walk_structure(vault_root, include_meta=True)
|
|
return {
|
|
"systems": {k: os.path.basename(v) for k, v in sorted(s["systems"].items())},
|
|
"areas": {k: os.path.basename(v) for k, v in sorted(s["areas"].items())},
|
|
"categories": {k: os.path.basename(v) for k, v in sorted(s["cats"].items())},
|
|
"ids": {k: os.path.basename(v) for k, v in sorted(s["ids"].items())},
|
|
"counts": {
|
|
"systems": len(s["systems"]),
|
|
"areas": len(s["areas"]),
|
|
"categories": len(s["cats"]),
|
|
"ids": len(s["ids"]),
|
|
},
|
|
}
|
|
|
|
|
|
def read_manifest(vault_root=VAULT):
|
|
"""Parse the JSON block out of the manifest note. Return dict or None."""
|
|
p = os.path.join(vault_root, MANIFEST_REL)
|
|
if not os.path.exists(p):
|
|
return None
|
|
text = open(p, encoding="utf-8").read()
|
|
i = text.find(MANIFEST_BEGIN)
|
|
j = text.find(MANIFEST_END)
|
|
if i < 0 or j < 0:
|
|
return None
|
|
block = text[i + len(MANIFEST_BEGIN):j]
|
|
block = block.strip().strip("`").strip()
|
|
if block.startswith("json"):
|
|
block = block[4:].strip()
|
|
try:
|
|
return json.loads(block)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
s = walk_structure(VAULT, include_meta=True)
|
|
print(json.dumps({k: len(v) for k, v in s.items()}, indent=2))
|