250 lines
8.0 KiB
Python
250 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Bookmark skill — CRUD for reading-list entries stored in SQLite."""
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
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
|
|
);
|
|
"""
|
|
|
|
|
|
def _init_db(conn: sqlite3.Connection) -> None:
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.executescript(SCHEMA)
|
|
|
|
|
|
@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.
|
|
"""
|
|
if read:
|
|
rows = conn.execute(
|
|
"SELECT id FROM bookmarks WHERE read_at IS NOT NULL ORDER BY read_at DESC"
|
|
).fetchall()
|
|
else:
|
|
rows = conn.execute(
|
|
"SELECT id FROM bookmarks WHERE read_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 _print_bookmark(
|
|
row: sqlite3.Row, display_id: int, *, show_status: bool = False, show_read_date: 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 ""
|
|
print(f"#{display_id} {_domain(row['url'])}{tag_str}")
|
|
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)
|
|
with _connect() as conn:
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
conn.execute(
|
|
"INSERT INTO bookmarks (url, description, tags, created_at) VALUES (?, ?, ?, ?)",
|
|
(args.url, args.description, json.dumps(tags, ensure_ascii=False), now),
|
|
)
|
|
conn.commit()
|
|
tag_info = f" [{', '.join(tags)}]" if tags else ""
|
|
# Newest unread sorts first, so a fresh bookmark is always display #1.
|
|
print(f"Added bookmark #1: {args.url}{tag_info}")
|
|
|
|
|
|
def cmd_list(args: argparse.Namespace) -> None:
|
|
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(
|
|
"""SELECT * FROM bookmarks
|
|
WHERE read_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(
|
|
"SELECT * FROM bookmarks WHERE read_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"]])
|
|
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)
|
|
|
|
|
|
def cmd_history(args: argparse.Namespace) -> None:
|
|
with _connect() as conn:
|
|
rows = conn.execute(
|
|
"SELECT * FROM bookmarks WHERE read_at IS NOT 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)
|
|
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")
|
|
|
|
# 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`")
|
|
|
|
# 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,
|
|
"history": cmd_history,
|
|
}
|
|
args = parser.parse_args()
|
|
dispatch[args.command](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|