CLI
This commit is contained in:
schmeeve
568
cli/snotes.py
Normal file
568
cli/snotes.py
Normal file
@@ -0,0 +1,568 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.11"
|
||||
# dependencies = ["httpx>=0.27.0", "textual>=1.0.0"]
|
||||
# ///
|
||||
"""snotes CLI — TUI + template launcher for the snotes notes app.
|
||||
|
||||
Install: curl -fsSL <server>/<user>/cli | bash
|
||||
Config: ~/.config/snotes/config.json
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
USAGE = """\
|
||||
snotes — command-line client for the snotes notes app
|
||||
|
||||
usage:
|
||||
snotes open the TUI (recent notes, list, search)
|
||||
snotes recent [-n N] print recent notes (default 20)
|
||||
snotes find <term> open the TUI with a search for <term>
|
||||
snotes new create a note in $EDITOR
|
||||
snotes templates list templates (a note containing $name defines one)
|
||||
snotes <template> [date] open/create a note from a template
|
||||
date: today (default), yesterday, tomorrow,
|
||||
YYYY-MM-DD, or +/-N (days from today)
|
||||
snotes login authenticate for remote (non-LAN) access
|
||||
snotes help this text
|
||||
|
||||
template variables: %date% %time% %datetime% %user%
|
||||
config: ~/.config/snotes/config.json
|
||||
"""
|
||||
|
||||
CONFIG_PATH = Path(
|
||||
os.environ.get("SNOTES_CONFIG", Path.home() / ".config" / "snotes" / "config.json")
|
||||
)
|
||||
|
||||
|
||||
class CliError(Exception):
|
||||
"""Fatal user-facing error."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
url: str
|
||||
user: str
|
||||
token: str | None = None
|
||||
|
||||
|
||||
def load_config() -> Config:
|
||||
try:
|
||||
data = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
|
||||
return Config(url=data["url"].rstrip("/"), user=data["user"], token=data.get("token"))
|
||||
except FileNotFoundError:
|
||||
raise CliError(
|
||||
f"no config at {CONFIG_PATH}\ninstall with: curl -fsSL <server>/<user>/cli | bash"
|
||||
) from None
|
||||
except (json.JSONDecodeError, KeyError) as exc:
|
||||
raise CliError(f"bad config at {CONFIG_PATH}: {exc}") from None
|
||||
|
||||
|
||||
def save_config(cfg: Config) -> None:
|
||||
CONFIG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = CONFIG_PATH.with_suffix(".tmp")
|
||||
tmp.write_text(
|
||||
json.dumps({"url": cfg.url, "user": cfg.user, "token": cfg.token}, indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
tmp.chmod(0o600)
|
||||
tmp.replace(CONFIG_PATH)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# API client
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class Api:
|
||||
def __init__(self, cfg: Config):
|
||||
self.cfg = cfg
|
||||
self.client = httpx.Client(base_url=cfg.url, timeout=15.0)
|
||||
self.client.cookies.set("snotes_user", cfg.user)
|
||||
if cfg.token:
|
||||
self.client.cookies.set("snotes_auth", cfg.token)
|
||||
|
||||
def _check(self, resp: httpx.Response) -> httpx.Response:
|
||||
if resp.status_code == 401:
|
||||
raise CliError("authentication required — run `snotes login`")
|
||||
if resp.status_code == 404:
|
||||
raise CliError("not found")
|
||||
if not resp.is_success:
|
||||
detail = resp.json().get("detail", resp.text) if resp.content else resp.reason_phrase
|
||||
raise CliError(f"server error {resp.status_code}: {detail}")
|
||||
return resp
|
||||
|
||||
def login(self, password: str) -> str:
|
||||
resp = self.client.post("/api/auth/login", json={"password": password})
|
||||
if resp.status_code == 401:
|
||||
raise CliError("invalid password")
|
||||
resp.raise_for_status()
|
||||
token = self.client.cookies.get("snotes_auth")
|
||||
if not token:
|
||||
raise CliError("server did not return an auth cookie")
|
||||
return token
|
||||
|
||||
def recent(self, limit: int = 30) -> list[dict]:
|
||||
return self._check(self.client.get("/api/notes/recent")).json()[:limit]
|
||||
|
||||
def search(self, q: str) -> list[dict]:
|
||||
return self._check(self.client.get("/api/search", params={"q": q})).json()
|
||||
|
||||
def get_note(self, note_id: str) -> dict:
|
||||
return self._check(self.client.get(f"/api/notes/{note_id}")).json()
|
||||
|
||||
def create(self, body: str) -> dict:
|
||||
return self._check(self.client.post("/api/notes", json={"body": body})).json()
|
||||
|
||||
def update(self, note_id: str, body: str) -> dict:
|
||||
return self._check(self.client.put(f"/api/notes/{note_id}", json={"body": body})).json()
|
||||
|
||||
def delete(self, note_id: str) -> None:
|
||||
self._check(self.client.delete(f"/api/notes/{note_id}"))
|
||||
|
||||
def templates(self) -> list[dict]:
|
||||
return self._check(self.client.get("/api/templates")).json()
|
||||
|
||||
def template(self, name: str) -> dict | None:
|
||||
resp = self.client.get(f"/api/templates/{name}")
|
||||
if resp.status_code == 404:
|
||||
return None
|
||||
return self._check(resp).json()
|
||||
|
||||
def lookup(self, title: str) -> dict | None:
|
||||
resp = self.client.get("/api/notes/lookup", params={"title": title})
|
||||
if resp.status_code == 404:
|
||||
return None
|
||||
return self._check(resp).json()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dates / template rendering (pure, testable)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
MARKER_LINE_RE = re.compile(r"^\s*\$[A-Za-z][A-Za-z0-9_-]*\s*$")
|
||||
|
||||
|
||||
def parse_date(s: str | None) -> date:
|
||||
today = date.today()
|
||||
if not s or s == "today":
|
||||
return today
|
||||
if s == "yesterday":
|
||||
return today - timedelta(days=1)
|
||||
if s == "tomorrow":
|
||||
return today + timedelta(days=1)
|
||||
if re.fullmatch(r"[+-]?\d+", s):
|
||||
return today + timedelta(days=int(s))
|
||||
try:
|
||||
return date.fromisoformat(s)
|
||||
except ValueError:
|
||||
raise CliError(
|
||||
f"bad date {s!r} — use today, yesterday, tomorrow, YYYY-MM-DD, or +/-N"
|
||||
) from None
|
||||
|
||||
|
||||
def render_template(body: str, d: date, user: str, now: datetime | None = None) -> str:
|
||||
"""Substitute variables and strip `$marker` template-definition lines."""
|
||||
now = now or datetime.now()
|
||||
out = body.replace("%datetime%", f"{d.isoformat()} {now:%H:%M}")
|
||||
out = out.replace("%date%", d.isoformat())
|
||||
out = out.replace("%time%", f"{now:%H:%M}")
|
||||
out = out.replace("%user%", user)
|
||||
lines = [ln for ln in out.splitlines() if not MARKER_LINE_RE.match(ln)]
|
||||
text = "\n".join(lines).strip("\n")
|
||||
return text + "\n" if text else ""
|
||||
|
||||
|
||||
def extract_title(text: str) -> str:
|
||||
for line in text.splitlines():
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
title = stripped[2:].strip() if stripped.startswith("# ") else stripped
|
||||
if title:
|
||||
return title[:200]
|
||||
return "Untitled"
|
||||
|
||||
|
||||
def tags_str(note: dict) -> str:
|
||||
tags = note.get("tags") or []
|
||||
return " ".join(f"#{t}" for t in tags)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# $EDITOR round-trip
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def edit_roundtrip(initial: str) -> str | None:
|
||||
"""Open `initial` in $EDITOR; return the edited text, or None on failure/abort."""
|
||||
editor = shlex.split(os.environ.get("EDITOR", "vim"))
|
||||
fd, path = tempfile.mkstemp(prefix="snotes-", suffix=".md")
|
||||
try:
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
||||
f.write(initial)
|
||||
rc = subprocess.call([*editor, path])
|
||||
if rc != 0:
|
||||
return None
|
||||
return Path(path).read_text(encoding="utf-8")
|
||||
finally:
|
||||
Path(path).unlink(missing_ok=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Commands
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def cmd_recent(api: Api, limit: int) -> int:
|
||||
for n in api.recent(limit):
|
||||
updated = (n.get("updated_at") or "")[:16]
|
||||
tags = tags_str(n)
|
||||
line = f"{updated:16} {n['title']}"
|
||||
if tags:
|
||||
line += f" {tags}"
|
||||
print(line)
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_templates(api: Api) -> int:
|
||||
rows = api.templates()
|
||||
if not rows:
|
||||
print("no templates — add e.g. $journal to a note to define one")
|
||||
return 0
|
||||
for t in rows:
|
||||
print(f"${t['name']:20} {t['title']} ({t['username']})")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_new(api: Api) -> int:
|
||||
text = edit_roundtrip("")
|
||||
if text and text.strip():
|
||||
note = api.create(text)
|
||||
print(f"created: {note['title']}")
|
||||
else:
|
||||
print("aborted — empty note")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_template(api: Api, cfg: Config, name: str, datestr: str | None) -> int:
|
||||
d = parse_date(datestr)
|
||||
tpl = api.template(name)
|
||||
if tpl is None:
|
||||
known = ", ".join(f"${t['name']}" for t in api.templates()) or "(none)"
|
||||
raise CliError(f"no template ${name!s} — known: {known}")
|
||||
body = render_template(tpl["body"], d, cfg.user)
|
||||
existing = api.lookup(extract_title(body))
|
||||
if existing:
|
||||
edited = edit_roundtrip(existing["body"])
|
||||
if edited is not None and edited != existing["body"] and edited.strip():
|
||||
api.update(existing["id"], edited)
|
||||
print(f"updated: {extract_title(edited)}")
|
||||
else:
|
||||
print("no changes")
|
||||
else:
|
||||
edited = edit_roundtrip(body)
|
||||
if edited and edited.strip():
|
||||
note = api.create(edited)
|
||||
print(f"created: {note['title']}")
|
||||
else:
|
||||
print("aborted — empty note")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_login(api: Api, cfg: Config) -> int:
|
||||
import getpass
|
||||
|
||||
cfg.token = api.login(getpass.getpass("snotes password: "))
|
||||
save_config(cfg)
|
||||
print(f"✓ logged in — token saved to {CONFIG_PATH}")
|
||||
return 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TUI
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def make_app(api: Api, initial_query: str | None):
|
||||
"""Build the Textual app (imports deferred so the CLI module stays importable
|
||||
without textual installed)."""
|
||||
import asyncio
|
||||
|
||||
from rich.text import Text
|
||||
from textual import work
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Horizontal, Vertical
|
||||
from textual.screen import ModalScreen
|
||||
from textual.widgets import Footer, Input, Label, ListItem, ListView, Markdown
|
||||
|
||||
class Confirm(ModalScreen[bool]):
|
||||
BINDINGS = [
|
||||
Binding("y", "confirm", "Yes"),
|
||||
Binding("n", "cancel", "No"),
|
||||
Binding("escape", "cancel"),
|
||||
]
|
||||
|
||||
def __init__(self, prompt: str):
|
||||
super().__init__()
|
||||
self._prompt = prompt
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Vertical(
|
||||
Label(self._prompt),
|
||||
Label("[y] yes [n/esc] no", id="confirm-hint"),
|
||||
id="confirm-box",
|
||||
)
|
||||
|
||||
def action_confirm(self) -> None:
|
||||
self.dismiss(True)
|
||||
|
||||
def action_cancel(self) -> None:
|
||||
self.dismiss(False)
|
||||
|
||||
class NotesApp(App):
|
||||
CSS = """
|
||||
Screen { background: #1a1b26; }
|
||||
#search { margin: 1 2 0 2; border: solid #3b4261; }
|
||||
#search:focus { border: solid #7aa2f7; }
|
||||
#main { height: 1fr; }
|
||||
#notes { width: 42; border: solid #3b4261; margin: 1 0 1 2; }
|
||||
#notes:focus { border: solid #7aa2f7; }
|
||||
#preview { border: solid #3b4261; margin: 1 2 1 1; padding: 0 2;
|
||||
background: #24283b; }
|
||||
ListItem { color: #c0caf5; }
|
||||
ListView { background: #1f2335; }
|
||||
Footer { background: #1f2335; color: #565f89; }
|
||||
#confirm-box { width: 50; height: auto; border: solid #f7768e;
|
||||
background: #24283b; padding: 1 2; margin: 10 0 0 4; }
|
||||
#confirm-hint { color: #565f89; margin-top: 1; }
|
||||
"""
|
||||
|
||||
BINDINGS = [
|
||||
Binding("q", "quit", "Quit"),
|
||||
Binding("n", "new_note", "New"),
|
||||
Binding("d", "delete_note", "Delete"),
|
||||
Binding("r", "refresh", "Refresh"),
|
||||
Binding("slash", "focus_search", "Search"),
|
||||
]
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._cache: dict[str, str] = {}
|
||||
self._gen = 0
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Input(placeholder="Search notes… (#tag works too)", id="search")
|
||||
with Horizontal(id="main"):
|
||||
yield ListView(id="notes")
|
||||
yield Markdown("", id="preview")
|
||||
yield Footer()
|
||||
|
||||
async def on_mount(self) -> None:
|
||||
self.title = "snotes"
|
||||
self.sub_title = f"{api.cfg.user} @ {api.cfg.url}"
|
||||
if initial_query:
|
||||
self.query_one("#search", Input).value = initial_query
|
||||
self.query_one("#search", Input).focus()
|
||||
else:
|
||||
await self.reload()
|
||||
self.query_one("#notes", ListView).focus()
|
||||
|
||||
# -- data ----------------------------------------------------------
|
||||
|
||||
async def reload(self) -> None:
|
||||
q = self.query_one("#search", Input).value.strip()
|
||||
try:
|
||||
if q:
|
||||
rows = await asyncio.to_thread(api.search, q)
|
||||
else:
|
||||
rows = await asyncio.to_thread(api.recent, 30)
|
||||
except CliError as exc:
|
||||
self.notify(str(exc), severity="error")
|
||||
return
|
||||
lv = self.query_one("#notes", ListView)
|
||||
await lv.clear()
|
||||
for n in rows:
|
||||
label = n["title"] or "Untitled"
|
||||
tags = tags_str(n)
|
||||
text = Text(f"{label}\n", overflow="ellipsis")
|
||||
text.append(
|
||||
f"{(n.get('updated_at') or '')[:10]}" + (f" {tags}" if tags else ""),
|
||||
style="dim",
|
||||
)
|
||||
item = ListItem(Label(text))
|
||||
item.note_id = n["id"]
|
||||
await lv.append(item)
|
||||
self.query_one("#preview", Markdown).update("")
|
||||
if lv.children:
|
||||
lv.index = 0
|
||||
|
||||
# -- events --------------------------------------------------------
|
||||
|
||||
async def on_input_changed(self, event: Input.Changed) -> None:
|
||||
self._gen += 1
|
||||
gen = self._gen
|
||||
await asyncio.sleep(0.25)
|
||||
if gen == self._gen:
|
||||
await self.reload()
|
||||
|
||||
async def on_list_view_highlighted(self, event: ListView.Highlighted) -> None:
|
||||
item = event.item
|
||||
note_id = getattr(item, "note_id", None)
|
||||
if not note_id:
|
||||
return
|
||||
body = self._cache.get(note_id)
|
||||
if body is None:
|
||||
try:
|
||||
note = await asyncio.to_thread(api.get_note, note_id)
|
||||
except CliError:
|
||||
return
|
||||
body = note["body"]
|
||||
self._cache[note_id] = body
|
||||
self.query_one("#preview", Markdown).update(body)
|
||||
|
||||
async def on_list_view_selected(self, event: ListView.Selected) -> None:
|
||||
note_id = getattr(event.item, "note_id", None)
|
||||
if not note_id:
|
||||
return
|
||||
try:
|
||||
note = await asyncio.to_thread(api.get_note, note_id)
|
||||
except CliError as exc:
|
||||
self.notify(str(exc), severity="error")
|
||||
return
|
||||
with self.suspend():
|
||||
edited = edit_roundtrip(note["body"])
|
||||
if edited is not None and edited != note["body"] and edited.strip():
|
||||
try:
|
||||
await asyncio.to_thread(api.update, note_id, edited)
|
||||
except CliError as exc:
|
||||
self.notify(str(exc), severity="error")
|
||||
self._cache.pop(note_id, None)
|
||||
await self.reload()
|
||||
|
||||
# -- actions -------------------------------------------------------
|
||||
|
||||
async def action_new_note(self) -> None:
|
||||
with self.suspend():
|
||||
text = edit_roundtrip("")
|
||||
if text and text.strip():
|
||||
try:
|
||||
await asyncio.to_thread(api.create, text)
|
||||
except CliError as exc:
|
||||
self.notify(str(exc), severity="error")
|
||||
await self.reload()
|
||||
|
||||
async def action_refresh(self) -> None:
|
||||
self._cache.clear()
|
||||
await self.reload()
|
||||
|
||||
def action_focus_search(self) -> None:
|
||||
self.query_one("#search", Input).focus()
|
||||
|
||||
def action_delete_note(self) -> None:
|
||||
lv = self.query_one("#notes", ListView)
|
||||
item = lv.highlighted_child
|
||||
note_id = getattr(item, "note_id", None)
|
||||
if not note_id:
|
||||
return
|
||||
self.push_screen(
|
||||
Confirm("Delete this note?"),
|
||||
lambda ok: self._delete(ok, note_id),
|
||||
)
|
||||
|
||||
@work
|
||||
async def _delete(self, ok: bool, note_id: str) -> None:
|
||||
if not ok:
|
||||
return
|
||||
try:
|
||||
await asyncio.to_thread(api.delete, note_id)
|
||||
except CliError as exc:
|
||||
self.notify(str(exc), severity="error")
|
||||
return
|
||||
self._cache.pop(note_id, None)
|
||||
await self.reload()
|
||||
|
||||
return NotesApp()
|
||||
|
||||
|
||||
def cmd_tui(api: Api, initial_query: str | None) -> int:
|
||||
make_app(api, initial_query).run()
|
||||
return 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Entry point
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if argv and argv[0] in ("-h", "--help", "help"):
|
||||
print(USAGE)
|
||||
return 0
|
||||
|
||||
cfg = load_config()
|
||||
api = Api(cfg)
|
||||
|
||||
if not argv:
|
||||
return cmd_tui(api, None)
|
||||
|
||||
cmd, rest = argv[0], argv[1:]
|
||||
|
||||
if cmd == "tui":
|
||||
return cmd_tui(api, None)
|
||||
if cmd == "recent":
|
||||
limit = 20
|
||||
if rest:
|
||||
if len(rest) == 2 and rest[0] == "-n" and rest[1].isdigit():
|
||||
limit = int(rest[1])
|
||||
else:
|
||||
raise CliError("usage: snotes recent [-n N]")
|
||||
return cmd_recent(api, limit)
|
||||
if cmd == "find":
|
||||
term = " ".join(rest).strip()
|
||||
if not term:
|
||||
raise CliError("usage: snotes find <term>")
|
||||
return cmd_tui(api, term)
|
||||
if cmd == "new":
|
||||
return cmd_new(api)
|
||||
if cmd == "templates":
|
||||
return cmd_templates(api)
|
||||
if cmd == "login":
|
||||
return cmd_login(api, cfg)
|
||||
if cmd.startswith("-"):
|
||||
raise CliError(f"unknown option {cmd!r}\n\n{USAGE}")
|
||||
if len(rest) > 1:
|
||||
raise CliError(f"usage: snotes {cmd} [date]")
|
||||
return cmd_template(api, cfg, cmd, rest[0] if rest else None)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
except CliError as exc:
|
||||
print(f"snotes: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except httpx.ConnectError:
|
||||
print(f"snotes: cannot reach the server — check the url in {CONFIG_PATH}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except KeyboardInterrupt:
|
||||
sys.exit(130)
|
||||
Reference in New Issue
Block a user