feat(cli): add memory build/status and paper-status commands

This commit is contained in:
Research Assistant 2026-05-12 17:52:30 +08:00
parent c100423353
commit 95bf0b33c9
4 changed files with 185 additions and 0 deletions

View file

@ -258,6 +258,18 @@ def build_parser() -> argparse.ArgumentParser:
p_dash = sub.add_parser("dashboard", help="Aggregated stats and permissions for the plugin dashboard")
p_dash.add_argument("--json", action="store_true", help="Output as PFResult JSON")
# Memory Layer commands
p_memory = sub.add_parser("memory", help="Manage the Memory Layer")
p_memory_sp = p_memory.add_subparsers(dest="memory_subcommand", required=True)
p_memory_build = p_memory_sp.add_parser("build", help="Build the memory database from canonical index")
p_memory_build.add_argument("--json", action="store_true", help="Output as JSON")
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")
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")
# base-refresh
p_base = sub.add_parser("base-refresh", help="Refresh Obsidian Base view files")
p_base.add_argument(
@ -470,6 +482,16 @@ def main(argv: list[str] | None = None) -> int:
return dashboard.run(args)
if args.command == "memory":
from paperforge.commands.memory import run
return run(args)
if args.command == "paper-status":
from paperforge.commands.paper_status import run
return run(args)
if args.command == "base-refresh":
force = getattr(args, "force", False)
paths = args.paths

View file

@ -10,6 +10,8 @@ _COMMAND_REGISTRY: dict[str, str] = {
"context": "paperforge.commands.context",
"dashboard": "paperforge.commands.dashboard",
"finalize": "paperforge.commands.finalize",
"memory": "paperforge.commands.memory",
"paper-status": "paperforge.commands.paper_status",
}

View file

@ -0,0 +1,91 @@
from __future__ import annotations
import argparse
import sys
from paperforge import __version__ as PF_VERSION
from paperforge.core.errors import ErrorCode
from paperforge.core.result import PFError, PFResult
from paperforge.memory.builder import build_from_index
from paperforge.memory.query import get_memory_status
def run(args: argparse.Namespace) -> int:
vault = args.vault_path
sub_cmd = args.memory_subcommand
if sub_cmd == "build":
try:
counts = build_from_index(vault)
result = PFResult(
ok=True,
command="memory build",
version=PF_VERSION,
data=counts,
)
except FileNotFoundError:
result = PFResult(
ok=False,
command="memory build",
version=PF_VERSION,
error=PFError(
code=ErrorCode.PATH_NOT_FOUND,
message="Canonical index not found. Run paperforge sync --rebuild-index.",
),
next_actions=[
{
"command": "paperforge sync --rebuild-index",
"reason": "Generate formal-library.json first",
}
],
)
except Exception as exc:
result = PFResult(
ok=False,
command="memory build",
version=PF_VERSION,
error=PFError(
code=ErrorCode.INTERNAL_ERROR,
message=str(exc),
),
)
if args.json:
print(result.to_json())
else:
if result.ok:
print(f"Memory built: {result.data}")
else:
print(f"Error: {result.error.message}", file=sys.stderr)
return 0 if result.ok else 1
if sub_cmd == "status":
try:
status = get_memory_status(vault)
result = PFResult(
ok=True,
command="memory status",
version=PF_VERSION,
data=status,
)
except Exception as exc:
result = PFResult(
ok=False,
command="memory status",
version=PF_VERSION,
error=PFError(
code=ErrorCode.INTERNAL_ERROR,
message=str(exc),
),
)
if args.json:
print(result.to_json())
else:
if result.ok:
for k, v in status.items():
print(f" {k}: {v}")
else:
print(f"Error: {result.error.message}", file=sys.stderr)
return 0 if result.ok else 1
print(f"Unknown memory subcommand: {sub_cmd}", file=sys.stderr)
return 1

View file

@ -0,0 +1,70 @@
from __future__ import annotations
import argparse
import sys
from paperforge import __version__ as PF_VERSION
from paperforge.core.errors import ErrorCode
from paperforge.core.result import PFError, PFResult
from paperforge.memory.query import get_paper_status
def run(args: argparse.Namespace) -> int:
vault = args.vault_path
query = args.query
try:
status = get_paper_status(vault, query)
if status is None:
result = PFResult(
ok=False,
command="paper-status",
version=PF_VERSION,
error=PFError(
code=ErrorCode.PATH_NOT_FOUND,
message=f"No paper found for: {query}",
),
next_actions=[
{
"command": "paperforge search",
"reason": "Search for papers by keyword",
}
],
)
else:
result = PFResult(
ok=True,
command="paper-status",
version=PF_VERSION,
data=status,
)
except Exception as exc:
result = PFResult(
ok=False,
command="paper-status",
version=PF_VERSION,
error=PFError(
code=ErrorCode.INTERNAL_ERROR,
message=str(exc),
),
)
if args.json:
print(result.to_json())
else:
if result.ok:
data = result.data
if data.get("resolved"):
print(f"Zotero Key: {data.get('zotero_key', '')}")
print(f"Title: {data.get('title', '')}")
print(f"Year: {data.get('year', '')}")
print(f"Lifecycle: {data.get('lifecycle', '')}")
print(f"Next Step: {data.get('next_step', '')}")
if data.get("candidates"):
print(f"\nMultiple candidates: {len(data['candidates'])}")
for c in data["candidates"]:
print(f" - {c['zotero_key']}: {c['title']} ({c['year']})")
else:
print(f"Error: {result.error.message}", file=sys.stderr)
return 0 if result.ok else 1