feat: MemoryContract CLI——load-context + 5个按需查询命令

webnovel-writer/scripts/data_modules/tests/test_memory_cli.py

@@ -0,0 +1,169 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""memory_cli.py 测试。"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+_scripts_dir = str(Path(__file__).resolve().parent.parent.parent)
+if _scripts_dir not in sys.path:
+    sys.path.insert(0, _scripts_dir)
+
+
+def _ensure_scripts_on_path():
+    scripts_dir = Path(__file__).resolve().parent.parent.parent
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+
+def _make_project(tmp_path: Path):
+    webnovel_dir = tmp_path / ".webnovel"
+    webnovel_dir.mkdir(parents=True, exist_ok=True)
+    (webnovel_dir / "state.json").write_text("{}", encoding="utf-8")
+    (webnovel_dir / "summaries").mkdir(exist_ok=True)
+    return tmp_path
+
+
+def test_load_context_cli(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "load-context", "--chapter", "1"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output["chapter"] == 1
+    assert "sections" in output
+
+
+def test_query_entity_not_found(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "query-entity", "--id", "nobody"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output["error"] == "not_found"
+
+
+def test_query_entity_found(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    state = {
+        "entities_v3": {
+            "角色": {
+                "xiaoyan": {"name": "萧炎", "tier": "核心", "aliases": [], "first_appearance": 1, "last_appearance": 10}
+            }
+        }
+    }
+    (project / ".webnovel" / "state.json").write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
+
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "query-entity", "--id", "xiaoyan"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output["name"] == "萧炎"
+
+
+def test_query_rules_empty(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "query-rules"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output == []
+
+
+def test_read_summary_missing(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "read-summary", "--chapter", "99"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output["chapter"] == 99
+    assert output["summary"] == ""
+
+
+def test_read_summary_exists(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    (project / ".webnovel" / "summaries" / "ch0005.md").write_text("第5章摘要", encoding="utf-8")
+
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "read-summary", "--chapter", "5"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert "第5章摘要" in output["summary"]
+
+
+def test_get_open_loops_empty(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "get-open-loops"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output == []
+
+
+def test_get_timeline_empty(tmp_path, capsys):
+    _ensure_scripts_on_path()
+    import memory_cli
+
+    project = _make_project(tmp_path)
+    old_argv = sys.argv
+    sys.argv = ["memory_cli", "--project-root", str(project), "get-timeline", "--from", "1", "--to", "100"]
+    try:
+        memory_cli.main()
+    finally:
+        sys.argv = old_argv
+
+    output = json.loads(capsys.readouterr().out)
+    assert output == []

webnovel-writer/scripts/data_modules/webnovel.py

@@ -249,6 +249,9 @@ def main() -> None:
     p_extract_context.add_argument("--chapter", type=int, required=True, help="目标章节号")
     p_extract_context.add_argument("--format", choices=["text", "json"], default="text", help="输出格式")
 
+    p_memory_contract = sub.add_parser("memory-contract", help="转发到 memory_cli.py")
+    p_memory_contract.add_argument("args", nargs=argparse.REMAINDER)
+
     p_review_pipeline = sub.add_parser("review-pipeline", help="转发到 review_pipeline.py")
     p_review_pipeline.add_argument("--chapter", type=int, required=True, help="目标章节号")
     p_review_pipeline.add_argument("--review-results", required=True, help="reviewer 原始结果 JSON 文件")
@@ -309,6 +312,8 @@ def main() -> None:
     if tool == "extract-context":
         return_args = [*forward_args, "--chapter", str(args.chapter), "--format", str(args.format)]
         raise SystemExit(_run_script("extract_chapter_context.py", return_args))
+    if tool == "memory-contract":
+        raise SystemExit(_run_script("memory_cli.py", [*forward_args, *rest]))
     if tool == "review-pipeline":
         return_args = [
             *forward_args,

webnovel-writer/scripts/memory_cli.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+memory_cli.py — MemoryContract CLI 入口。
+
+提供 load-context / query-entity / query-rules / read-summary /
+get-open-loops / get-timeline 六个子命令，输出 JSON。
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from runtime_compat import enable_windows_utf8_stdio
+
+
+def _ensure_scripts_path() -> None:
+    scripts_dir = Path(__file__).resolve().parent
+    if str(scripts_dir) not in sys.path:
+        sys.path.insert(0, str(scripts_dir))
+
+
+_ensure_scripts_path()
+
+from data_modules.config import DataModulesConfig
+from data_modules.memory_contract_adapter import MemoryContractAdapter
+
+
+def _adapter(project_root: str) -> MemoryContractAdapter:
+    cfg = DataModulesConfig.from_project_root(project_root)
+    return MemoryContractAdapter(cfg)
+
+
+def _json_out(data) -> None:
+    print(json.dumps(data, ensure_ascii=False, indent=2))
+
+
+def cmd_load_context(args: argparse.Namespace) -> None:
+    adapter = _adapter(args.project_root)
+    pack = adapter.load_context(args.chapter)
+    _json_out(pack.to_dict())
+
+
+def cmd_query_entity(args: argparse.Namespace) -> None:
+    adapter = _adapter(args.project_root)
+    snap = adapter.query_entity(args.id)
+    if snap is None:
+        _json_out({"error": "not_found", "entity_id": args.id})
+    else:
+        _json_out(snap.to_dict())
+
+
+def cmd_query_rules(args: argparse.Namespace) -> None:
+    adapter = _adapter(args.project_root)
+    rules = adapter.query_rules(domain=args.domain or "")
+    _json_out([r.to_dict() for r in rules])
+
+
+def cmd_read_summary(args: argparse.Namespace) -> None:
+    adapter = _adapter(args.project_root)
+    text = adapter.read_summary(args.chapter)
+    _json_out({"chapter": args.chapter, "summary": text})
+
+
+def cmd_get_open_loops(args: argparse.Namespace) -> None:
+    adapter = _adapter(args.project_root)
+    loops = adapter.get_open_loops(status=args.status or "active")
+    _json_out([l.to_dict() for l in loops])
+
+
+def cmd_get_timeline(args: argparse.Namespace) -> None:
+    adapter = _adapter(args.project_root)
+    events = adapter.get_timeline(args.from_ch, args.to_ch)
+    _json_out([e.to_dict() for e in events])
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="MemoryContract CLI")
+    parser.add_argument("--project-root", required=True, help="项目根目录")
+    sub = parser.add_subparsers(dest="command")
+
+    p_load = sub.add_parser("load-context", help="加载章节上下文基础包")
+    p_load.add_argument("--chapter", type=int, required=True)
+
+    p_entity = sub.add_parser("query-entity", help="查询实体快照")
+    p_entity.add_argument("--id", required=True, help="实体 ID")
+
+    p_rules = sub.add_parser("query-rules", help="查询世界规则")
+    p_rules.add_argument("--domain", default="", help="按 domain 过滤")
+
+    p_summary = sub.add_parser("read-summary", help="读取章节摘要")
+    p_summary.add_argument("--chapter", type=int, required=True)
+
+    p_loops = sub.add_parser("get-open-loops", help="查询未闭合伏笔")
+    p_loops.add_argument("--status", default="active", help="状态过滤")
+
+    p_timeline = sub.add_parser("get-timeline", help="查询时间线事件")
+    p_timeline.add_argument("--from", type=int, required=True, dest="from_ch", help="起始章节")
+    p_timeline.add_argument("--to", type=int, required=True, dest="to_ch", help="结束章节")
+
+    args = parser.parse_args()
+    if not args.command:
+        parser.print_help()
+        sys.exit(1)
+
+    dispatch = {
+        "load-context": cmd_load_context,
+        "query-entity": cmd_query_entity,
+        "query-rules": cmd_query_rules,
+        "read-summary": cmd_read_summary,
+        "get-open-loops": cmd_get_open_loops,
+        "get-timeline": cmd_get_timeline,
+    }
+    dispatch[args.command](args)
+
+
+if __name__ == "__main__":
+    if sys.platform == "win32":
+        enable_windows_utf8_stdio()
+    main()