317 lines
10 KiB
Python
Executable File
317 lines
10 KiB
Python
Executable File
#!/usr/bin/env -S uv run --script
|
|
# /// script
|
|
# dependencies = []
|
|
# ///
|
|
|
|
"""
|
|
note.py — backend for /note skill.
|
|
SQLite-backed note store with tags, soft-delete, and operation log.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from collections.abc import Generator
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
DB_PATH = Path(__file__).resolve().parent.parent.parent.parent / "db" / "note.sqlite"
|
|
LOG_PATH = Path(__file__).resolve().parent.parent.parent.parent / "log" / "note.log"
|
|
|
|
_TAG_RE = re.compile(r"^[a-z][a-z0-9-]*$")
|
|
# A URL together with an immediately preceding "Label:" token, if any.
|
|
# The leading separator class swallows the connector that introduced the URL
|
|
# (em-dash, comma, etc.) so it does not dangle once the URL moves to its own line.
|
|
_LABELED_URL_RE = re.compile(r"[\s,;—–-]*([^\s,]+:\s*)?(https?://[^\s,]+)")
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS notes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
content TEXT NOT NULL,
|
|
tags TEXT NOT NULL DEFAULT '[]',
|
|
created_at TEXT NOT NULL,
|
|
deleted_at TEXT
|
|
);
|
|
CREATE TABLE IF NOT EXISTS tags (
|
|
name TEXT PRIMARY KEY,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
"""
|
|
|
|
|
|
def _init_db(conn: sqlite3.Connection) -> None:
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.executescript(SCHEMA)
|
|
_migrate(conn)
|
|
|
|
|
|
def _migrate(conn: sqlite3.Connection) -> None:
|
|
cols = {row[1] for row in conn.execute("PRAGMA table_info(notes)")}
|
|
if "tags" not in cols:
|
|
conn.execute("ALTER TABLE notes ADD COLUMN tags TEXT NOT NULL DEFAULT '[]'")
|
|
if "deleted_at" not in cols:
|
|
conn.execute("ALTER TABLE notes ADD COLUMN deleted_at TEXT")
|
|
conn.commit()
|
|
_backfill_tags(conn)
|
|
|
|
|
|
def _backfill_tags(conn: sqlite3.Connection) -> None:
|
|
"""On first introduction of the registry, seed it from tags already used in notes."""
|
|
existing = {row[0] for row in conn.execute("SELECT name FROM tags")}
|
|
if existing:
|
|
return
|
|
used = {row[0] for row in conn.execute("SELECT DISTINCT value FROM notes, json_each(notes.tags)")}
|
|
if not used:
|
|
return
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
conn.executemany(
|
|
"INSERT OR IGNORE INTO tags(name, created_at) VALUES(?, ?)",
|
|
[(tag, now) for tag in sorted(used)],
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
@contextmanager
|
|
def _connect() -> Generator[sqlite3.Connection, None, None]:
|
|
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
_init_db(conn)
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def _validate_tags(tags: list[str]) -> None:
|
|
for tag in tags:
|
|
if not _TAG_RE.match(tag):
|
|
raise ValueError(
|
|
f"Invalid tag '{tag}' — use lowercase letters, digits, hyphens only (e.g. cli, soft-delete)"
|
|
)
|
|
|
|
|
|
def _tags_display(tags_json: str) -> str:
|
|
tags = json.loads(tags_json)
|
|
if not tags:
|
|
return ""
|
|
return " [" + " ".join(f"#{t}" for t in tags) + "]"
|
|
|
|
|
|
def _urls_on_own_lines(text: str) -> str:
|
|
"""Lay out each URL (and its inline "Label:", if any) on its own bullet line.
|
|
|
|
The chat UI merges two adjacent links into one block and hides the second,
|
|
which also overlays the list number. Putting each URL on its own line keeps
|
|
them separate and the number visible. URLs stay bare so they autolink.
|
|
"""
|
|
if not _LABELED_URL_RE.search(text):
|
|
return text
|
|
|
|
def repl(match: re.Match[str]) -> str:
|
|
label = match.group(1) or ""
|
|
return f"\n - {label}{match.group(2)}"
|
|
|
|
return _LABELED_URL_RE.sub(repl, text)
|
|
|
|
|
|
def _log(op: str, detail: str) -> None:
|
|
LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
|
|
with LOG_PATH.open("a") as f:
|
|
f.write(f"{ts} {op} {detail}\n")
|
|
|
|
|
|
def _active_ids(conn: sqlite3.Connection) -> list[int]:
|
|
rows = conn.execute(
|
|
"SELECT id FROM notes WHERE deleted_at IS NULL ORDER BY created_at DESC"
|
|
).fetchall()
|
|
return [row["id"] for row in rows]
|
|
|
|
|
|
def cmd_add(args: argparse.Namespace) -> int:
|
|
tags: list[str] = args.tags or []
|
|
try:
|
|
_validate_tags(tags)
|
|
except ValueError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
content = args.text.strip()
|
|
tags_json = json.dumps(tags)
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
with _connect() as conn:
|
|
known = {row[0] for row in conn.execute("SELECT name FROM tags")}
|
|
unknown = [tag for tag in tags if tag not in known]
|
|
if unknown:
|
|
print(f"Unknown tag(s): {', '.join(unknown)}", file=sys.stderr)
|
|
return 2
|
|
cur = conn.execute(
|
|
"INSERT INTO notes(content, tags, created_at) VALUES(?, ?, ?)",
|
|
(content, tags_json, created_at),
|
|
)
|
|
conn.commit()
|
|
nid = cur.lastrowid
|
|
tags_log = ",".join(tags)
|
|
_log("ADD", f"id={nid} tags=[{tags_log}] {content}")
|
|
print(f"Noted [#1]: {content}{_tags_display(tags_json)}")
|
|
return 0
|
|
|
|
|
|
def cmd_list(args: argparse.Namespace) -> int:
|
|
with _connect() as conn:
|
|
id_to_display = {nid: i + 1 for i, nid in enumerate(_active_ids(conn))}
|
|
if args.tag:
|
|
placeholders = ",".join("?" * len(args.tag))
|
|
rows = conn.execute(
|
|
f"""
|
|
SELECT id, content, tags FROM notes
|
|
WHERE deleted_at IS NULL
|
|
AND (
|
|
SELECT count(*) FROM json_each(notes.tags)
|
|
WHERE value IN ({placeholders})
|
|
) > 0
|
|
ORDER BY created_at DESC
|
|
LIMIT ? OFFSET ?
|
|
""",
|
|
(*args.tag, args.limit, args.offset),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT id, content, tags FROM notes"
|
|
" WHERE deleted_at IS NULL"
|
|
" ORDER BY created_at DESC LIMIT ? OFFSET ?",
|
|
(args.limit, args.offset),
|
|
).fetchall()
|
|
tag_filter = ",".join(args.tag) if args.tag else "None"
|
|
_log("LIST", f"tag={tag_filter} returned={len(rows)}")
|
|
if not rows:
|
|
print("No notes.")
|
|
return 0
|
|
for row in rows:
|
|
head, sep, rest = _urls_on_own_lines(row["content"]).partition("\n")
|
|
print(f"{id_to_display[row['id']]}. {head}{_tags_display(row['tags'])}{sep}{rest}")
|
|
return 0
|
|
|
|
|
|
def cmd_delete(args: argparse.Namespace) -> int:
|
|
display_id: int = args.id
|
|
deleted_at = datetime.now(timezone.utc).isoformat()
|
|
with _connect() as conn:
|
|
ids = _active_ids(conn)
|
|
idx = display_id - 1
|
|
if idx < 0 or idx >= len(ids):
|
|
print(f"No active note with display id={display_id}.")
|
|
return 1
|
|
nid = ids[idx]
|
|
row = conn.execute(
|
|
"SELECT id, content, tags FROM notes WHERE id = ?", (nid,)
|
|
).fetchone()
|
|
conn.execute("UPDATE notes SET deleted_at = ? WHERE id = ?", (deleted_at, nid))
|
|
conn.commit()
|
|
tags_log = ",".join(json.loads(row["tags"]))
|
|
_log("DELETE", f"display_id={display_id} id={nid} tags=[{tags_log}] content={row['content']!r}")
|
|
print(f"Deleted: {row['content']}")
|
|
return 0
|
|
|
|
|
|
def cmd_show(args: argparse.Namespace) -> int:
|
|
display_id: int = args.id
|
|
with _connect() as conn:
|
|
ids = _active_ids(conn)
|
|
idx = display_id - 1
|
|
if idx < 0 or idx >= len(ids):
|
|
print(f"No active note with display id={display_id}.")
|
|
return 1
|
|
nid = ids[idx]
|
|
row = conn.execute(
|
|
"SELECT id, content, tags, created_at FROM notes WHERE id = ?", (nid,)
|
|
).fetchone()
|
|
_log("SHOW", f"display_id={display_id} id={nid}")
|
|
tags = json.loads(row["tags"])
|
|
tags_line = " ".join(f"#{t}" for t in tags) if tags else "(none)"
|
|
print(f"Note [#{display_id}] (id={row['id']})")
|
|
print(f"created: {row['created_at']}")
|
|
print(f"tags: {tags_line}")
|
|
print(f"content: {row['content']}")
|
|
return 0
|
|
|
|
|
|
def cmd_tag_add(args: argparse.Namespace) -> int:
|
|
name = args.name.strip()
|
|
try:
|
|
_validate_tags([name])
|
|
except ValueError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
with _connect() as conn:
|
|
exists = conn.execute("SELECT 1 FROM tags WHERE name = ?", (name,)).fetchone()
|
|
if exists:
|
|
print(f"Tag '#{name}' already exists.")
|
|
return 0
|
|
created_at = datetime.now(timezone.utc).isoformat()
|
|
conn.execute("INSERT INTO tags(name, created_at) VALUES(?, ?)", (name, created_at))
|
|
conn.commit()
|
|
_log("TAG-ADD", f"name={name}")
|
|
print(f"Tag created: #{name}")
|
|
return 0
|
|
|
|
|
|
def cmd_tag_list(args: argparse.Namespace) -> int:
|
|
with _connect() as conn:
|
|
rows = conn.execute("SELECT name FROM tags ORDER BY name").fetchall()
|
|
_log("TAG-LIST", f"returned={len(rows)}")
|
|
if not rows:
|
|
print("No tags.")
|
|
return 0
|
|
for row in rows:
|
|
print(f"#{row['name']}")
|
|
return 0
|
|
|
|
|
|
def _main() -> int:
|
|
parser = argparse.ArgumentParser(description="Note store")
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
|
|
p_add = sub.add_parser("add", help="Add a note")
|
|
p_add.add_argument("text", help="Note content")
|
|
p_add.add_argument("--tags", nargs="+", metavar="TAG", default=[], help="Tags (lowercase, hyphens allowed)")
|
|
|
|
p_list = sub.add_parser("list", help="List active notes")
|
|
p_list.add_argument("--limit", type=int, default=50)
|
|
p_list.add_argument("--offset", type=int, default=0)
|
|
p_list.add_argument("--tag", nargs="+", metavar="TAG", help="Filter by tag (OR logic)")
|
|
|
|
p_show = sub.add_parser("show", help="Show one note in full by display ID")
|
|
p_show.add_argument("id", type=int, help="Display ID")
|
|
|
|
p_del = sub.add_parser("delete", help="Soft-delete a note by ID")
|
|
p_del.add_argument("id", type=int, help="Note ID")
|
|
|
|
p_tag_add = sub.add_parser("tag-add", help="Register a tag")
|
|
p_tag_add.add_argument("name", help="Tag name (lowercase, hyphens allowed)")
|
|
|
|
sub.add_parser("tag-list", help="List registered tags")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.cmd == "add":
|
|
return cmd_add(args)
|
|
if args.cmd == "list":
|
|
return cmd_list(args)
|
|
if args.cmd == "show":
|
|
return cmd_show(args)
|
|
if args.cmd == "delete":
|
|
return cmd_delete(args)
|
|
if args.cmd == "tag-add":
|
|
return cmd_tag_add(args)
|
|
if args.cmd == "tag-list":
|
|
return cmd_tag_list(args)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(_main())
|