mirror of
https://github.com/lllin000/PaperForge.git
synced 2026-07-22 06:50:53 +00:00
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
This commit is contained in:
Research Assistant
parent
f880d50092
commit
d2bfed9d12
3 changed files with 472 additions and 0 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
265
paperforge/commands/annotation.py
Normal file
265
paperforge/commands/annotation.py
Normal file
|
|
@ -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)
|
||||
153
tests/cli/test_annotation_commands.py
Normal file
153
tests/cli/test_annotation_commands.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue