569 lines
19 KiB
Python
569 lines
19 KiB
Python
#!/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)
|