ryan/vault-cli
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

vault-cli: Librarian tooling for the R/J/F/I vault

Browse Source 
Adds packages/vault-cli (the `vault` CLI: search/show/tree, clobber-safe
create/move/rename across both /vault and /data + manifest regen, sync-data,
report, analyze) and wires it onto $PATH via modules/hosts/core/vault.nix for
server + vault-client hosts. Scripts self-locate siblings via realpath so they
run from the Nix store. Includes the deterministic $0 report generator, the
topology-manifest generator, and the shared Librarian prompt.
This commit is contained in:
Hermes Agent
2026-07-22 11:11:03 -04:00
commit 48681b00c5
7 changed files with 1039 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vault_ops.py
+223
View File
@@ -0,0 +1,223 @@
#!/usr/bin/env python3
"""
vault_ops.py — write operations for the `vault` CLI.
Clobber-safe by construction: refuse if destination exists, refuse if source missing.
Every op is dry-run unless execute=True. Operates on BOTH /vault and /data + manifest.
"""
import os, re, sys, subprocess
_SELF_DIR = os.path.dirname(os.path.realpath(__file__))
sys.path.insert(0, _SELF_DIR)
import vault_lib as vl
WIKILINK_RE = re.compile(r'(!?\[\[)([^\]#|]+)((?:#[^\]|]+)?(?:\|[^\]]+)?)(\]\])')
def _die(m):
print("ERROR: " + m, file=sys.stderr); sys.exit(1)
def _sys_root(letter):
s = vl.walk_structure(vl.VAULT, include_meta=True)["systems"]
if letter not in s:
_die(f"unknown system '{letter}'")
return os.path.join(vl.VAULT, s[letter])
def _find_id_folder(root, code):
for dp, dns, fns in os.walk(root):
dns[:] = [d for d in dns if d not in vl.SKIP_DIRS]
if vl.id_code(os.path.basename(dp)) == code:
return dp
return None
def _find_cat_folder(root, catcode):
for dp, dns, fns in os.walk(root):
dns[:] = [d for d in dns if d not in vl.SKIP_DIRS]
if vl.cat_code(os.path.basename(dp)) == catcode:
return dp
return None
def _rewrite_links(old_stem, new_stem, execute):
"""Rewrite [[old_stem]] -> [[new_stem]] vault-wide. Returns count."""
n = 0
for p in vl.all_notes(vl.VAULT):
try:
txt = open(p, encoding="utf-8").read()
except Exception:
continue
def repl(m):
nonlocal n
if m.group(2).strip() == old_stem:
n += 1
return f"{m.group(1)}{new_stem}{m.group(3)}{m.group(4)}"
return m.group(0)
new = WIKILINK_RE.sub(repl, txt)
if new != txt and execute:
open(p, "w", encoding="utf-8").write(new)
return n
def _git_commit(msg, execute):
if not execute:
return
subprocess.run(["git", "-C", vl.VAULT, "add", "-A"], check=False)
subprocess.run(["git", "-C", vl.VAULT, "commit", "-m", msg], check=False)
def op_create(argv, execute):
if len(argv) < 2:
_die('usage: vault create <ID> "<title>"')
code, title = argv[0], argv[1]
if not vl.ID_RE.match(code + " x"):
_die(f"'{code}' is not a valid SYS.AC.ID code")
root = _sys_root(code[0])
catcode = code[:code.rindex(".")]
catfolder = _find_cat_folder(root, catcode)
if not catfolder:
_die(f"category {catcode} does not exist — create it first")
if _find_id_folder(root, code):
_die(f"CLOBBER: ID {code} already exists")
leaf = f"{code} {title}"
vpath = os.path.join(catfolder, leaf)
vnote = os.path.join(vpath, leaf + ".md")
drel = os.path.relpath(vpath, vl.VAULT)
dpath = os.path.join(vl.DATA, drel)
print(f"CREATE /vault: {os.path.relpath(vpath, vl.VAULT)}")
print(f" folder-note: {leaf}.md")
print(f" /data: {drel}")
if os.path.exists(dpath):
_die(f"CLOBBER: /data path already exists: {drel}")
if execute:
os.makedirs(vpath)
with open(vnote, "w", encoding="utf-8") as f:
f.write(f"# {leaf}\n")
os.makedirs(dpath)
_git_commit(f"vault create {code} {title}", execute)
print("EXECUTED.")
else:
print("DRY-RUN — nothing changed. Add --execute.")
def op_move(argv, execute):
if len(argv) < 2:
_die("usage: vault move <oldID> <newID>")
old, new = argv[0], argv[1]
root = _sys_root(old[0])
oldf = _find_id_folder(root, old)
if not oldf:
_die(f"source ID {old} not found")
newroot = _sys_root(new[0])
if _find_id_folder(newroot, new):
_die(f"CLOBBER: destination ID {new} already exists")
newcat = new[:new.rindex(".")]
newcatf = _find_cat_folder(newroot, newcat)
if not newcatf:
_die(f"destination category {newcat} does not exist")
old_base = os.path.basename(oldf)
title = old_base.split(" ", 1)[1] if " " in old_base else ""
new_base = f"{new} {title}".strip()
new_folder = os.path.join(newcatf, new_base)
old_stem, new_stem = old_base, new_base
old_drel = os.path.relpath(oldf, vl.VAULT)
new_drel = os.path.relpath(new_folder, vl.VAULT)
print(f"MOVE {old_base} -> {new_base}")
print(f" /vault: {old_drel} -> {new_drel}")
print(f" /data: {old_drel} -> {new_drel}")
print(f" backlink rewrite: [[{old_stem}]] -> [[{new_stem}]]")
n = _rewrite_links(old_stem, new_stem, execute)
print(f" ({n} backlink(s))")
if execute:
# rename folder-note first, then folder
oldnote = os.path.join(oldf, old_base + ".md")
if os.path.exists(oldnote):
os.rename(oldnote, os.path.join(oldf, new_base + ".md"))
os.rename(oldf, new_folder)
olddata = os.path.join(vl.DATA, old_drel)
if os.path.isdir(olddata):
os.makedirs(os.path.dirname(os.path.join(vl.DATA, new_drel)), exist_ok=True)
os.rename(olddata, os.path.join(vl.DATA, new_drel))
_regen_manifest(execute)
_git_commit(f"vault move {old} -> {new}", execute)
print("EXECUTED.")
else:
print("DRY-RUN — nothing changed. Add --execute.")
def op_rename(argv, execute):
if len(argv) < 2:
_die('usage: vault rename <ID> "<new title>"')
code, title = argv[0], argv[1]
root = _sys_root(code[0])
f = _find_id_folder(root, code)
if not f:
_die(f"ID {code} not found")
old_base = os.path.basename(f)
new_base = f"{code} {title}"
if old_base == new_base:
_die("new title identical to current")
parent = os.path.dirname(f)
if os.path.exists(os.path.join(parent, new_base)):
_die(f"CLOBBER: {new_base} already exists")
print(f"RENAME {old_base} -> {new_base}")
n = _rewrite_links(old_base, new_base, execute)
print(f" ({n} backlink(s))")
if execute:
oldnote = os.path.join(f, old_base + ".md")
if os.path.exists(oldnote):
os.rename(oldnote, os.path.join(f, new_base + ".md"))
newf = os.path.join(parent, new_base)
os.rename(f, newf)
drel_old = os.path.relpath(f, vl.VAULT)
drel_new = os.path.relpath(newf, vl.VAULT)
od = os.path.join(vl.DATA, drel_old)
if os.path.isdir(od):
os.rename(od, os.path.join(vl.DATA, drel_new))
_regen_manifest(execute)
_git_commit(f"vault rename {code}", execute)
print("EXECUTED.")
else:
print("DRY-RUN — nothing changed. Add --execute.")
def op_sync_data(argv, execute):
"""Create any /data skeleton dir that a /vault ID/cat/area lacks. Never deletes."""
v = vl.walk_structure(vl.VAULT, include_meta=False)
made = []
for level in ("areas", "cats", "ids"):
for code, rel in sorted(v[level].items()):
dpath = os.path.join(vl.DATA, rel)
if not os.path.isdir(dpath):
made.append(rel)
if execute:
os.makedirs(dpath, exist_ok=True)
print(f"sync-data: {len(made)} /data dir(s) {'created' if execute else 'to create'}")
for m in made[:40]:
print(" + " + m)
if len(made) > 40:
print(f" ... and {len(made)-40} more")
if execute:
print("EXECUTED.")
else:
print("DRY-RUN — nothing changed. Add --execute.")
def _regen_manifest(execute):
if execute:
subprocess.run([sys.executable, os.path.join(_SELF_DIR, "vault-topology.py"), "--execute"], check=False)
def op_manifest(argv, execute):
subprocess.run([sys.executable, os.path.join(_SELF_DIR, "vault-topology.py")] +
(["--execute"] if execute else []), check=False)
def run(cmd, argv, execute):
{
"create": op_create,
"move": op_move,
"rename": op_rename,
"sync-data": op_sync_data,
"manifest": op_manifest,
}[cmd](argv, execute)