201 lines
6.2 KiB
Python
201 lines
6.2 KiB
Python
#!/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-]*$")
|
|
|
|
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
|
|
);
|
|
"""
|
|
|
|
|
|
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()
|
|
|
|
|
|
@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 _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:
|
|
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:
|
|
print(f"{id_to_display[row['id']]}. {row['content']}{_tags_display(row['tags'])}")
|
|
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 _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_del = sub.add_parser("delete", help="Soft-delete a note by ID")
|
|
p_del.add_argument("id", type=int, help="Note ID")
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.cmd == "add":
|
|
return cmd_add(args)
|
|
if args.cmd == "list":
|
|
return cmd_list(args)
|
|
if args.cmd == "delete":
|
|
return cmd_delete(args)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(_main())
|