From d2bfed9d129337d025aac8d50927212b5d2eabda Mon Sep 17 00:00:00 2001 From: Research Assistant Date: Wed, 20 May 2026 15:51:32 +0800 Subject: [PATCH] feat: add annotation cli commands - paperforge/cli.py adds annotation subparser with 7 subcommands - paperforge/commands/annotation.py dispatch + command implementations - import/list/create/patch/delete/export/status - all support --json output via PFResult envelope - tests/cli/test_annotation_commands.py 10 tests: parser validation + integration create/list/patch/delete lifecycle --- paperforge/cli.py | 54 ++++++ paperforge/commands/annotation.py | 265 ++++++++++++++++++++++++++ tests/cli/test_annotation_commands.py | 153 +++++++++++++++ 3 files changed, 472 insertions(+) create mode 100644 paperforge/commands/annotation.py create mode 100644 tests/cli/test_annotation_commands.py diff --git a/paperforge/cli.py b/paperforge/cli.py index ecfe0206..92c4fc81 100644 --- a/paperforge/cli.py +++ b/paperforge/cli.py @@ -292,6 +292,55 @@ def build_parser() -> argparse.ArgumentParser: p_memory_status = p_memory_sp.add_parser("status", help="Check memory database status") p_memory_status.add_argument("--json", action="store_true", help="Output as JSON") + # Annotation commands + p_ann = sub.add_parser("annotation", help="Manage PDF annotations") + p_ann_sp = p_ann.add_subparsers(dest="annotation_subcommand", required=True) + + p_ann_import = p_ann_sp.add_parser("import", help="Import annotations from Zotero SQLite") + p_ann_import.add_argument("--zotero-db", metavar="PATH", help="Path to zotero.sqlite") + p_ann_import.add_argument("--paper", metavar="KEY", help="Filter by paper key") + p_ann_import.add_argument("--dry-run", action="store_true", help="Preview without importing") + p_ann_import.add_argument("--no-copy", action="store_true", help="Read zotero.sqlite directly (risk if Zotero is running)") + p_ann_import.add_argument("--json", action="store_true") + + p_ann_list = p_ann_sp.add_parser("list", help="List annotations for a paper") + p_ann_list.add_argument("paper_key", help="Paper zotero key") + p_ann_list.add_argument("--page", type=int, help="Filter by page index") + p_ann_list.add_argument("--type", dest="ann_type", help="Filter by annotation type") + p_ann_list.add_argument("--limit", type=int, default=100, help="Max results") + p_ann_list.add_argument("--json", action="store_true") + + p_ann_create = p_ann_sp.add_parser("create", help="Create a local annotation") + p_ann_create.add_argument("--paper", required=True, help="Paper zotero key") + p_ann_create.add_argument("--type", dest="ann_type", required=True, help="Annotation type") + p_ann_create.add_argument("--page-index", type=int, help="Page index") + p_ann_create.add_argument("--page-label", default="", help="Page label") + p_ann_create.add_argument("--selected-text", default="", help="Selected text") + p_ann_create.add_argument("--comment", default="", help="Comment") + p_ann_create.add_argument("--color", default="#ffd400", help="Color hex") + p_ann_create.add_argument("--sort-index", default="", help="Sort index") + p_ann_create.add_argument("--json", action="store_true") + + p_ann_patch = p_ann_sp.add_parser("patch", help="Update an annotation") + p_ann_patch.add_argument("annotation_id", help="Annotation ID") + p_ann_patch.add_argument("--comment", help="New comment") + p_ann_patch.add_argument("--color", help="New color hex") + p_ann_patch.add_argument("--json", action="store_true") + + p_ann_delete = p_ann_sp.add_parser("delete", help="Delete an annotation") + p_ann_delete.add_argument("annotation_id", help="Annotation ID") + p_ann_delete.add_argument("--hard", action="store_true", help="Permanent delete") + p_ann_delete.add_argument("--json", action="store_true") + + p_ann_export = p_ann_sp.add_parser("export", help="Export annotations") + p_ann_export.add_argument("paper_key", help="Paper zotero key") + p_ann_export.add_argument("--format", choices=["json", "markdown"], default="json") + p_ann_export.add_argument("--output", help="Write to file") + p_ann_export.add_argument("--json", action="store_true") + + p_ann_status = p_ann_sp.add_parser("status", help="Check annotation DB status") + p_ann_status.add_argument("--json", action="store_true") + p_paper_status = sub.add_parser("paper-status", help="Look up a paper's status") p_paper_status.add_argument("query", help="Paper identifier (zotero_key, DOI, title, alias)") p_paper_status.add_argument("--json", action="store_true", help="Output as JSON") @@ -567,6 +616,11 @@ def main(argv: list[str] | None = None) -> int: return run(args) + if args.command == "annotation": + from paperforge.commands.annotation import run + + return run(args) + if args.command == "embed": from paperforge.commands.embed import run diff --git a/paperforge/commands/annotation.py b/paperforge/commands/annotation.py new file mode 100644 index 00000000..f6e076c4 --- /dev/null +++ b/paperforge/commands/annotation.py @@ -0,0 +1,265 @@ +"""paperforge annotation — CLI command family for PDF annotation management.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from paperforge import __version__ as PF_VERSION +from paperforge.annotation.db import get_annotations_db_path, get_annotations_connection +from paperforge.annotation.importer import run_import +from paperforge.annotation.probe import copy_db_to_temp, open_readonly, fetch_annotations +from paperforge.annotation.schema import ensure_schema, get_schema_version +from paperforge.annotation.service import ( + create_annotation, + delete_annotation, + export_annotations_json, + export_annotations_markdown, + get_annotation, + hard_delete, + list_annotations, + patch_annotation, +) +from paperforge.config import paperforge_paths +from paperforge.core.result import PFError, PFResult + + +def _db_path(vault: Path) -> Path: + return get_annotations_db_path(vault) + + +def _db_conn(vault: Path): + db_path = _db_path(vault) + conn = get_annotations_connection(db_path, read_only=False) + ensure_schema(conn) + return conn + + +def _probe_and_import( + vault: Path, + zotero_db: Path, + paper_key: str, + dry_run: bool, + no_copy: bool, +) -> dict: + """Core import pipeline: detect Zotero DB → probe → import.""" + probe_path = zotero_db + if not no_copy: + probe_path = copy_db_to_temp(zotero_db) + + probe_conn = open_readonly(probe_path) + try: + anns = fetch_annotations(probe_conn, limit=10000) + finally: + probe_conn.close() + + if paper_key: + anns = [a for a in anns if a.get("parentItemKey") == paper_key] + + if dry_run: + return {"paper_key": paper_key, "annotations_found": len(anns), "dry_run": True} + + conn = _db_conn(vault) + try: + result = run_import(conn, anns, source="zotero_db") + result["annotations_found"] = len(anns) + return result + finally: + conn.close() + + +def run(args: argparse.Namespace) -> int: + vault = args.vault_path + sub = getattr(args, "annotation_subcommand", "") + json_output = getattr(args, "json", False) + + if sub == "import": + return _cmd_import(vault, args, json_output) + elif sub == "list": + return _cmd_list(vault, args, json_output) + elif sub == "create": + return _cmd_create(vault, args, json_output) + elif sub == "patch": + return _cmd_patch(vault, args, json_output) + elif sub == "delete": + return _cmd_delete(vault, args, json_output) + elif sub == "export": + return _cmd_export(vault, args, json_output) + elif sub == "status": + return _cmd_status(vault, args, json_output) + else: + print(f"Unknown annotation subcommand: {sub}", file=sys.stderr) + return 1 + + +def _emit(result: PFResult, json_output: bool) -> int: + if json_output: + print(result.to_json()) + else: + data = result.data or {} + if result.ok: + for k, v in data.items(): + if isinstance(v, list): + print(f"{k}: {len(v)}") + else: + print(f"{k}: {v}") + else: + print(f"Error: {result.error.message}", file=sys.stderr) + return 0 if result.ok else 1 + + +def _cmd_import(vault, args, json_output): + paths = paperforge_paths(vault) + zotero_dir = paths.get("zotero_dir") + zotero_db_raw = getattr(args, "zotero_db", None) or str(zotero_dir / "zotero.sqlite") + zotero_db = Path(zotero_db_raw).expanduser().resolve() + + if not zotero_db.exists(): + result = PFResult( + ok=False, command="annotation import", version=PF_VERSION, + error=PFError(code="ZOTERO_DB_NOT_FOUND", message=f"Zotero SQLite not found: {zotero_db}"), + ) + return _emit(result, json_output) + + try: + data = _probe_and_import( + vault, zotero_db, + paper_key=getattr(args, "paper", ""), + dry_run=getattr(args, "dry_run", False), + no_copy=getattr(args, "no_copy", False), + ) + result = PFResult(ok=True, command="annotation import", version=PF_VERSION, data=data) + return _emit(result, json_output) + except Exception as exc: + result = PFResult( + ok=False, command="annotation import", version=PF_VERSION, + error=PFError(code="IMPORT_FAILED", message=str(exc)), + ) + return _emit(result, json_output) + + +def _cmd_list(vault, args, json_output): + conn = _db_conn(vault) + try: + anns = list_annotations( + conn, + paper_id=args.paper_key, + page_index=getattr(args, "page", None), + annotation_type=getattr(args, "ann_type", ""), + limit=getattr(args, "limit", 100), + ) + result = PFResult(ok=True, command="annotation list", version=PF_VERSION, data={ + "paper_id": args.paper_key, "annotations": anns, "count": len(anns), + }) + return _emit(result, json_output) + finally: + conn.close() + + +def _cmd_create(vault, args, json_output): + conn = _db_conn(vault) + try: + ann = create_annotation( + conn, + paper_id=args.paper, + annotation_type=args.ann_type, + page_index=getattr(args, "page_index", None), + page_label=getattr(args, "page_label", ""), + selected_text=getattr(args, "selected_text", ""), + comment=getattr(args, "comment", ""), + color=getattr(args, "color", "#ffd400"), + sort_index=getattr(args, "sort_index", ""), + ) + result = PFResult(ok=True, command="annotation create", version=PF_VERSION, data=ann) + return _emit(result, json_output) + except Exception as exc: + result = PFResult( + ok=False, command="annotation create", version=PF_VERSION, + error=PFError(code="CREATE_FAILED", message=str(exc)), + ) + return _emit(result, json_output) + + +def _cmd_patch(vault, args, json_output): + conn = _db_conn(vault) + try: + kwargs = {} + if getattr(args, "comment", None) is not None: + kwargs["comment"] = args.comment + if getattr(args, "color", None) is not None: + kwargs["color"] = args.color + ann = patch_annotation(conn, args.annotation_id, **kwargs) + result = PFResult(ok=True, command="annotation patch", version=PF_VERSION, data=ann) + return _emit(result, json_output) + except ValueError as exc: + result = PFResult( + ok=False, command="annotation patch", version=PF_VERSION, + error=PFError(code="PATCH_FAILED", message=str(exc)), + ) + return _emit(result, json_output) + + +def _cmd_delete(vault, args, json_output): + conn = _db_conn(vault) + try: + if getattr(args, "hard", False): + hard_delete(conn, args.annotation_id) + else: + delete_annotation(conn, args.annotation_id) + result = PFResult(ok=True, command="annotation delete", version=PF_VERSION, data={"id": args.annotation_id}) + return _emit(result, json_output) + except ValueError as exc: + result = PFResult( + ok=False, command="annotation delete", version=PF_VERSION, + error=PFError(code="DELETE_FAILED", message=str(exc)), + ) + return _emit(result, json_output) + + +def _cmd_export(vault, args, json_output): + conn = _db_conn(vault) + try: + fmt = getattr(args, "format", "json") + paper_key = args.paper_key + if fmt == "markdown": + text = export_annotations_markdown(conn, paper_id=paper_key) + else: + text = export_annotations_json(conn, paper_id=paper_key) + + output_path = getattr(args, "output", None) + if output_path: + Path(output_path).write_text(text, encoding="utf-8") + result = PFResult(ok=True, command="annotation export", version=PF_VERSION, data={"path": output_path}) + else: + if json_output: + print(text) + else: + result = PFResult(ok=True, command="annotation export", version=PF_VERSION, data={"text": text}) + return _emit(result, json_output) + except Exception as exc: + result = PFResult( + ok=False, command="annotation export", version=PF_VERSION, + error=PFError(code="EXPORT_FAILED", message=str(exc)), + ) + return _emit(result, json_output) + + +def _cmd_status(vault, args, json_output): + db_path = _db_path(vault) + data = { + "db_exists": db_path.exists(), + "db_path": str(db_path), + } + if db_path.exists(): + conn = _db_conn(vault) + try: + count = conn.execute("SELECT COUNT(*) as cnt FROM annotations WHERE deleted_at IS NULL").fetchone()["cnt"] + schema_ver = get_schema_version(conn) + data["annotation_count"] = count + data["schema_version"] = schema_ver + finally: + conn.close() + result = PFResult(ok=True, command="annotation status", version=PF_VERSION, data=data) + return _emit(result, json_output) diff --git a/tests/cli/test_annotation_commands.py b/tests/cli/test_annotation_commands.py new file mode 100644 index 00000000..a1aa6a73 --- /dev/null +++ b/tests/cli/test_annotation_commands.py @@ -0,0 +1,153 @@ +"""CLI parser and JSON contract tests for annotation commands.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + + +@pytest.fixture +def vault(tmp_path): + """Minimal vault with paperforge.json.""" + pf_json = tmp_path / "paperforge.json" + pf_json.write_text('{"vault_config": {"system_dir": "99_System"}}', encoding="utf-8") + return tmp_path + + +def _run_cli(vault: Path, args: list[str]): + """Parse annotation subcommand args with vault_path injection.""" + import argparse + from paperforge.cli import build_parser + + parser = build_parser() + full_args = ["annotation", *args] + ns = parser.parse_args(full_args) + ns.vault_path = vault + return ns + + +class TestParser: + """Unit tests for argparse parsing only (no DB).""" + + def test_import(self, vault): + ns = _run_cli(vault, ["import", "--dry-run"]) + assert ns.command == "annotation" + assert ns.annotation_subcommand == "import" + assert ns.dry_run is True + + def test_list(self, vault): + ns = _run_cli(vault, ["list", "PAPER001", "--page", "2", "--json"]) + assert ns.annotation_subcommand == "list" + assert ns.paper_key == "PAPER001" + assert ns.page == 2 + assert ns.json is True + + def test_create(self, vault): + ns = _run_cli(vault, [ + "create", "--paper", "PAPER001", "--type", "highlight", + "--page-index", "3", "--selected-text", "foo", "--comment", "bar", + ]) + assert ns.annotation_subcommand == "create" + assert ns.paper == "PAPER001" + assert ns.ann_type == "highlight" + assert ns.page_index == 3 + + def test_patch(self, vault): + ns = _run_cli(vault, ["patch", "ann123", "--comment", "new"]) + assert ns.annotation_subcommand == "patch" + assert ns.annotation_id == "ann123" + assert ns.comment == "new" + + def test_delete(self, vault): + ns = _run_cli(vault, ["delete", "ann123", "--hard"]) + assert ns.annotation_subcommand == "delete" + assert ns.annotation_id == "ann123" + assert ns.hard is True + + def test_export(self, vault): + ns = _run_cli(vault, ["export", "PAPER001", "--format", "markdown"]) + assert ns.annotation_subcommand == "export" + assert ns.paper_key == "PAPER001" + + def test_status(self, vault): + ns = _run_cli(vault, ["status", "--json"]) + assert ns.annotation_subcommand == "status" + + def test_import_zotero_db_flag(self, vault): + ns = _run_cli(vault, ["import", "--zotero-db", "/custom/path.sqlite"]) + assert ns.zotero_db == "/custom/path.sqlite" + + +class TestIntegration: + """Tests that exercise commands against a real vault + annotations.db.""" + + def test_status_json_envelope(self, vault): + from paperforge.commands.annotation import run as ann_run + from paperforge.cli import build_parser + + parser = build_parser() + ns = parser.parse_args(["annotation", "status", "--json"]) + ns.vault_path = vault + + import io, sys + captured = io.StringIO() + old_stdout = sys.stdout + sys.stdout = captured + try: + exit_code = ann_run(ns) + finally: + sys.stdout = old_stdout + + assert exit_code == 0 + output = json.loads(captured.getvalue()) + assert output["ok"] is True + assert output["command"] == "annotation status" + assert "data" in output + assert output["data"]["db_exists"] is False + + def test_create_and_list(self, vault): + """Create an annotation then list it.""" + from paperforge.commands.annotation import run as ann_run + from paperforge.cli import build_parser + + def _run(sub_args: list[str]): + parser = build_parser() + ns = parser.parse_args(["annotation", *sub_args]) + ns.vault_path = vault + import io, sys + captured = io.StringIO() + old_stdout = sys.stdout + sys.stdout = captured + try: + code = ann_run(ns) + finally: + sys.stdout = old_stdout + return code, captured.getvalue() + + # Create + code, out = _run([ + "create", "--paper", "PAPER001", "--type", "highlight", + "--page-index", "2", "--selected-text", "hello", "--json", + ]) + assert code == 0 + created = json.loads(out) + assert created["data"]["paper_id"] == "PAPER001" + ann_id = created["data"]["id"] + + # List + code, out = _run(["list", "PAPER001", "--json"]) + assert code == 0 + listed = json.loads(out) + assert listed["data"]["count"] >= 1 + + # Patch + code, out = _run(["patch", ann_id, "--comment", "updated", "--json"]) + assert code == 0 + patched = json.loads(out) + assert patched["data"]["comment"] == "updated" + + # Delete + code, out = _run(["delete", ann_id, "--json"]) + assert code == 0