326 lines
11 KiB
Python
326 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Bookmark skill — CRUD for reading-list entries stored in SQLite."""
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
DB_PATH = (
|
|
Path(__file__).resolve().parent.parent.parent.parent / "db" / "bookmark.sqlite"
|
|
)
|
|
|
|
EMPTY_TAGS_JSON = "[]"
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS bookmarks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
url TEXT NOT NULL,
|
|
description TEXT NOT NULL DEFAULT '',
|
|
tags TEXT NOT NULL DEFAULT '[]',
|
|
created_at TEXT NOT NULL,
|
|
read_at TEXT,
|
|
content TEXT,
|
|
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:
|
|
"""Add columns missing from a pre-existing DB (CREATE IF NOT EXISTS won't)."""
|
|
columns = {row["name"] for row in conn.execute("PRAGMA table_info(bookmarks)")}
|
|
if "content" not in columns:
|
|
conn.execute("ALTER TABLE bookmarks ADD COLUMN content TEXT")
|
|
if "deleted_at" not in columns:
|
|
conn.execute("ALTER TABLE bookmarks ADD COLUMN deleted_at TEXT")
|
|
|
|
|
|
@contextmanager
|
|
def _connect() -> sqlite3.Connection:
|
|
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 _ordered_ids(conn: sqlite3.Connection, *, read: bool) -> list[int]:
|
|
"""Internal ids of one bookmark set in display order.
|
|
|
|
Unread (`read=False`) is what `list` shows, read (`read=True`) what `history`
|
|
shows. Display IDs are 1-based positions here, computed on the fly — never
|
|
stored — so they renumber whenever the set changes. Soft-deleted rows
|
|
(`deleted_at` set) are excluded from both sets.
|
|
"""
|
|
if read:
|
|
rows = conn.execute(
|
|
"SELECT id FROM bookmarks WHERE read_at IS NOT NULL AND deleted_at IS NULL ORDER BY read_at DESC"
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT id FROM bookmarks WHERE read_at IS NULL AND deleted_at IS NULL ORDER BY created_at DESC"
|
|
).fetchall()
|
|
return [row["id"] for row in rows]
|
|
|
|
|
|
def _resolve_display_id(conn: sqlite3.Connection, display_id: int, *, read: bool) -> int | None:
|
|
"""Translate a display ID into an internal id, or None if out of range."""
|
|
order = _ordered_ids(conn, read=read)
|
|
idx = display_id - 1
|
|
if idx < 0 or idx >= len(order):
|
|
return None
|
|
return order[idx]
|
|
|
|
|
|
def _parse_tags(raw: str) -> list[str]:
|
|
"""Parse comma-separated tags into a deduplicated sorted list."""
|
|
if not raw:
|
|
return []
|
|
tags = [t.strip() for t in raw.split(",") if t.strip()]
|
|
return sorted(set(tags))
|
|
|
|
|
|
def _tags_display(tags_json: str) -> str:
|
|
tags = json.loads(tags_json)
|
|
return ", ".join(tags) if tags else ""
|
|
|
|
|
|
def _domain(url: str) -> str:
|
|
"""Extract domain from URL (strip www. prefix)."""
|
|
try:
|
|
parsed = urlparse(url)
|
|
host = parsed.hostname or ""
|
|
return host.removeprefix("www.")
|
|
except (ValueError, AttributeError):
|
|
return url
|
|
|
|
|
|
def _read_content(content_file: str | None) -> str | None:
|
|
"""Read cleaned article markdown from a file path, or stdin when path is '-'."""
|
|
if content_file is None:
|
|
return None
|
|
text = sys.stdin.read() if content_file == "-" else Path(content_file).read_text(encoding="utf-8")
|
|
return text.strip() or None
|
|
|
|
|
|
def _print_bookmark(
|
|
row: sqlite3.Row,
|
|
display_id: int,
|
|
*,
|
|
show_status: bool = False,
|
|
show_read_date: bool = False,
|
|
has_content: bool = False,
|
|
) -> None:
|
|
"""Format and print a single bookmark row under its display ID."""
|
|
tags = json.loads(row["tags"])
|
|
tag_str = f" [{', '.join(tags)}]" if tags else ""
|
|
content_marker = " 📄" if has_content else ""
|
|
print(f"#{display_id} {_domain(row['url'])}{tag_str}{content_marker}")
|
|
print(f" {row['description']}")
|
|
print(f" {row['url']}")
|
|
line = f" added: {row['created_at'][:10]}"
|
|
if show_status:
|
|
status = "read" if row["read_at"] else "unread"
|
|
line += f" status: {status}"
|
|
if show_read_date and row["read_at"]:
|
|
line += f" read: {row['read_at'][:10]}"
|
|
print(line)
|
|
|
|
|
|
def cmd_add(args: argparse.Namespace) -> None:
|
|
tags = _parse_tags(args.tags)
|
|
content = _read_content(args.content_file)
|
|
with _connect() as conn:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
conn.execute(
|
|
"INSERT INTO bookmarks (url, description, tags, created_at, content) VALUES (?, ?, ?, ?, ?)",
|
|
(args.url, args.description, json.dumps(tags, ensure_ascii=False), now, content),
|
|
)
|
|
conn.commit()
|
|
tag_info = f" [{', '.join(tags)}]" if tags else ""
|
|
content_info = " (article content stored)" if content else ""
|
|
# Newest unread sorts first, so a fresh bookmark is always display #1.
|
|
print(f"Added bookmark #1: {args.url}{tag_info}{content_info}")
|
|
|
|
|
|
def cmd_list(args: argparse.Namespace) -> None:
|
|
# Skip the (potentially large) content blob here — only whether it exists.
|
|
columns = "id, url, description, tags, created_at, read_at, content IS NOT NULL AS has_content"
|
|
with _connect() as conn:
|
|
display_by_id = {nid: i + 1 for i, nid in enumerate(_ordered_ids(conn, read=False))}
|
|
if args.tag:
|
|
rows = conn.execute(
|
|
f"""SELECT {columns} FROM bookmarks
|
|
WHERE read_at IS NULL AND deleted_at IS NULL AND EXISTS (
|
|
SELECT 1 FROM json_each(tags) WHERE json_each.value = ?
|
|
)
|
|
ORDER BY created_at DESC""",
|
|
(args.tag,),
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
f"SELECT {columns} FROM bookmarks WHERE read_at IS NULL AND deleted_at IS NULL ORDER BY created_at DESC"
|
|
).fetchall()
|
|
|
|
if not rows:
|
|
print(
|
|
"No bookmarks." if not args.tag else f"No bookmarks with tag '{args.tag}'."
|
|
)
|
|
return
|
|
|
|
# Display IDs come from the full unread set so a tag-filtered list keeps the
|
|
# same numbers `read`/`show` resolve against (gaps are expected when filtered).
|
|
for r in rows:
|
|
_print_bookmark(r, display_by_id[r["id"]], has_content=bool(r["has_content"]))
|
|
print()
|
|
|
|
|
|
def cmd_read(args: argparse.Namespace) -> None:
|
|
with _connect() as conn:
|
|
internal_id = _resolve_display_id(conn, args.id, read=False)
|
|
if internal_id is None:
|
|
print(f"No unread bookmark #{args.id}.")
|
|
return
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
conn.execute("UPDATE bookmarks SET read_at = ? WHERE id = ?", (now, internal_id))
|
|
conn.commit()
|
|
print(f"Marked bookmark #{args.id} as read.")
|
|
|
|
|
|
def cmd_unread(args: argparse.Namespace) -> None:
|
|
with _connect() as conn:
|
|
internal_id = _resolve_display_id(conn, args.id, read=True)
|
|
if internal_id is None:
|
|
print(f"No read bookmark #{args.id} in history.")
|
|
return
|
|
conn.execute("UPDATE bookmarks SET read_at = NULL WHERE id = ?", (internal_id,))
|
|
conn.commit()
|
|
print(f"Unmarked bookmark #{args.id}.")
|
|
|
|
|
|
def cmd_show(args: argparse.Namespace) -> None:
|
|
with _connect() as conn:
|
|
internal_id = _resolve_display_id(conn, args.id, read=False)
|
|
if internal_id is None:
|
|
print(f"No unread bookmark #{args.id}.")
|
|
return
|
|
row = conn.execute("SELECT * FROM bookmarks WHERE id = ?", (internal_id,)).fetchone()
|
|
_print_bookmark(row, args.id, show_status=True, has_content=row["content"] is not None)
|
|
|
|
|
|
def cmd_content(args: argparse.Namespace) -> None:
|
|
with _connect() as conn:
|
|
internal_id = _resolve_display_id(conn, args.id, read=False)
|
|
if internal_id is None:
|
|
print(f"No unread bookmark #{args.id}.")
|
|
return
|
|
row = conn.execute("SELECT content FROM bookmarks WHERE id = ?", (internal_id,)).fetchone()
|
|
if not row["content"]:
|
|
print(f"Bookmark #{args.id} has no stored article content.")
|
|
return
|
|
print(row["content"])
|
|
|
|
|
|
def cmd_delete(args: argparse.Namespace) -> None:
|
|
# Soft delete: set deleted_at so the row drops out of list/history but stays
|
|
# in the DB. The agent confirms with the user before calling this (see SKILL.md).
|
|
with _connect() as conn:
|
|
internal_id = _resolve_display_id(conn, args.id, read=False)
|
|
if internal_id is None:
|
|
print(f"No unread bookmark #{args.id}.")
|
|
return
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
conn.execute("UPDATE bookmarks SET deleted_at = ? WHERE id = ?", (now, internal_id))
|
|
conn.commit()
|
|
print(f"Deleted bookmark #{args.id}.")
|
|
|
|
|
|
def cmd_history(args: argparse.Namespace) -> None:
|
|
columns = "id, url, description, tags, created_at, read_at, content IS NOT NULL AS has_content"
|
|
with _connect() as conn:
|
|
rows = conn.execute(
|
|
f"SELECT {columns} FROM bookmarks WHERE read_at IS NOT NULL AND deleted_at IS NULL ORDER BY read_at DESC"
|
|
).fetchall()
|
|
|
|
if not rows:
|
|
print("No read bookmarks.")
|
|
return
|
|
|
|
for display_id, r in enumerate(rows, start=1):
|
|
_print_bookmark(r, display_id, show_read_date=True, has_content=bool(r["has_content"]))
|
|
print()
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Bookmark CRUD")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# add
|
|
p_add = sub.add_parser("add", help="Add a bookmark")
|
|
p_add.add_argument("url", help="URL to bookmark")
|
|
p_add.add_argument("description", help="Short description")
|
|
p_add.add_argument("--tags", default="", help="Comma-separated tags")
|
|
p_add.add_argument(
|
|
"--content-file",
|
|
dest="content_file",
|
|
default=None,
|
|
help="Path to cleaned article markdown; '-' reads it from stdin",
|
|
)
|
|
|
|
# list
|
|
p_list = sub.add_parser("list", help="List unread bookmarks")
|
|
p_list.add_argument("--tag", help="Filter by tag (exact match)")
|
|
|
|
# read (mark as read)
|
|
p_read = sub.add_parser("read", help="Mark bookmark as read")
|
|
p_read.add_argument("id", type=int, help="Display ID from `list`")
|
|
|
|
# unread (unmark)
|
|
p_unread = sub.add_parser("unread", help="Unmark bookmark as read")
|
|
p_unread.add_argument("id", type=int, help="Display ID from `history`")
|
|
|
|
# show (display details)
|
|
p_show = sub.add_parser("show", help="Show bookmark details")
|
|
p_show.add_argument("id", type=int, help="Display ID from `list`")
|
|
|
|
# content (print stored article markdown)
|
|
p_content = sub.add_parser("content", help="Print stored article markdown")
|
|
p_content.add_argument("id", type=int, help="Display ID from `list`")
|
|
|
|
# delete (soft delete)
|
|
p_delete = sub.add_parser("delete", help="Soft-delete a bookmark (hidden, kept in DB)")
|
|
p_delete.add_argument("id", type=int, help="Display ID from `list`")
|
|
|
|
# history (list read)
|
|
sub.add_parser("history", help="List read bookmarks")
|
|
|
|
dispatch = {
|
|
"add": cmd_add,
|
|
"list": cmd_list,
|
|
"read": cmd_read,
|
|
"unread": cmd_unread,
|
|
"show": cmd_show,
|
|
"content": cmd_content,
|
|
"delete": cmd_delete,
|
|
"history": cmd_history,
|
|
}
|
|
args = parser.parse_args()
|
|
dispatch[args.command](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|