diff --git a/.gitignore b/.gitignore index 1db3e8209..cfaff7127 100644 --- a/.gitignore +++ b/.gitignore @@ -100,6 +100,7 @@ !/3MF/ !/calibre/ !/rekordbox/ +!/siyuan/ # Step 5: Inside each software dir, ignore everything (including dotfiles) /gimp/* /gimp/.* @@ -207,6 +208,8 @@ /calibre/.* /rekordbox/* /rekordbox/.* +/siyuan/* +/siyuan/.* # Step 6: ...except agent-harness/ !/gimp/agent-harness/ @@ -272,6 +275,7 @@ !/quietshrink/agent-harness/ !/mailchimp/agent-harness/ !/rekordbox/agent-harness/ +!/siyuan/agent-harness/ # Exclude non-gedit demo macros from macrocli (local only) /macrocli/agent-harness/cli_anything/macrocli/macro_definitions/demo/flameshot* diff --git a/registry.json b/registry.json index 7ab1bc345..1e3eeca3e 100644 --- a/registry.json +++ b/registry.json @@ -1303,6 +1303,26 @@ "url": "https://github.com/octo-patch" } ] + }, + { + "name": "siyuan", + "display_name": "SiYuan CLI", + "version": "1.0.0", + "description": "Knowledge management and note-taking via SiYuan HTTP API — manage notebooks, documents, blocks, and search content from the command line", + "category": "knowledge-management", + "requires": "SiYuan running with HTTP API enabled (default http://127.0.0.1:6806)", + "homepage": "https://github.com/siyuan-note/siyuan", + "source_url": null, + "install_strategy": "pip", + "install_cmd": "pip install git+https://github.com/HKUDS/CLI-Anything.git#subdirectory=siyuan/agent-harness", + "skill_md": "skills/cli-anything-siyuan/SKILL.md", + "entry_point": "cli-anything-siyuan", + "contributors": [ + { + "name": "hito0512", + "url": "https://github.com/hito0512" + } + ] } ] } diff --git a/siyuan/agent-harness/SIYUAN.md b/siyuan/agent-harness/SIYUAN.md new file mode 100644 index 000000000..0cc07d2cd --- /dev/null +++ b/siyuan/agent-harness/SIYUAN.md @@ -0,0 +1,159 @@ +# SiYuan (思源笔记) — Agent CLI Harness Analysis + +## Overview + +SiYuan is a local-first, privacy-focused knowledge management and note-taking application. +It uses a Go backend with an Electron/TypeScript frontend. Data is stored as `.sy` files +(Siyuan JSON format) with a SQLite database for indexing and search. + +## Backend + +- **Language**: Go +- **HTTP Server**: Gin framework, runs on `http://127.0.0.1:6806` +- **Auth**: Token-based (`Authorization: Token xxx`) +- **Entry point**: `kernel/main.go` → starts HTTP server + background jobs + +## Data Model + +``` +Notebook (Box) + └── Document (tree node, .sy file) + └── Blocks (paragraphs, headings, lists, code, etc.) + └── Attributes (key-value, custom-* prefix) +``` + +- **Notebook**: Top-level container, identified by ID (e.g., `20210817205410-2kvfpfn`) +- **Document**: Tree node in a notebook, stored as `.sy` file +- **Block**: Content unit (paragraph, heading, list item, etc.), each has unique ID +- **SQLite Database**: blocks, refs, attributes, history, asset_content tables +- **IDs**: Timestamp-based IDs like `20210817205410-2kvfpfn` (datetime + random suffix) + +## API Surface + +Base URL: `http://127.0.0.1:6806` + +### Notebook API (`/api/notebook/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `lsNotebooks` | POST | List all notebooks | +| `openNotebook` | POST | Open a notebook | +| `closeNotebook` | POST | Close a notebook | +| `createNotebook` | POST | Create new notebook | +| `removeNotebook` | POST | Delete notebook | +| `renameNotebook` | POST | Rename notebook | +| `getNotebookConf` | POST | Get notebook config | +| `setNotebookConf` | POST | Set notebook config | +| `setNotebookIcon` | POST | Set notebook icon | + +### Document/Filetree API (`/api/filetree/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `createDocWithMd` | POST | Create doc with Markdown content | +| `createDailyNote` | POST | Create daily note | +| `renameDoc` | POST | Rename doc by path | +| `renameDocByID` | POST | Rename doc by ID | +| `removeDoc` | POST | Delete doc by path | +| `removeDocByID` | POST | Delete doc by ID | +| `moveDocs` | POST | Move docs by paths | +| `moveDocsByID` | POST | Move docs by IDs | +| `getHPathByID` | POST | Get human-readable path by ID | +| `getHPathByPath` | POST | Get human-readable path by path | +| `getIDsByHPath` | POST | Get IDs by human-readable path | +| `getPathByID` | POST | Get storage path by ID | +| `listDocsByPath` | POST | List docs at a path | +| `listDocTree` | POST | List full document tree | +| `searchDocs` | POST | Search document names | + +### Block API (`/api/block/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `insertBlock` | POST | Insert block at position | +| `prependBlock` | POST | Insert as first child | +| `appendBlock` | POST | Insert as last child | +| `updateBlock` | POST | Update block content | +| `deleteBlock` | POST | Delete a block | +| `moveBlock` | POST | Move a block | +| `foldBlock` | POST | Fold/collapse a block | +| `unfoldBlock` | POST | Unfold/expand a block | +| `getBlockKramdown` | POST | Get block kramdown source | +| `getChildBlocks` | POST | Get child blocks | + +### Attribute API (`/api/attr/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `setBlockAttrs` | POST | Set block attributes | +| `getBlockAttrs` | POST | Get block attributes | + +### SQL Query API (`/api/query/sql`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `sql` | POST | Execute SQL query on block database | + +### Search API (`/api/search/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `fullTextSearchBlock` | POST | Full-text search in blocks | +| `searchTag` | POST | Search tags | +| `findReplace` | POST | Find and replace | + +### Export API (`/api/export/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `exportMdContent` | POST | Export doc as Markdown | +| `exportResources` | POST | Export files as ZIP | + +### Asset API (`/api/asset/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `upload` | POST | Upload asset file | + +### Tag API (`/api/tag/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `getTag` | POST | List tags | +| `renameTag` | POST | Rename tag | +| `removeTag` | POST | Remove tag | + +### System API (`/api/system/*`) +| Endpoint | Method | Description | +|----------|--------|-------------| +| `version` | GET/POST | Get system version | +| `currentTime` | POST | Get current server time | + +## Existing CLI Tools + +None. SiYuan ships with only the desktop GUI (Electron) and mobile apps. +The kernel starts automatically with the GUI; there is no standalone CLI mode. + +## CLI Harness Strategy + +Since SiYuan has no headless CLI backend, the harness will: + +1. **Connect to an already-running SiYuan instance** via its HTTP API +2. **Provide commands for all major API groups** (notebooks, documents, blocks, search, export) +3. **Store connection state** (host, port, token) for multi-session use +4. **Support SQL queries** for advanced data access +5. **Support import/export** of Markdown content + +The SiYuan instance must be running with the API server enabled (default: port 6806). +The API token can be found in Settings → About. + +## CLI Usage Notes + +### Block insert/update with multi-line content + +`block insert` and `block update` accept data via stdin pipe when the argument is `-` or omitted: + +```bash +# Pipe multi-line content +cat note.md | cli-anything-siyuan block insert --parent + +# Explicit stdin marker +echo "hello" | cli-anything-siyuan block update - + +# Requires one of: --parent , --previous , --next +``` + +### Tag list + +`tag list` sends `ignoreMaxListHint: true` to ensure all tags are returned regardless of `Conf.FileTree.MaxListCount`. Tag-heavy workspaces return the complete list. diff --git a/siyuan/agent-harness/cli_anything/siyuan/README.md b/siyuan/agent-harness/cli_anything/siyuan/README.md new file mode 100644 index 000000000..323d64471 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/README.md @@ -0,0 +1,143 @@ +# cli-anything-siyuan + +A CLI harness for [SiYuan](https://github.com/siyuan-note/siyuan) (思源笔记) — +interact with your knowledge base from the terminal. + +## Prerequisites + +- **SiYuan** must be running (version 3.x). Download from [b3log.org/siyuan](https://b3log.org/siyuan). +- **Python 3.10+** + +## Installation + +```bash +cd agent-harness +pip install -e . +``` + +For REPL support (recommended): + +```bash +pip install -e ".[repl]" +``` + +## Configuration + +Create `~/.siyuan-cli.json`: + +```json +{ + "host": "127.0.0.1", + "port": 6806, + "token": "your-api-token-here" +} +``` + +Or use environment variables: + +```bash +export SIYUAN_HOST=127.0.0.1 +export SIYUAN_PORT=6806 +export SIYUAN_TOKEN=your-token +``` + +**Finding your API token:** Open SiYuan → Settings → About → API Token. + +## Usage + +### One-shot commands + +```bash +# List notebooks +cli-anything-siyuan notebook list + +# List notebooks (JSON output) +cli-anything-siyuan --json notebook list + +# Get version +cli-anything-siyuan version + +# Execute SQL query +cli-anything-siyuan sql "SELECT * FROM blocks LIMIT 5" + +# Search blocks +cli-anything-siyuan search "keyword" + +# Export document as Markdown +cli-anything-siyuan export md + +# Show status +cli-anything-siyuan status + +# Insert a block (multi-line content via stdin) +cat note.md | cli-anything-siyuan block insert --parent + +# Update a block content via stdin +echo "new content" | cli-anything-siyuan block update +``` + +### REPL mode + +```bash +# Just run without arguments to enter REPL +cli-anything-siyuan +``` + +Inside the REPL: + +``` +◆ cli-anything · Siyuan v1.0.0 + ◇ Install: npx skills add HKUDS/CLI-Anything --skill cli-anything-siyuan -g -y + ◇ Global skill: ~/.agents/skills/cli-anything-siyuan/SKILL.md + + Type help for commands, quit to exit + +siyuan ❯ notebook list +siyuan ❯ doc tree +siyuan ❯ search "meeting notes" +siyuan ❯ help +siyuan ❯ quit +``` + +## Command Groups + +| Command | Description | +|---------|-------------| +| `notebook list` | List all notebooks | +| `notebook create ` | Create a notebook | +| `notebook rename ` | Rename a notebook | +| `notebook remove ` | Delete a notebook | +| `notebook open ` | Open a notebook | +| `doc create ` | Create a document | +| `doc list [path]` | List documents | +| `doc tree ` | Show document tree | +| `doc get ` | Get document path by ID | +| `doc rename ` | Rename a document | +| `doc remove <id>` | Delete a document | +| `block insert <data>` | Insert a block (use `-` or omit for stdin pipe) | +| `block update <id> <data>` | Update block content (use `-` or omit for stdin pipe) | +| `block delete <id>` | Delete a block | +| `block get <id>` | Get block kramdown source | +| `block children <id>` | Get child blocks | +| `sql <stmt>` | Execute SQL query | +| `search <query>` | Full-text search | +| `export md <doc-id>` | Export as Markdown | +| `tag list` | List all tags | +| `version` | Show SiYuan version | +| `status` | Show connection status | + +## Running Tests + +```bash +# Unit tests (no external dependencies) +cd agent-harness +pip install -e ".[test]" +python -m pytest cli_anything/siyuan/tests/test_core.py -v + +# E2E tests (requires running SiYuan) +python -m pytest cli_anything/siyuan/tests/test_full_e2e.py -v -s +``` + +## License + +AGPL-3.0 (same as SiYuan) diff --git a/siyuan/agent-harness/cli_anything/siyuan/__init__.py b/siyuan/agent-harness/cli_anything/siyuan/__init__.py new file mode 100644 index 000000000..448d0013b --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/__init__.py @@ -0,0 +1 @@ +# cli-anything-siyuan: SiYuan (思源笔记) CLI harness diff --git a/siyuan/agent-harness/cli_anything/siyuan/__main__.py b/siyuan/agent-harness/cli_anything/siyuan/__main__.py new file mode 100644 index 000000000..fc05ffd3a --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/__main__.py @@ -0,0 +1,5 @@ +"""Allow running as: python -m cli_anything.siyuan""" +from cli_anything.siyuan.siyuan_cli import cli + +if __name__ == "__main__": + cli() diff --git a/siyuan/agent-harness/cli_anything/siyuan/core/__init__.py b/siyuan/agent-harness/cli_anything/siyuan/core/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/siyuan/agent-harness/cli_anything/siyuan/core/client.py b/siyuan/agent-harness/cli_anything/siyuan/core/client.py new file mode 100644 index 000000000..f274d8810 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/core/client.py @@ -0,0 +1,307 @@ +"""SiYuan HTTP API client. + +Handles connection, authentication, and request/response to the SiYuan kernel. +""" + +import json +import os +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +import requests + + +@dataclass(frozen=True) +class SiYuanConfig: + host: str = "127.0.0.1" + port: int = 6806 + token: str = "" + + @property + def base_url(self) -> str: + return f"http://{self.host}:{self.port}" + + +def load_config(config_path: str | None = None) -> SiYuanConfig: + """Load SiYuan connection config from file or environment. + + Priority: explicit path -> env vars -> defaults. + Config file is JSON: {"host": "...", "port": 6806, "token": "..."} + """ + if config_path: + config_file = Path(config_path) + else: + config_file = Path.home() / ".siyuan-cli.json" + + if config_file.is_file(): + try: + data = json.loads(config_file.read_text(encoding="utf-8-sig")) + except (json.JSONDecodeError, OSError): + data = None + + if data is not None: + # File values as base, env vars override + return SiYuanConfig( + host=os.environ.get("SIYUAN_HOST", data.get("host", "127.0.0.1")), + port=int(os.environ.get("SIYUAN_PORT", data.get("port", 6806))), + token=os.environ.get("SIYUAN_TOKEN", data.get("token", "")), + ) + + return SiYuanConfig( + host=os.environ.get("SIYUAN_HOST", "127.0.0.1"), + port=int(os.environ.get("SIYUAN_PORT", "6806")), + token=os.environ.get("SIYUAN_TOKEN", ""), + ) + + +class SiYuanClientError(Exception): + """Raised when the SiYuan API returns an error.""" + + +class SiYuanClient: + """HTTP client for the SiYuan kernel API.""" + + def __init__(self, config: SiYuanConfig | None = None): + self.config = config or load_config() + self._session = requests.Session() + self._session.trust_env = False + if self.config.token: + self._session.headers["Authorization"] = f"Token {self.config.token}" + self._session.headers["Content-Type"] = "application/json" + + def _post(self, endpoint: str, data: dict[str, Any] | None = None) -> Any: + """Make a POST request to the SiYuan API.""" + url = f"{self.config.base_url}{endpoint}" + try: + resp = self._session.post(url, json=data or {}, timeout=30) + except requests.ConnectionError as e: + raise SiYuanClientError( + f"Cannot connect to SiYuan at {self.config.base_url}. " + f"Is SiYuan running? ({e})" + ) from e + except requests.Timeout as e: + raise SiYuanClientError( + f"Request to SiYuan timed out after 30s ({e})" + ) from e + except requests.RequestException as e: + raise SiYuanClientError( + f"Request to SiYuan failed: {e}" + ) from e + + if resp.status_code != 200: + raise SiYuanClientError( + f"API returned status {resp.status_code}: {resp.text[:200]}" + ) + + body = resp.json() + if body.get("code", 0) != 0: + raise SiYuanClientError( + f"API error: {body.get('msg', 'unknown error')}" + ) + return body.get("data") + + def ping(self) -> bool: + """Check if SiYuan kernel is reachable.""" + try: + return self._post("/api/system/version") is not None + except SiYuanClientError: + return False + + # ── Notebook API ─────────────────────────────────────────────────── + + def list_notebooks(self) -> list[dict[str, Any]]: + """List all notebooks.""" + data = self._post("/api/notebook/lsNotebooks") + return data.get("notebooks", []) + + def open_notebook(self, notebook_id: str) -> None: + self._post("/api/notebook/openNotebook", {"notebook": notebook_id}) + + def close_notebook(self, notebook_id: str) -> None: + self._post("/api/notebook/closeNotebook", {"notebook": notebook_id}) + + def create_notebook(self, name: str) -> dict[str, Any]: + data = self._post("/api/notebook/createNotebook", {"name": name}) + if isinstance(data, dict) and "notebook" in data: + return data["notebook"] + return data + + def remove_notebook(self, notebook_id: str) -> None: + self._post("/api/notebook/removeNotebook", {"notebook": notebook_id}) + + def rename_notebook(self, notebook_id: str, name: str) -> None: + self._post("/api/notebook/renameNotebook", {"notebook": notebook_id, "name": name}) + + def get_notebook_conf(self, notebook_id: str) -> dict[str, Any]: + return self._post("/api/notebook/getNotebookConf", {"notebook": notebook_id}) + + def set_notebook_conf(self, notebook_id: str, conf: dict[str, Any]) -> dict[str, Any]: + return self._post("/api/notebook/setNotebookConf", {"notebook": notebook_id, "conf": conf}) + + def set_notebook_icon(self, notebook_id: str, icon: str) -> None: + self._post("/api/notebook/setNotebookIcon", {"notebook": notebook_id, "icon": icon}) + + def get_notebook_info(self, notebook_id: str) -> dict[str, Any]: + return self._post("/api/notebook/getNotebookInfo", {"notebook": notebook_id}) + + # ── Document / Filetree API ──────────────────────────────────────── + + def create_doc_with_md(self, notebook_id: str, path: str, markdown: str = "") -> str: + """Create a document with Markdown content. Returns the doc ID.""" + return self._post("/api/filetree/createDocWithMd", { + "notebook": notebook_id, "path": path, "markdown": markdown, + }) + + def rename_doc(self, notebook_id: str, path: str, title: str) -> None: + self._post("/api/filetree/renameDoc", {"notebook": notebook_id, "path": path, "title": title}) + + def rename_doc_by_id(self, doc_id: str, title: str) -> None: + self._post("/api/filetree/renameDocByID", {"id": doc_id, "title": title}) + + def remove_doc(self, notebook_id: str, path: str) -> None: + self._post("/api/filetree/removeDoc", {"notebook": notebook_id, "path": path}) + + def remove_doc_by_id(self, doc_id: str) -> None: + self._post("/api/filetree/removeDocByID", {"id": doc_id}) + + def move_docs(self, from_paths: list[str], to_notebook: str, to_path: str) -> None: + self._post("/api/filetree/moveDocs", { + "fromPaths": from_paths, "toNotebook": to_notebook, "toPath": to_path, + }) + + def move_docs_by_id(self, from_ids: list[str], to_id: str) -> None: + self._post("/api/filetree/moveDocsByID", {"fromIDs": from_ids, "toID": to_id}) + + def get_hpath_by_id(self, block_id: str) -> str: + return self._post("/api/filetree/getHPathByID", {"id": block_id}) + + def get_path_by_id(self, block_id: str) -> dict[str, str]: + return self._post("/api/filetree/getPathByID", {"id": block_id}) + + def get_ids_by_hpath(self, notebook_id: str, path: str) -> list[str]: + return self._post("/api/filetree/getIDsByHPath", {"notebook": notebook_id, "path": path}) + + def list_docs_by_path(self, notebook_id: str, path: str) -> list[dict[str, Any]]: + return self._post("/api/filetree/listDocsByPath", { + "notebook": notebook_id, "path": path, "maxListCount": 0, + }) + + def list_doc_tree(self, notebook_id: str, path: str = "/", max_depth: int = -1, sort: int = 0) -> list[dict[str, Any]]: + return self._post("/api/filetree/listDocTree", { + "notebook": notebook_id, "path": path, "maxDepth": max_depth, "sort": sort, + }) + + def search_docs(self, keyword: str) -> list[dict[str, Any]]: + return self._post("/api/filetree/searchDocs", {"k": keyword}) + + def create_daily_note(self, notebook_id: str) -> dict[str, Any]: + return self._post("/api/filetree/createDailyNote", {"notebook": notebook_id}) + + # ── Block API ────────────────────────────────────────────────────── + + def insert_block(self, data_type: str, data: str, + parent_id: str = "", previous_id: str = "", next_id: str = "") -> list[dict[str, Any]]: + params = {"dataType": data_type, "data": data} + if parent_id: + params["parentID"] = parent_id + if previous_id: + params["previousID"] = previous_id + if next_id: + params["nextID"] = next_id + return self._post("/api/block/insertBlock", params) + + def prepend_block(self, data_type: str, data: str, parent_id: str) -> list[dict[str, Any]]: + return self._post("/api/block/prependBlock", { + "dataType": data_type, "data": data, "parentID": parent_id, + }) + + def append_block(self, data_type: str, data: str, parent_id: str) -> list[dict[str, Any]]: + return self._post("/api/block/appendBlock", { + "dataType": data_type, "data": data, "parentID": parent_id, + }) + + def update_block(self, data_type: str, data: str, block_id: str) -> list[dict[str, Any]]: + return self._post("/api/block/updateBlock", { + "dataType": data_type, "data": data, "id": block_id, + }) + + def delete_block(self, block_id: str) -> list[dict[str, Any]]: + return self._post("/api/block/deleteBlock", {"id": block_id}) + + def move_block(self, block_id: str, previous_id: str = "", parent_id: str = "") -> list[dict[str, Any]]: + params = {"id": block_id} + if previous_id: + params["previousID"] = previous_id + if parent_id: + params["parentID"] = parent_id + return self._post("/api/block/moveBlock", params) + + def fold_block(self, block_id: str) -> None: + self._post("/api/block/foldBlock", {"id": block_id}) + + def unfold_block(self, block_id: str) -> None: + self._post("/api/block/unfoldBlock", {"id": block_id}) + + def get_block_kramdown(self, block_id: str) -> str: + data = self._post("/api/block/getBlockKramdown", {"id": block_id}) + return data.get("kramdown", "") + + def get_child_blocks(self, block_id: str) -> list[dict[str, Any]]: + return self._post("/api/block/getChildBlocks", {"id": block_id}) + + # ── Attribute API ────────────────────────────────────────────────── + + def set_block_attrs(self, block_id: str, attrs: dict[str, str]) -> None: + self._post("/api/attr/setBlockAttrs", {"id": block_id, "attrs": attrs}) + + def get_block_attrs(self, block_id: str) -> dict[str, str]: + return self._post("/api/attr/getBlockAttrs", {"id": block_id}) + + # ── SQL Query API ────────────────────────────────────────────────── + + def query_sql(self, stmt: str) -> list[dict[str, Any]]: + return self._post("/api/query/sql", {"stmt": stmt}) + + # ── Search API ───────────────────────────────────────────────────── + + def search_blocks(self, query: str) -> list[dict[str, Any]]: + return self._post("/api/search/fullTextSearchBlock", {"query": query}) + + def search_tag(self, tag: str = "") -> list[str]: + """Search tags. Returns list of tag name strings.""" + data = self._post("/api/search/searchTag", {"k": tag}) + if isinstance(data, dict): + return data.get("tags", []) + return data or [] + + def find_replace(self, keyword: str, replacement: str, ids: list[str]) -> None: + """Search and replace text in blocks by ID.""" + self._post("/api/search/findReplace", { + "k": keyword, "r": replacement, "ids": ids, + }) + + # ── Export API ───────────────────────────────────────────────────── + + def export_md_content(self, doc_id: str) -> dict[str, str]: + return self._post("/api/export/exportMdContent", {"id": doc_id}) + + def export_resources(self, paths: list[str], name: str = "") -> str: + params = {"paths": paths} + if name: + params["name"] = name + data = self._post("/api/export/exportResources", params) + return data.get("path", "") + + # ── Tag API ──────────────────────────────────────────────────────── + + def get_tags(self) -> list[dict[str, Any]]: + return self._post("/api/tag/getTag", {"ignoreMaxListHint": True}) + + # ── System API ───────────────────────────────────────────────────── + + def get_version(self) -> str: + return self._post("/api/system/version") + + def get_current_time(self) -> int: + return self._post("/api/system/currentTime") diff --git a/siyuan/agent-harness/cli_anything/siyuan/core/session.py b/siyuan/agent-harness/cli_anything/siyuan/core/session.py new file mode 100644 index 000000000..252bbf4a3 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/core/session.py @@ -0,0 +1,75 @@ +"""Session state management for the SiYuan CLI REPL.""" + +import json +import os +from dataclasses import dataclass, field, asdict +from pathlib import Path +from typing import Any + + +@dataclass +class SessionState: + """Mutable session state — represents current notebook/doc context.""" + current_notebook_id: str = "" + current_notebook_name: str = "" + current_doc_id: str = "" + current_doc_path: str = "" + connected: bool = False + history: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "SessionState": + return cls( + current_notebook_id=data.get("current_notebook_id", ""), + current_notebook_name=data.get("current_notebook_name", ""), + current_doc_id=data.get("current_doc_id", ""), + current_doc_path=data.get("current_doc_path", ""), + connected=data.get("connected", False), + history=data.get("history", []), + ) + + +class SessionManager: + """Manages session persistence for the REPL.""" + + def __init__(self, state_dir: str | None = None): + if state_dir is None: + state_dir = str(Path.home() / ".cli-anything-siyuan") + self.state_dir = Path(state_dir) + self.state_dir.mkdir(parents=True, exist_ok=True) + self.session_path = self.state_dir / "session.json" + self.state = SessionState() + self._dirty = False + + def load(self) -> SessionState: + """Load session state from disk.""" + if self.session_path.is_file(): + try: + data = json.loads(self.session_path.read_text(encoding="utf-8")) + self.state = SessionState.from_dict(data) + except (json.JSONDecodeError, KeyError): + self.state = SessionState() + return self.state + + def save(self) -> None: + """Save current session state to disk.""" + self.session_path.write_text( + json.dumps(self.state.to_dict(), ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + def update(self, **kwargs: Any) -> None: + """Update session state fields and mark dirty.""" + for key, value in kwargs.items(): + if hasattr(self.state, key): + setattr(self.state, key, value) + self._dirty = True + + def flush(self) -> None: + """Write to disk if dirty.""" + if self._dirty: + self.save() + self._dirty = False diff --git a/siyuan/agent-harness/cli_anything/siyuan/siyuan_cli.py b/siyuan/agent-harness/cli_anything/siyuan/siyuan_cli.py new file mode 100644 index 000000000..de4d1f345 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/siyuan_cli.py @@ -0,0 +1,802 @@ +"""cli-anything-siyuan — SiYuan (思源笔记) CLI harness. + +Connects to a running SiYuan kernel via its HTTP API and provides +commands for notebooks, documents, blocks, search, and export. +""" + +import json +import os +import shlex +import sys +from typing import Any + +import click + +from cli_anything.siyuan.core.client import ( + SiYuanClient, + SiYuanClientError, + SiYuanConfig, + load_config, +) +from cli_anything.siyuan.core.session import SessionManager + + +def _read_stdin() -> str: + """Read stdin as raw bytes and decode as UTF-8 (with BOM handling). + + PowerShell pipes text using the console code page (e.g. GBK on Chinese + Windows), which mangles CJK characters before Python sees them. Reading + raw bytes then decoding as UTF-8-sig is the most robust approach. + """ + if sys.stdin.isatty(): + raise click.UsageError( + "stdin pipe expected (e.g. echo 'content' | sy block insert --parent pid -)" + ) + raw = sys.stdin.buffer.read() + try: + return raw.decode('utf-8-sig') + except UnicodeDecodeError: + return raw.decode('utf-8-sig', errors='replace') + + +# ── Click context ───────────────────────────────────────────────────── + +class SiYuanContext: + def __init__(self, client: SiYuanClient | None = None, + session: SessionManager | None = None, + json_output: bool = False): + self.json_output = json_output + self.client = client + self.session = session or SessionManager() + + +class _CatchErrors(click.Group): + """Click Group that converts SiYuanClientError to clean CLI errors.""" + + def invoke(self, ctx: click.Context) -> object: + try: + return super().invoke(ctx) + except SiYuanClientError as e: + click.echo(f"Error: {e}", err=True) + sys.exit(1) + + +@click.group(cls=_CatchErrors, invoke_without_command=True) +@click.option("--json", "json_output", is_flag=True, help="Output in JSON format") +@click.option("--host", default="", help="SiYuan host (default: 127.0.0.1)") +@click.option("--port", default=0, type=int, help="SiYuan port (default: 6806)") +@click.option("--token", default="", help="SiYuan API token") +@click.option("--config", "config_path", default="", help="Config file path") +@click.pass_context +def cli(ctx: click.Context, json_output: bool, host: str, port: int, + token: str, config_path: str): + # Force UTF-8 output to handle CJK characters on Windows + for stream in (sys.stdout, sys.stderr): + if hasattr(stream, 'reconfigure'): + stream.reconfigure(encoding='utf-8') + + """CLI for SiYuan (思源笔记) — interact with your knowledge base. + + Connects to a running SiYuan instance via its HTTP API. + Default: http://127.0.0.1:6806 + + Configure connection via ~/.siyuan-cli.json, env vars + (SIYUAN_HOST, SIYUAN_PORT, SIYUAN_TOKEN), or CLI flags. + """ + session = SessionManager() + session.load() + cfg = load_config(config_path or None) + if host or port or token: + cfg = SiYuanConfig( + host=host or cfg.host, + port=port or cfg.port, + token=token or cfg.token, + ) + client = SiYuanClient(cfg) + ctx.obj = SiYuanContext(client=client, session=session, json_output=json_output) + + if ctx.invoked_subcommand is None: + if not client.ping(): + click.echo("Error: Cannot connect to SiYuan. Is it running?", err=True) + click.echo(" Configure via: --host --port --token", err=True) + click.echo(" Or set: SIYUAN_HOST, SIYUAN_PORT, SIYUAN_TOKEN", err=True) + sys.exit(1) + ctx.invoke(repl) + + +def _walk_tree(items: list[dict], depth: int = 0) -> list[dict]: + """Recursively flatten a doc tree, setting depth on each entry. + + Returns new dicts; does not mutate the originals. + """ + result: list[dict] = [] + for t in items: + entry = {**t, "depth": depth} + result.append(entry) + children = t.get("children") + if children: + result.extend(_walk_tree(children, depth + 1)) + return result + + +# ── REPL ─────────────────────────────────────────────────────────────── + +def _build_repl_commands() -> dict[str, str]: + return { + "notebook list": "List all notebooks", + "notebook create <name>": "Create a notebook", + "notebook rename <id> <name>": "Rename a notebook", + "notebook remove <id>": "Remove a notebook", + "doc create <notebook> <path>": "Create a document with optional --md content", + "doc list <notebook> <path>": "List documents at path", + "doc tree <notebook>": "List full document tree", + "doc get <id>": "Get document info by ID", + "block insert <parent> <data>": "Insert a block (use '-' for stdin)", + "block update <id> <data>": "Update a block (use '-' for stdin)", + "block delete <id>": "Delete a block", + "block get <id>": "Get block kramdown source", + "sql <stmt>": "Execute SQL query", + "search <query>": "Full-text search", + "export md <doc-id>": "Export doc as Markdown", + "status": "Show connection and session status", + "help": "Show this help", + "quit": "Exit REPL", + } + + +@cli.command() +@click.pass_context +def repl(ctx: click.Context): + """Start interactive REPL mode.""" + from cli_anything.siyuan.utils.repl_skin import ReplSkin + + obj: SiYuanContext = ctx.obj + + try: + kernel_version = obj.client.get_version() + except SiYuanClientError: + kernel_version = "?" + skin = ReplSkin("siyuan", version=kernel_version) + skin.print_banner() + + pt_session = skin.create_prompt_session() + commands = _build_repl_commands() + + state = obj.session.state + + while True: + ctx_str = "" + if state.current_notebook_name: + ctx_str = state.current_notebook_name + user_input = skin.get_input( + pt_session, + project_name=state.current_doc_path or "", + modified=False, + context=ctx_str, + ) + + if not user_input: + continue + + if user_input in ("quit", "exit", "q"): + obj.session.flush() + break + + if user_input == "help": + skin.help(commands) + continue + + if user_input == "status": + connected = obj.client.ping() + skin.status_block({ + "Connected": str(connected), + "Notebook": state.current_notebook_name or "(none)", + "Document": state.current_doc_path or "(none)", + }, title="Status") + continue + + try: + _dispatch_repl(skin, obj, user_input) + except SiYuanClientError as e: + skin.error(str(e)) + except Exception as e: + skin.error(f"Error: {e}") + + skin.print_goodbye() + + +def _dispatch_repl(skin: Any, ctx: SiYuanContext, cmd: str) -> None: + """Parse REPL command and route to the appropriate handler.""" + parts = shlex.split(cmd.strip()) + if not parts: + return + + json_mode = "--json" in parts + parts = [p for p in parts if p != "--json"] + + client = ctx.client + session = ctx.session + + command = parts[0] + if command == "notebook": + _handle_notebook_repl(skin, client, session, parts, json_mode) + elif command == "doc": + _handle_doc_repl(skin, client, session, parts, json_mode) + elif command == "block": + _handle_block_repl(skin, client, parts, json_mode) + elif command == "sql" and len(parts) >= 2: + _handle_sql_repl(skin, client, parts, json_mode) + elif command == "search" and len(parts) >= 2: + _handle_search_repl(skin, client, parts, json_mode) + elif command == "export": + _handle_export_repl(skin, client, parts, json_mode) + else: + skin.error(f"Unknown command: {parts[0]}") + + +# ── REPL sub-handlers ────────────────────────────────────────────────── + +def _handle_notebook_repl(skin: Any, client: SiYuanClient, + session: SessionManager, parts: list[str], + json_mode: bool) -> None: + if len(parts) < 2: + skin.error("Usage: notebook <list|create|rename|remove>") + return + sub = parts[1] + if sub == "list": + notebooks = client.list_notebooks() + if json_mode: + click.echo(json.dumps(notebooks, ensure_ascii=False)) + else: + skin.table(["ID", "Name", "Icon", "Closed"], + [[n["id"], n["name"], n.get("icon", ""), str(n.get("closed", ""))] + for n in notebooks]) + elif sub == "create" and len(parts) >= 3: + name = " ".join(parts[2:]) + nb = client.create_notebook(name) + session.update(current_notebook_id=nb["id"], current_notebook_name=nb["name"]) + client.open_notebook(nb["id"]) + skin.success(f'Created notebook: {nb["name"]} ({nb["id"]})') + elif sub == "rename" and len(parts) >= 4: + client.rename_notebook(parts[2], " ".join(parts[3:])) + skin.success("Renamed") + elif sub == "remove" and len(parts) >= 3: + client.remove_notebook(parts[2]) + skin.success("Removed") + else: + skin.error("Invalid notebook command") + + +def _handle_doc_repl(skin: Any, client: SiYuanClient, + session: SessionManager, parts: list[str], + json_mode: bool) -> None: + if len(parts) < 2: + skin.error("Usage: doc <create|list|tree|get|rename|remove|export>") + return + sub = parts[1] + if sub == "create" and len(parts) >= 4: + md = "" + if "--md" in parts: + idx = parts.index("--md") + 1 + if idx < len(parts): + md = parts[idx] + parts = parts[:parts.index("--md")] + if md == "-": + md = _read_stdin() + nb_id = parts[2] + doc_path = parts[3] + doc_id = client.create_doc_with_md(nb_id, doc_path, md) + session.update(current_doc_id=doc_id, current_doc_path=doc_path) + if json_mode: + click.echo(json.dumps({"id": doc_id}, ensure_ascii=False)) + else: + skin.success(f"Created doc: {doc_id}") + elif sub == "list" and len(parts) >= 4: + docs = client.list_docs_by_path(parts[2], parts[3]) + items = docs.get("files", []) if isinstance(docs, dict) else docs + if json_mode: + click.echo(json.dumps(items, ensure_ascii=False)) + else: + skin.table(["ID", "Name", "Type"], + [[d.get("id", ""), d.get("name", ""), d.get("type", "")] + for d in items]) + elif sub == "tree" and len(parts) >= 3: + tree = client.list_doc_tree(parts[2]) + if isinstance(tree, dict): + items = tree.get("files") or tree.get("tree") or [] + else: + items = tree + if json_mode: + click.echo(json.dumps(items, ensure_ascii=False)) + else: + skin.table(["ID", "Name", "Path"], + [[t.get("id", ""), " " * t.get("depth", 0) + t.get("name", ""), t.get("path", "")] + for t in _walk_tree(items)]) + elif sub == "get" and len(parts) >= 3: + hpath = client.get_hpath_by_id(parts[2]) + if json_mode: + click.echo(json.dumps({"hpath": hpath}, ensure_ascii=False)) + else: + skin.success(f"Path: {hpath}") + elif sub == "rename" and len(parts) >= 4: + client.rename_doc_by_id(parts[2], " ".join(parts[3:])) + skin.success("Renamed") + elif sub == "remove" and len(parts) >= 3: + client.remove_doc_by_id(parts[2]) + skin.success("Removed") + elif sub == "export" and len(parts) >= 3: + md = client.export_md_content(parts[2]) + if json_mode: + click.echo(json.dumps(md, ensure_ascii=False)) + else: + skin.section(md.get("hPath", "")) + click.echo(md.get("content", "")) + else: + skin.error("Invalid doc command") + + +def _handle_block_repl(skin: Any, client: SiYuanClient, + parts: list[str], json_mode: bool) -> None: + if len(parts) < 2: + skin.error("Usage: block <insert|prepend|append|update|delete|get|child>") + return + sub = parts[1] + if sub == "insert": + if len(parts) < 4: + skin.error("Usage: block insert <parent_id> <data>") + return + data = parts[3] + if data == "-": + data = _read_stdin() + result = client.insert_block("markdown", data, parent_id=parts[2]) + if json_mode: + click.echo(json.dumps(result, ensure_ascii=False)) + else: + skin.success("Block inserted") + elif sub == "prepend" and len(parts) >= 4: + data = parts[3] + if data == "-": + data = _read_stdin() + client.prepend_block("markdown", data, parts[2]) + skin.success("Block prepended") + elif sub == "append" and len(parts) >= 4: + data = parts[3] + if data == "-": + data = _read_stdin() + client.append_block("markdown", data, parts[2]) + skin.success("Block appended") + elif sub == "update" and len(parts) >= 4: + data = parts[3] + if data == "-": + data = _read_stdin() + client.update_block("markdown", data, parts[2]) + skin.success("Block updated") + elif sub == "delete" and len(parts) >= 3: + client.delete_block(parts[2]) + skin.success("Block deleted") + elif sub == "get" and len(parts) >= 3: + kramdown = client.get_block_kramdown(parts[2]) + if json_mode: + click.echo(json.dumps({"kramdown": kramdown}, ensure_ascii=False)) + else: + click.echo(kramdown) + elif sub == "child" and len(parts) >= 3: + children = client.get_child_blocks(parts[2]) + if json_mode: + click.echo(json.dumps(children, ensure_ascii=False)) + else: + skin.table(["ID", "Type", "SubType"], + [[c.get("id", ""), c.get("type", ""), c.get("subType", "")] + for c in children]) + else: + skin.error("Invalid block command") + + +def _handle_sql_repl(skin: Any, client: SiYuanClient, + parts: list[str], json_mode: bool) -> None: + stmt = " ".join(parts[1:]) + results = client.query_sql(stmt) + if json_mode: + click.echo(json.dumps(results, ensure_ascii=False)) + elif not results: + skin.info("No results") + else: + headers = list(results[0].keys()) + rows = [[str(r.get(h, "")) for h in headers] for r in results] + skin.table(headers, rows) + + +def _handle_search_repl(skin: Any, client: SiYuanClient, + parts: list[str], json_mode: bool) -> None: + query = " ".join(parts[1:]) + data = client.search_blocks(query) + blocks = data.get("blocks", []) if isinstance(data, dict) else data + if json_mode: + click.echo(json.dumps(blocks, ensure_ascii=False)) + elif not blocks: + skin.info("No results") + else: + skin.table(["ID", "Content"], + [[r.get("id", ""), r.get("content", "")[:80]] for r in blocks]) + + +def _handle_export_repl(skin: Any, client: SiYuanClient, + parts: list[str], json_mode: bool) -> None: + if parts[1] == "md" and len(parts) >= 3: + md = client.export_md_content(parts[2]) + if json_mode: + click.echo(json.dumps(md, ensure_ascii=False)) + else: + skin.section(md.get("hPath", "")) + click.echo(md.get("content", "")) + else: + skin.error("Usage: export md <doc-id>") + + +# ── Notebook commands ────────────────────────────────────────────────── + +@cli.group() +def notebook(): + """Manage notebooks (笔记本).""" + + +@notebook.command("list") +@click.pass_obj +def notebook_list(ctx: SiYuanContext): + """List all notebooks.""" + notebooks = ctx.client.list_notebooks() + if ctx.json_output: + click.echo(json.dumps(notebooks, ensure_ascii=False)) + else: + click.echo(f"{'ID':<30} {'Name':<30} {'Closed':<8}") + click.echo("-" * 70) + for nb in notebooks: + click.echo(f"{nb['id']:<30} {nb['name']:<30} {str(nb.get('closed', '')):<8}") + + +@notebook.command("create") +@click.argument("name") +@click.pass_obj +def notebook_create(ctx: SiYuanContext, name: str): + """Create a new notebook.""" + nb = ctx.client.create_notebook(name) + if ctx.json_output: + click.echo(json.dumps(nb, ensure_ascii=False)) + else: + click.echo(f"Created notebook: {nb['name']} ({nb['id']})") + + +@notebook.command("remove") +@click.argument("notebook_id") +@click.pass_obj +def notebook_remove(ctx: SiYuanContext, notebook_id: str): + """Remove a notebook by ID.""" + ctx.client.remove_notebook(notebook_id) + click.echo(f"Removed notebook: {notebook_id}") + + +@notebook.command("rename") +@click.argument("notebook_id") +@click.argument("name") +@click.pass_obj +def notebook_rename(ctx: SiYuanContext, notebook_id: str, name: str): + """Rename a notebook.""" + ctx.client.rename_notebook(notebook_id, name) + click.echo(f"Renamed notebook {notebook_id} to: {name}") + + +@notebook.command("open") +@click.argument("notebook_id") +@click.pass_obj +def notebook_open(ctx: SiYuanContext, notebook_id: str): + """Open a notebook.""" + ctx.client.open_notebook(notebook_id) + name = notebook_id + for nb in ctx.client.list_notebooks(): + if nb.get("id") == notebook_id: + name = nb.get("name", notebook_id) + break + ctx.session.update(current_notebook_id=notebook_id, current_notebook_name=name) + ctx.session.flush() + click.echo(f"Opened notebook: {name} ({notebook_id})") + + +# ── Document commands ────────────────────────────────────────────────── + +@cli.group() +def doc(): + """Manage documents (文档).""" + + +@doc.command("create") +@click.argument("notebook_id") +@click.argument("path") +@click.option("--md", default="", help="Markdown content. Use '-' to read from stdin.") +@click.pass_obj +def doc_create(ctx: SiYuanContext, notebook_id: str, path: str, md: str): + """Create a document with optional Markdown content. + + To avoid shell escaping issues with special characters + (backticks, quotes, parentheses), pipe Markdown via stdin: + echo "## Title" | sy doc create nb1 /test --md - + """ + if md == "-": + md = _read_stdin() + doc_id = ctx.client.create_doc_with_md(notebook_id, path, md) + if ctx.json_output: + click.echo(json.dumps({"id": doc_id}, ensure_ascii=False)) + else: + click.echo(f"Created doc: {doc_id}") + + +@doc.command("list") +@click.argument("notebook_id") +@click.argument("path", default="/") +@click.pass_obj +def doc_list(ctx: SiYuanContext, notebook_id: str, path: str): + """List documents at a path.""" + docs = ctx.client.list_docs_by_path(notebook_id, path) + items = docs.get("files", []) if isinstance(docs, dict) else docs + if ctx.json_output: + click.echo(json.dumps(items, ensure_ascii=False)) + else: + click.echo(f"{'ID':<30} {'Name':<30} {'Type':<10}") + click.echo("-" * 70) + for d in items: + click.echo(f"{d.get('id', ''):<30} {d.get('name', ''):<30} {d.get('type', ''):<10}") + + +@doc.command("tree") +@click.argument("notebook_id") +@click.option("--path", default="/", help="Root path") +@click.option("--depth", default=-1, type=int, help="Max depth") +@click.pass_obj +def doc_tree(ctx: SiYuanContext, notebook_id: str, path: str, depth: int): + """List document tree.""" + tree = ctx.client.list_doc_tree(notebook_id, path=path, max_depth=depth) + if isinstance(tree, dict): + items = tree.get("files") or tree.get("tree") or [] + else: + items = tree + if ctx.json_output: + click.echo(json.dumps(items, ensure_ascii=False)) + else: + for t in _walk_tree(items): + indent = " " * t.get("depth", 0) + click.echo(f"{indent}{t.get('name', '')} ({t.get('id', '')})") + + +@doc.command("get") +@click.argument("doc_id") +@click.pass_obj +def doc_get(ctx: SiYuanContext, doc_id: str): + """Get document info by ID.""" + hpath = ctx.client.get_hpath_by_id(doc_id) + if ctx.json_output: + click.echo(json.dumps({"hpath": hpath}, ensure_ascii=False)) + else: + click.echo(f"Path: {hpath}") + + +@doc.command("rename") +@click.argument("doc_id") +@click.argument("title") +@click.pass_obj +def doc_rename(ctx: SiYuanContext, doc_id: str, title: str): + """Rename a document.""" + ctx.client.rename_doc_by_id(doc_id, title) + click.echo(f"Renamed {doc_id} to: {title}") + + +@doc.command("remove") +@click.argument("doc_id") +@click.pass_obj +def doc_remove(ctx: SiYuanContext, doc_id: str): + """Remove a document.""" + ctx.client.remove_doc_by_id(doc_id) + click.echo(f"Removed: {doc_id}") + + +# ── Block commands ───────────────────────────────────────────────────── + +@cli.group() +def block(): + """Manage blocks (内容块).""" + + +@block.command("insert") +@click.argument("data", required=False) +@click.option("--previous", default="", help="Previous block ID") +@click.option("--parent", default="", help="Parent block ID") +@click.option("--next", "next_", default="", help="Next block ID") +@click.option("--data-type", default="markdown", help="Data type (markdown/dom)") +@click.pass_obj +def block_insert(ctx: SiYuanContext, data: str | None, previous: str, parent: str, next_: str, data_type: str): + """Insert a block. Data reads from stdin when '-' or omitted.""" + if not parent and not previous and not next_: + raise click.UsageError("An anchor is required: --parent, --previous, or --next") + if not data or data == "-": + data = _read_stdin() + result = ctx.client.insert_block(data_type, data, parent_id=parent, previous_id=previous, next_id=next_) + if ctx.json_output: + click.echo(json.dumps(result, ensure_ascii=False)) + else: + click.echo("Block inserted") + + +@block.command("update") +@click.argument("block_id") +@click.argument("data", required=False) +@click.option("--data-type", default="markdown", help="Data type") +@click.pass_obj +def block_update(ctx: SiYuanContext, block_id: str, data: str | None, data_type: str): + """Update a block's content. Data reads from stdin when '-' or omitted.""" + if not data or data == "-": + data = _read_stdin() + ctx.client.update_block(data_type, data, block_id) + click.echo(f"Updated block: {block_id}") + + +@block.command("delete") +@click.argument("block_id") +@click.pass_obj +def block_delete(ctx: SiYuanContext, block_id: str): + """Delete a block.""" + ctx.client.delete_block(block_id) + click.echo(f"Deleted block: {block_id}") + + +@block.command("get") +@click.argument("block_id") +@click.pass_obj +def block_get(ctx: SiYuanContext, block_id: str): + """Get block kramdown source.""" + kramdown = ctx.client.get_block_kramdown(block_id) + if ctx.json_output: + click.echo(json.dumps({"kramdown": kramdown}, ensure_ascii=False)) + else: + click.echo(kramdown) + + +@block.command("children") +@click.argument("block_id") +@click.pass_obj +def block_children(ctx: SiYuanContext, block_id: str): + """Get child blocks.""" + children = ctx.client.get_child_blocks(block_id) + if ctx.json_output: + click.echo(json.dumps(children, ensure_ascii=False)) + else: + for c in children: + click.echo(f"{c.get('id', ''):<30} {c.get('type', ''):<8} {c.get('subType', '')}") + + +# ── SQL command ──────────────────────────────────────────────────────── + +@cli.command() +@click.argument("stmt") +@click.pass_obj +def sql(ctx: SiYuanContext, stmt: str): + """Execute a SQL query on the block database. + + Example: sql "SELECT * FROM blocks WHERE content LIKE '%keyword%' LIMIT 10" + """ + results = ctx.client.query_sql(stmt) + if ctx.json_output: + click.echo(json.dumps(results, ensure_ascii=False)) + elif not results: + click.echo("No results") + else: + for r in results: + click.echo(json.dumps(r, ensure_ascii=False)) + + +# ── Search commands ──────────────────────────────────────────────────── + +@cli.command() +@click.argument("query") +@click.pass_obj +def search(ctx: SiYuanContext, query: str): + """Full-text search across all blocks.""" + data = ctx.client.search_blocks(query) + blocks = data.get("blocks", []) if isinstance(data, dict) else data + if ctx.json_output: + click.echo(json.dumps(blocks, ensure_ascii=False)) + elif not blocks: + click.echo("No results") + else: + for r in blocks[:20]: + click.echo(f"- {r.get('id', '')}: {r.get('content', '')[:120]}") + + +# ── Export commands ──────────────────────────────────────────────────── + +@cli.group() +def export(): + """Export content from SiYuan.""" + + +@export.command("md") +@click.argument("doc_id") +@click.pass_obj +def export_md(ctx: SiYuanContext, doc_id: str): + """Export a document as Markdown.""" + md = ctx.client.export_md_content(doc_id) + if ctx.json_output: + click.echo(json.dumps(md, ensure_ascii=False)) + else: + click.echo(f"# {md.get('hPath', '')}") + click.echo("") + click.echo(md.get("content", "")) + + +# ── System commands ──────────────────────────────────────────────────── + +@cli.command() +@click.pass_obj +def version(ctx: SiYuanContext): + """Show SiYuan kernel version.""" + ver = ctx.client.get_version() + if ctx.json_output: + click.echo(json.dumps({"version": ver}, ensure_ascii=False)) + else: + click.echo(f"SiYuan version: {ver}") + + +@cli.command() +@click.pass_obj +def status(ctx: SiYuanContext): + """Show connection and session status.""" + connected = ctx.client.ping() + siyuan_ver = ctx.client.get_version() if connected else "" + session = ctx.session + state = session.state + + info = { + "connected": connected, + "siyuan_version": siyuan_ver, + "host": ctx.client.config.host, + "port": ctx.client.config.port, + "current_notebook": state.current_notebook_name or "", + "current_doc": state.current_doc_path or "", + } + + if ctx.json_output: + click.echo(json.dumps(info, ensure_ascii=False)) + else: + click.echo(f" Connected: {info['connected']}") + click.echo(f" SiYuan: v{info['siyuan_version']}") + click.echo(f" Host: {info['host']}:{info['port']}") + click.echo(f" Notebook: {info['current_notebook']}") + click.echo(f" Document: {info['current_doc']}") + + +# ── Tag commands ─────────────────────────────────────────────────────── + +@cli.group() +def tag(): + """Manage tags (标签).""" + + +def _print_tags(tags: list[dict], indent: int = 0) -> None: + """Recursively print tags with hierarchical indentation.""" + pad = " " * indent + for t in tags: + if indent == 0: + click.echo(f"{t.get('name', ''):<30} ({t.get('count', 0)})") + else: + click.echo(f"{pad}{t.get('name', ''):<{30 - 2 * indent}} ({t.get('count', 0)})") + children = t.get("children") + if children: + _print_tags(children, indent + 1) + + +@tag.command("list") +@click.pass_obj +def tag_list(ctx: SiYuanContext): + """List all tags (including nested tags).""" + tags = ctx.client.get_tags() + if ctx.json_output: + click.echo(json.dumps(tags, ensure_ascii=False)) + else: + _print_tags(tags) diff --git a/siyuan/agent-harness/cli_anything/siyuan/skills/SKILL.md b/siyuan/agent-harness/cli_anything/siyuan/skills/SKILL.md new file mode 100644 index 000000000..fc7c14539 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/skills/SKILL.md @@ -0,0 +1,106 @@ +--- +name: cli-anything-siyuan +description: SiYuan (思源笔记) CLI — manage notebooks, documents, blocks, and search your knowledge base from the terminal. +--- + +# cli-anything-siyuan + +CLI harness for [SiYuan](https://github.com/siyuan-note/siyuan) (思源笔记), +a local-first knowledge management and note-taking application. + +This CLI connects to a running SiYuan kernel via its HTTP API +(`http://127.0.0.1:6806`) and provides structured access to notebooks, +documents, blocks, search, and export. + +## Prerequisites + +- SiYuan must be running (Settings → About shows the API token) +- Python 3.10+ +- Install: `pip install cli-anything-siyuan` or `pip install -e .` from agent-harness/ + +## Command Groups + +### notebook — Notebook management +| Subcommand | Description | +|------------|-------------| +| `list` | List all notebooks | +| `create <name>` | Create a new notebook | +| `rename <id> <name>` | Rename a notebook | +| `remove <id>` | Delete a notebook | +| `open <id>` | Open a notebook | + +### doc — Document management +| Subcommand | Description | +|------------|-------------| +| `create <notebook> <path> [--md "content"]` | Create a document | +| `list <notebook> [path]` | List documents | +| `tree <notebook> [--path / --depth]` | Show doc tree | +| `get <id>` | Get document path by ID | +| `rename <id> <title>` | Rename a document | +| `remove <id>` | Delete a document | + +### block — Block operations +| Subcommand | Description | +|------------|-------------| +| `insert <data> [--previous / --parent]` | Insert a block | +| `update <id> <data>` | Update a block | +| `delete <id>` | Delete a block | +| `get <id>` | Get block kramdown source | +| `children <id>` | Get child blocks | + +### Other commands +| Command | Description | +|---------|-------------| +| `sql <stmt>` | Execute SQL on the block database | +| `search <query>` | Full-text search across blocks | +| `export md <doc-id>` | Export document as Markdown | +| `tag list` | List all tags | +| `version` | Show SiYuan kernel version | +| `status` | Show connection and session status | +| `repl` | Start interactive REPL | + +## Agent Guidance + +- Always use `--json` for machine-readable output +- Power search: use `sql "SELECT * FROM blocks WHERE content LIKE '%keyword%'"` for SQL-level access +- Document IDs look like `20210817205410-2kvfpfn` (timestamp-based) +- API token can be found in SiYuan Settings → About +- Connection defaults: `http://127.0.0.1:6806` + +## Examples + +```bash +# List notebooks (JSON) +cli-anything-siyuan --json notebook list + +# Create a document with Markdown (--md flag, watch shell escaping) +cli-anything-siyuan doc create nb1 /projects/new --md "## Title\n\nContent" + +# Create a document — pipe Markdown via stdin (avoids shell escaping) +# PowerShell here-string (literal, no escaping needed): +@' +## Title +Content with `backticks` and (parentheses) and "quotes" +'@ | cli-anything-siyuan doc create nb1 /projects/new --md - + +# Bash heredoc: +# cat <<'EOF' | cli-anything-siyuan doc create nb1 /projects/new --md - +# ## Title +# Content with `backticks` and (parentheses) +# EOF + +# SQL search +cli-anything-siyuan sql "SELECT id, content FROM blocks WHERE content LIKE '%meeting%' LIMIT 5" + +# Export +cli-anything-siyuan export md doc123 + +# Enter REPL +cli-anything-siyuan +``` + +## Error Handling + +- Connection errors: check that SiYuan is running and the API token is correct +- API errors: returned as `{"code": N, "msg": "..."}` — check the message field +- Auth errors: verify the token in `~/.siyuan-cli.json` or `SIYUAN_TOKEN` env var diff --git a/siyuan/agent-harness/cli_anything/siyuan/tests/__init__.py b/siyuan/agent-harness/cli_anything/siyuan/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/siyuan/agent-harness/cli_anything/siyuan/tests/test_cli_commands.py b/siyuan/agent-harness/cli_anything/siyuan/tests/test_cli_commands.py new file mode 100644 index 000000000..a97568510 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/tests/test_cli_commands.py @@ -0,0 +1,408 @@ +"""Unit tests for cli-anything-siyuan CLI commands. + +Tests CLI output formatting using Click's CliRunner with mocked client. +No external dependencies or running SiYuan instance required. +""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from click.testing import CliRunner + +from cli_anything.siyuan.siyuan_cli import cli + + +@pytest.fixture +def runner(): + return CliRunner() + + +@pytest.fixture +def mock_ctx(): + """Create a mock SiYuanContext with a mock client.""" + ctx = MagicMock() + ctx.json_output = False + return ctx + + +# ── Search command ───────────────────────────────────────────────────── + + +class TestSearchCommand: + def test_search_returns_blocks_list(self, runner, mock_ctx): + """search handles API returning list of blocks directly.""" + mock_ctx.client.search_blocks.return_value = [ + {"id": "b1", "content": "Dit模型训练结果如何"}, + {"id": "b2", "content": "训练完成，准确率90%"}, + ] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["search", "Dit模型"]) + assert result.exit_code == 0 + assert "b1" in result.output + assert "Dit模型训练结果如何" in result.output + assert "b2" in result.output + + def test_search_returns_dict_with_blocks_key(self, runner, mock_ctx): + """search handles API returning dict with 'blocks' key (real SiYuan format). + + This is the fix for KeyError: slice — real SiYuan API returns + {"blocks": [...], "rootBlocks": {...}} not a flat list. + """ + mock_ctx.client.search_blocks.return_value = { + "blocks": [ + {"id": "b1", "content": "Dit模型训练完成"}, + {"id": "b2", "content": "loss=0.02"}, + ], + "rootBlocks": {}, + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["search", "Dit模型"]) + assert result.exit_code == 0 + assert "b1" in result.output + assert "Dit模型训练完成" in result.output + + def test_search_json_output(self, runner, mock_ctx): + """--json search returns raw data.""" + mock_ctx.json_output = True + mock_ctx.client.search_blocks.return_value = { + "blocks": [{"id": "b1", "content": "test"}], + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["--json", "search", "test"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data[0]["id"] == "b1" + + def test_search_no_results(self, runner, mock_ctx): + """search shows 'No results' when empty list.""" + mock_ctx.client.search_blocks.return_value = [] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["search", "nonexistent"]) + assert result.exit_code == 0 + assert "No results" in result.output + + def test_search_no_results_dict(self, runner, mock_ctx): + """search shows 'No results' when dict with empty blocks list.""" + mock_ctx.client.search_blocks.return_value = {"blocks": [], "rootBlocks": {}} + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["search", "nonexistent"]) + assert result.exit_code == 0 + assert "No results" in result.output + + +# ── Doc list command ─────────────────────────────────────────────────── + + +class TestDocListCommand: + def test_doc_list_returns_files(self, runner, mock_ctx): + """doc list handles API returning dict with 'files' key.""" + mock_ctx.client.list_docs_by_path.return_value = { + "box": "nb1", + "files": [ + {"id": "doc1", "name": "测试文档", "type": "d"}, + {"id": "doc2", "name": "笔记", "type": "d"}, + ], + "path": "/", + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["doc", "list", "nb1"]) + assert result.exit_code == 0 + assert "doc1" in result.output + assert "测试文档" in result.output + assert "doc2" in result.output + + def test_doc_list_empty(self, runner, mock_ctx): + """doc list handles empty files list.""" + mock_ctx.client.list_docs_by_path.return_value = { + "box": "nb1", "files": [], "path": "/", + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["doc", "list", "nb1"]) + assert result.exit_code == 0 + # Should show header but no data rows + assert "ID" in result.output + assert "Name" in result.output + + def test_doc_list_json(self, runner, mock_ctx): + """--json doc list returns raw files array.""" + mock_ctx.json_output = True + mock_ctx.client.list_docs_by_path.return_value = { + "box": "nb1", "files": [{"id": "doc1", "name": "Test"}], "path": "/", + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["--json", "doc", "list", "nb1"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data[0]["id"] == "doc1" + + +# ── Doc tree command ─────────────────────────────────────────────────── + + +class TestDocTreeCommand: + def test_doc_tree_returns_files(self, runner, mock_ctx): + """doc tree handles API returning dict with 'files' key.""" + mock_ctx.client.list_doc_tree.return_value = { + "files": [ + {"id": "doc1", "name": "根文档", "depth": 0}, + {"id": "doc2", "name": "子文档", "depth": 1}, + ], + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["doc", "tree", "nb1"]) + assert result.exit_code == 0 + assert "根文档" in result.output + assert "子文档" in result.output + + def test_doc_tree_json(self, runner, mock_ctx): + """--json doc tree returns raw files array.""" + mock_ctx.json_output = True + mock_ctx.client.list_doc_tree.return_value = { + "files": [{"id": "doc1", "name": "Root"}], + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["--json", "doc", "tree", "nb1"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data[0]["id"] == "doc1" + + +# ── Notebook list command ────────────────────────────────────────────── + + +class TestNotebookListCommand: + def test_notebook_list(self, runner, mock_ctx): + """notebook list returns formatted table.""" + mock_ctx.client.list_notebooks.return_value = [ + {"id": "nb1", "name": "工作笔记", "closed": False}, + {"id": "nb2", "name": "归档", "closed": True}, + ] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["notebook", "list"]) + assert result.exit_code == 0 + assert "工作笔记" in result.output + assert "nb1" in result.output + + +# ── Tag list command ─────────────────────────────────────────────────── + + +class TestTagListCommand: + def test_tag_list(self, runner, mock_ctx): + """tag list shows tag names with counts.""" + mock_ctx.client.get_tags.return_value = [ + {"name": "AI", "count": 5}, + {"name": "Python", "count": 12}, + {"name": "笔记", "count": 3}, + ] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["tag", "list"]) + assert result.exit_code == 0 + assert "AI" in result.output + assert "5" in result.output + assert "Python" in result.output + assert "12" in result.output + + +# ── SQL command ──────────────────────────────────────────────────────── + + +class TestSQLCommand: + def test_sql_query(self, runner, mock_ctx): + """sql returns query results.""" + mock_ctx.client.query_sql.return_value = [ + {"id": "b1", "content": "hello"}, + {"id": "b2", "content": "world"}, + ] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["sql", "SELECT * FROM blocks"]) + assert result.exit_code == 0 + assert "b1" in result.output + assert "hello" in result.output + + +# ── Status command ───────────────────────────────────────────────────── + + +class TestStatusCommand: + def test_status(self, runner, mock_ctx): + """status shows connection info.""" + mock_ctx.client.status.return_value = { + "connected": True, + "version": "3.6.5", + } + mock_ctx.client.get_version.return_value = "3.6.5" + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + mock_session = MagicMock() + mock_session.state.connected = True + mock_session.state.current_notebook_id = "nb1" + mock_session.state.current_notebook_name = "工作" + mock_session.state.current_doc_id = "doc1" + mock_ctx.client.get_version.return_value = "3.6.5" + mock_ctx.session = mock_session + mock_ctx.current_notebook_id = "nb1" + mock_ctx.current_notebook_name = "工作" + mock_ctx.current_doc_id = "doc1" + + result = runner.invoke(cli, ["status"]) + assert result.exit_code == 0 + assert "Connected" in result.output or "connected" in result.output.lower() + + +# ── Block insert command ──────────────────────────────────────────────── + + +class TestBlockInsertCommand: + def test_block_insert_with_parent(self, runner, mock_ctx): + """block insert with --parent succeeds.""" + mock_ctx.client.insert_block.return_value = [{"id": "new-block"}] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["block", "insert", "hello", "--parent", "pid"]) + assert result.exit_code == 0 + assert "Block inserted" in result.output + mock_ctx.client.insert_block.assert_called_once_with( + "markdown", "hello", parent_id="pid", previous_id="", next_id="" + ) + + def test_block_insert_with_previous(self, runner, mock_ctx): + """block insert with --previous succeeds.""" + mock_ctx.client.insert_block.return_value = [{"id": "new-block"}] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["block", "insert", "hello", "--previous", "prev"]) + assert result.exit_code == 0 + assert "Block inserted" in result.output + mock_ctx.client.insert_block.assert_called_once_with( + "markdown", "hello", parent_id="", previous_id="prev", next_id="" + ) + + def test_block_insert_with_next(self, runner, mock_ctx): + """block insert with --next succeeds.""" + mock_ctx.client.insert_block.return_value = [{"id": "new-block"}] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["block", "insert", "hello", "--next", "nid"]) + assert result.exit_code == 0 + assert "Block inserted" in result.output + mock_ctx.client.insert_block.assert_called_once_with( + "markdown", "hello", parent_id="", previous_id="", next_id="nid" + ) + + def test_block_insert_without_anchor_errors(self, runner, mock_ctx): + """block insert without any anchor raises UsageError.""" + mock_ctx.client.insert_block.return_value = [{"id": "new-block"}] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["block", "insert", "hello"]) + assert result.exit_code == 2 + assert "anchor" in result.output.lower() or "Error" in result.output + + def test_block_insert_json_output(self, runner, mock_ctx): + """--json block insert returns raw data.""" + mock_ctx.json_output = True + mock_ctx.client.insert_block.return_value = [{"id": "new-block"}] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["--json", "block", "insert", "hello", "--parent", "pid"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data[0]["id"] == "new-block" + + +# ── Doc get command ──────────────────────────────────────────────────── + + +class TestDocGetCommand: + def test_doc_get(self, runner, mock_ctx): + """doc get shows document path.""" + mock_ctx.client.get_hpath_by_id.return_value = "/我的文档/测试" + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["doc", "get", "doc123"]) + assert result.exit_code == 0 + assert "/我的文档/测试" in result.output + + +# ── Doc create command ────────────────────────────────────────────────── + + +class TestDocCreateCommand: + def test_doc_create_without_md(self, runner, mock_ctx): + """doc create without --md does not read stdin (passes empty string).""" + mock_ctx.client.create_doc_with_md.return_value = "doc123" + with ( + patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx), + ): + result = runner.invoke(cli, ["doc", "create", "nb1", "/test"]) + assert result.exit_code == 0 + assert "doc123" in result.output + + def test_doc_create_with_md(self, runner, mock_ctx): + """doc create with --md passes the markdown content.""" + mock_ctx.client.create_doc_with_md.return_value = "doc123" + with ( + patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx), + ): + result = runner.invoke(cli, ["doc", "create", "nb1", "/test", "--md", "# Hello"]) + assert result.exit_code == 0 + mock_ctx.client.create_doc_with_md.assert_called_with("nb1", "/test", "# Hello") + assert "doc123" in result.output + + def test_doc_create_json_output(self, runner, mock_ctx): + """--json doc create returns doc ID.""" + mock_ctx.json_output = True + mock_ctx.client.create_doc_with_md.return_value = "doc123" + with ( + patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx), + ): + result = runner.invoke(cli, ["--json", "doc", "create", "nb1", "/test"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data["id"] == "doc123" + + +# ── Doc tree recursive rendering ────────────────────────────────────── + + +class TestDocTreeRecursiveCommand: + def test_doc_tree_handle_nested_children(self, runner, mock_ctx): + """doc tree recursively renders nested children.""" + mock_ctx.client.list_doc_tree.return_value = { + "files": [ + { + "id": "doc1", "name": "Root", "depth": 0, + "children": [ + {"id": "doc2", "name": "Child", "depth": 1, "children": []}, + ], + }, + {"id": "doc3", "name": "Sibling", "depth": 0, "children": []}, + ], + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["doc", "tree", "nb1"]) + assert result.exit_code == 0 + assert "Root" in result.output + assert "Child" in result.output + assert "Sibling" in result.output + assert "doc1" in result.output + assert "doc2" in result.output + + def test_doc_tree_flat_items(self, runner, mock_ctx): + """doc tree also works with flat items (no children key).""" + mock_ctx.client.list_doc_tree.return_value = [ + {"id": "doc1", "name": "Flat1", "depth": 0}, + {"id": "doc2", "name": "Flat2", "depth": 1}, + ] + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["doc", "tree", "nb1"]) + assert result.exit_code == 0 + assert "Flat1" in result.output + assert "Flat2" in result.output + + def test_doc_tree_nested_json_output(self, runner, mock_ctx): + """--json doc tree with nested children returns raw data.""" + mock_ctx.json_output = True + mock_ctx.client.list_doc_tree.return_value = { + "files": [{"id": "doc1", "name": "Root", "children": []}], + } + with patch("cli_anything.siyuan.siyuan_cli.SiYuanContext", return_value=mock_ctx): + result = runner.invoke(cli, ["--json", "doc", "tree", "nb1"]) + assert result.exit_code == 0 + data = json.loads(result.output) + assert data[0]["id"] == "doc1" diff --git a/siyuan/agent-harness/cli_anything/siyuan/tests/test_core.py b/siyuan/agent-harness/cli_anything/siyuan/tests/test_core.py new file mode 100644 index 000000000..4f81c107d --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/tests/test_core.py @@ -0,0 +1,204 @@ +"""Unit tests for cli-anything-siyuan core modules. + +Tests use synthetic data and mocking — no external dependencies. +""" + +import json +import os +import tempfile +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from cli_anything.siyuan.core.client import SiYuanClient, SiYuanConfig, load_config +from cli_anything.siyuan.core.session import SessionManager, SessionState + + +# ── Client config tests ──────────────────────────────────────────────── + +class TestClientConfig: + def test_default_config(self): + """Default config has correct default values.""" + config = SiYuanConfig() + assert config.host == "127.0.0.1" + assert config.port == 6806 + assert config.token == "" + assert config.base_url == "http://127.0.0.1:6806" + + def test_custom_config(self): + """Custom config values are used.""" + config = SiYuanConfig(host="localhost", port=6807, token="abc123") + assert config.host == "localhost" + assert config.port == 6807 + assert config.token == "abc123" + assert config.base_url == "http://localhost:6807" + + @patch.dict(os.environ, {"SIYUAN_HOST": "192.168.1.100", "SIYUAN_PORT": "6808", "SIYUAN_TOKEN": "envtoken"}, clear=True) + @patch("cli_anything.siyuan.core.client.Path.home") + def test_load_config_from_env(self, mock_home): + """load_config reads from environment variables.""" + mock_home.return_value = Path(tempfile.mkdtemp()) + config = load_config() + assert config.host == "192.168.1.100" + assert config.port == 6808 + assert config.token == "envtoken" + + def test_load_config_from_file(self): + """load_config reads from JSON config file.""" + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False, encoding="utf-8") as f: + json.dump({"host": "10.0.0.1", "port": 6809, "token": "filetoken"}, f) + fname = f.name + try: + config = load_config(fname) + assert config.host == "10.0.0.1" + assert config.port == 6809 + assert config.token == "filetoken" + finally: + os.unlink(fname) + + +# ── Client API tests ─────────────────────────────────────────────────── + +class TestSiYuanClient: + @pytest.fixture + def client(self): + return SiYuanClient(SiYuanConfig(token="test-token")) + + def test_ping_success(self, client): + """ping returns True when server responds.""" + mock_session = MagicMock() + mock_session.post.return_value.status_code = 200 + mock_session.post.return_value.json.return_value = {"code": 0, "data": "3.6.5"} + client._session = mock_session + assert client.ping() is True + + def test_ping_failure(self, client): + """ping returns False when server is unreachable.""" + mock_session = MagicMock() + mock_session.post.side_effect = __import__("requests").ConnectionError("refused") + client._session = mock_session + assert client.ping() is False + + def test_list_notebooks(self, client): + """list_notebooks parses API response correctly.""" + mock_data = { + "notebooks": [ + {"id": "nb1", "name": "Test Notebook", "icon": "1f4d4", "closed": False}, + {"id": "nb2", "name": "Daily Notes", "icon": "1f4cb", "closed": True}, + ] + } + mock_session = MagicMock() + mock_session.post.return_value.status_code = 200 + mock_session.post.return_value.json.return_value = {"code": 0, "data": mock_data} + client._session = mock_session + + notebooks = client.list_notebooks() + assert len(notebooks) == 2 + assert notebooks[0]["name"] == "Test Notebook" + assert notebooks[1]["id"] == "nb2" + + def test_create_doc_with_md(self, client): + """create_doc_with_md returns the doc ID.""" + mock_session = MagicMock() + mock_session.post.return_value.status_code = 200 + mock_session.post.return_value.json.return_value = {"code": 0, "data": "doc123"} + client._session = mock_session + + doc_id = client.create_doc_with_md("nb1", "/test/path", "# Hello") + assert doc_id == "doc123" + # Verify correct API call (requests uses json= kwarg, not data=) + call_kwargs = mock_session.post.call_args[1] + body = call_kwargs["json"] + assert body["notebook"] == "nb1" + assert body["path"] == "/test/path" + assert body["markdown"] == "# Hello" + + def test_query_sql(self, client): + """query_sql returns list of rows.""" + mock_session = MagicMock() + mock_session.post.return_value.status_code = 200 + mock_session.post.return_value.json.return_value = { + "code": 0, "data": [{"id": "b1", "content": "hello"}, {"id": "b2", "content": "world"}], + } + client._session = mock_session + + results = client.query_sql("SELECT * FROM blocks") + assert len(results) == 2 + assert results[0]["content"] == "hello" + + +# ── Session tests ────────────────────────────────────────────────────── + +class TestSessionManager: + @pytest.fixture + def mgr(self): + tmp_dir = tempfile.mkdtemp() + return SessionManager(state_dir=tmp_dir) + + def test_default_state(self, mgr): + """Fresh session has correct defaults.""" + assert mgr.state.connected is False + assert mgr.state.current_notebook_id == "" + assert mgr.state.current_notebook_name == "" + assert mgr.state.current_doc_id == "" + + def test_save_load_roundtrip(self, mgr): + """Save then load preserves all state.""" + mgr.update(current_notebook_id="nb1", current_notebook_name="Test", connected=True) + mgr.save() + + mgr2 = SessionManager(state_dir=str(mgr.state_dir)) + mgr2.load() + assert mgr2.state.current_notebook_id == "nb1" + assert mgr2.state.current_notebook_name == "Test" + assert mgr2.state.connected is True + + def test_partial_update(self, mgr): + """Update only changes specified fields.""" + mgr.update(current_notebook_id="nb1") + assert mgr.state.current_notebook_id == "nb1" + assert mgr.state.connected is False # unchanged + + +class TestFindReplace: + """Tests for find_replace using SiYuan API k/r/ids payload.""" + + @pytest.fixture + def client(self): + return SiYuanClient(SiYuanConfig(token="test-token")) + + def test_find_replace_uses_k_r_ids_payload(self, client): + """find_replace sends k/r/ids payload keys.""" + mock_session = MagicMock() + mock_session.post.return_value.status_code = 200 + mock_session.post.return_value.json.return_value = {"code": 0, "data": {"count": 3}} + client._session = mock_session + client.find_replace("old", "new", ["id1", "id2"]) + call_kwargs = mock_session.post.call_args[1] + body = call_kwargs["json"] + assert "k" in body + assert "r" in body + assert "ids" in body + assert body["k"] == "old" + assert body["r"] == "new" + assert body["ids"] == ["id1", "id2"] + +class TestUncappedListings: + """Tests for list_docs_by_path with maxListCount.""" + + @pytest.fixture + def client(self): + return SiYuanClient(SiYuanConfig(token="test-token")) + + def test_list_docs_by_path_sends_max_list_count(self, client): + """list_docs_by_path sends maxListCount: 0 for uncapped listings.""" + mock_session = MagicMock() + mock_session.post.return_value.status_code = 200 + mock_session.post.return_value.json.return_value = {"code": 0, "data": {"files": [{"id": "doc1", "name": "test"}]}} + client._session = mock_session + client.list_docs_by_path("nb1", "/") + call_kwargs = mock_session.post.call_args[1] + body = call_kwargs["json"] + assert "maxListCount" in body + assert body["maxListCount"] == 0 diff --git a/siyuan/agent-harness/cli_anything/siyuan/tests/test_full_e2e.py b/siyuan/agent-harness/cli_anything/siyuan/tests/test_full_e2e.py new file mode 100644 index 000000000..2efee700a --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/tests/test_full_e2e.py @@ -0,0 +1,122 @@ +"""E2E tests for cli-anything-siyuan. + +Requires a running SiYuan instance at http://127.0.0.1:6806. +Set SIYUAN_TOKEN env var if authentication is enabled. +""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from cli_anything.siyuan.core.client import SiYuanClient, SiYuanClientError, load_config + + +def _resolve_cli(name: str) -> list[str]: + """Resolve installed CLI command; falls back to python -m for dev.""" + import shutil + force = os.environ.get("CLI_ANYTHING_FORCE_INSTALLED", "").strip() == "1" + path = shutil.which(name) + if path: + print(f"[_resolve_cli] Using installed command: {path}") + return [path] + if force: + raise RuntimeError(f"{name} not found in PATH. Install with: pip install -e .") + module = name.replace("cli-anything-", "cli_anything.") # e.g. cli_anything.siyuan (has __main__.py) + print(f"[_resolve_cli] Falling back to: {sys.executable} -m {module}") + return [sys.executable, "-m", module] + + +CLI_BASE = _resolve_cli("cli-anything-siyuan") + + +# ── Fixtures ─────────────────────────────────────────────────────────── + +@pytest.fixture(scope="session") +def client() -> SiYuanClient: + """Create a client connected to the real SiYuan instance.""" + config = load_config() + c = SiYuanClient(config) + if not c.ping(): + pytest.skip("SiYuan is not running — E2E tests require a running instance") + return c + + +@pytest.fixture +def tmp_dir() -> Path: + """Create a temporary directory for test artifacts.""" + import tempfile + path = Path(tempfile.mkdtemp(prefix="siyuan_e2e_")) + yield path + import shutil + shutil.rmtree(path, ignore_errors=True) + + +class TestCLISubprocess: + CLI_BASE = CLI_BASE + + def _run(self, args: list[str], check: bool = True) -> subprocess.CompletedProcess: + return subprocess.run( + self.CLI_BASE + args, + capture_output=True, text=True, check=check, + ) + + def test_help(self): + """--help returns 0 and shows usage.""" + result = self._run(["--help"]) + assert result.returncode == 0 + assert "Usage:" in result.stdout + print(f"\n --help: OK ({len(result.stdout)} chars)") + + def test_version_json(self, client: SiYuanClient): + """--json version returns valid JSON with version string.""" + result = self._run(["--json", "version"]) + assert result.returncode == 0 + data = json.loads(result.stdout) + assert "version" in data + assert data["version"] + print(f"\n SiYuan version: {data['version']}") + + def test_status_json(self, client: SiYuanClient): + """--json status returns valid JSON with connection info.""" + result = self._run(["--json", "status"]) + assert result.returncode == 0 + data = json.loads(result.stdout) + assert "siyuan_version" in data + assert data["siyuan_version"] + print(f"\n Status: connected to SiYuan v{data['siyuan_version']}") + + +class TestBackendE2E: + """Tests that hit the real SiYuan API via Python client.""" + + def test_get_version(self, client: SiYuanClient): + """Get the SiYuan kernel version from a running instance.""" + ver = client.get_version() + assert ver + assert isinstance(ver, str) + assert len(ver) > 0 + print(f"\n SiYuan kernel version: {ver}") + + def test_list_notebooks(self, client: SiYuanClient): + """List all notebooks from the running instance.""" + notebooks = client.list_notebooks() + assert isinstance(notebooks, list) + if notebooks: + nb = notebooks[0] + assert "id" in nb + assert "name" in nb + print(f"\n Notebooks found: {len(notebooks)}") + print(f" First: {nb['name']} ({nb['id']})") + else: + print("\n No notebooks found (empty instance)") + + def test_get_current_time(self, client: SiYuanClient): + """Get current time from server.""" + t = client.get_current_time() + assert t > 0 + assert isinstance(t, int) + print(f"\n Server time (ts): {t}") diff --git a/siyuan/agent-harness/cli_anything/siyuan/utils/__init__.py b/siyuan/agent-harness/cli_anything/siyuan/utils/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/siyuan/agent-harness/cli_anything/siyuan/utils/repl_skin.py b/siyuan/agent-harness/cli_anything/siyuan/utils/repl_skin.py new file mode 100644 index 000000000..bc1fb6d1d --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/utils/repl_skin.py @@ -0,0 +1,567 @@ +"""cli-anything REPL Skin — Unified terminal interface for all CLI harnesses. + +Copy this file into your CLI package at: + cli_anything/<software>/utils/repl_skin.py + +Usage: + from cli_anything.<software>.utils.repl_skin import ReplSkin + + skin = ReplSkin("shotcut", version="1.0.0") + skin.print_banner() # auto-detects repo-root or packaged SKILL.md + prompt_text = skin.prompt(project_name="my_video.mlt", modified=True) + skin.success("Project saved") + skin.error("File not found") + skin.warning("Unsaved changes") + skin.info("Processing 24 clips...") + skin.status("Track 1", "3 clips, 00:02:30") + skin.table(headers, rows) + skin.print_goodbye() +""" + +import os +import sys +from pathlib import Path + +# ── ANSI color codes (no external deps for core styling) ────────────── + +_RESET = "\033[0m" +_BOLD = "\033[1m" +_DIM = "\033[2m" +_ITALIC = "\033[3m" +_UNDERLINE = "\033[4m" + +# Brand colors +_CYAN = "\033[38;5;80m" # cli-anything brand cyan +_CYAN_BG = "\033[48;5;80m" +_WHITE = "\033[97m" +_GRAY = "\033[38;5;245m" +_DARK_GRAY = "\033[38;5;240m" +_LIGHT_GRAY = "\033[38;5;250m" + +# Software accent colors — each software gets a unique accent +_ACCENT_COLORS = { + "gimp": "\033[38;5;214m", # warm orange + "blender": "\033[38;5;208m", # deep orange + "inkscape": "\033[38;5;39m", # bright blue + "audacity": "\033[38;5;33m", # navy blue + "libreoffice": "\033[38;5;40m", # green + "obs_studio": "\033[38;5;55m", # purple + "kdenlive": "\033[38;5;69m", # slate blue + "shotcut": "\033[38;5;35m", # teal green +} +_DEFAULT_ACCENT = "\033[38;5;75m" # default sky blue + +# Status colors +_GREEN = "\033[38;5;78m" +_YELLOW = "\033[38;5;220m" +_RED = "\033[38;5;196m" +_BLUE = "\033[38;5;75m" +_MAGENTA = "\033[38;5;176m" + +_SKILL_SOURCE_REPO = os.environ.get("CLI_ANYTHING_SKILL_REPO", "HKUDS/CLI-Anything") + +# ── Brand icon ──────────────────────────────────────────────────────── + +# The cli-anything icon: a small colored diamond/chevron mark +_ICON = f"{_CYAN}{_BOLD}◆{_RESET}" +_ICON_SMALL = f"{_CYAN}▸{_RESET}" + +# ── Box drawing characters ──────────────────────────────────────────── + +_H_LINE = "─" +_V_LINE = "│" +_TL = "╭" +_TR = "╮" +_BL = "╰" +_BR = "╯" +_T_DOWN = "┬" +_T_UP = "┴" +_T_RIGHT = "├" +_T_LEFT = "┤" +_CROSS = "┼" + + +def _strip_ansi(text: str) -> str: + """Remove ANSI escape codes for length calculation.""" + import re + return re.sub(r"\033\[[^m]*m", "", text) + + +def _visible_len(text: str) -> int: + """Get visible length of text (excluding ANSI codes).""" + return len(_strip_ansi(text)) + + +def _display_home_path(path: str) -> str: + """Display a path relative to the home directory when possible.""" + expanded = Path(path).expanduser().resolve() + home = Path.home().resolve() + try: + relative = expanded.relative_to(home) + return f"~/{relative.as_posix()}" + except ValueError: + return str(expanded) + + +class ReplSkin: + """Unified REPL skin for cli-anything CLIs. + + Provides consistent branding, prompts, and message formatting + across all CLI harnesses built with the cli-anything methodology. + """ + + def __init__(self, software: str, version: str = "1.0.0", + history_file: str | None = None, skill_path: str | None = None): + """Initialize the REPL skin. + + Args: + software: Software name (e.g., "gimp", "shotcut", "blender"). + version: CLI version string. + history_file: Path for persistent command history. + Defaults to ~/.cli-anything-<software>/history + skill_path: Path to the SKILL.md file for agent discovery. + Auto-detected from the repo-root skills/ tree when present, + otherwise from the package's skills/ directory. + Displayed in banner for AI agents to know where to read skill info. + """ + self.software = software.lower().replace("-", "_") + self.display_name = software.replace("_", " ").title() + self.version = version + software_aliases = {"iterm2_ctl": "iterm2"} + self.skill_slug = software_aliases.get(self.software, self.software).replace("_", "-") + self.skill_id = f"cli-anything-{self.skill_slug}" + self.skill_install_cmd = ( + f"npx skills add {_SKILL_SOURCE_REPO} --skill {self.skill_id} -g -y" + ) + global_skill_root = Path( + os.environ.get("CLI_ANYTHING_GLOBAL_SKILLS_DIR", str(Path.home() / ".agents" / "skills")) + ).expanduser() + self.global_skill_path = str(global_skill_root / self.skill_id / "SKILL.md") + + # Prefer repo-root canonical skills/<skill-id>/SKILL.md when running + # inside the CLI-Anything monorepo. Fall back to the packaged + # cli_anything/<software>/skills/SKILL.md for installed harnesses. + if skill_path is None: + package_skill = Path(__file__).resolve().parent.parent / "skills" / "SKILL.md" + repo_skill = None + for parent in Path(__file__).resolve().parents: + candidate = parent / "skills" / self.skill_id / "SKILL.md" + if candidate.is_file(): + repo_skill = candidate + break + if repo_skill and repo_skill.is_file(): + skill_path = str(repo_skill) + elif package_skill.is_file(): + skill_path = str(package_skill) + self.skill_path = skill_path + self.accent = _ACCENT_COLORS.get(self.software, _DEFAULT_ACCENT) + + # History file + if history_file is None: + hist_dir = Path.home() / f".cli-anything-{self.software}" + hist_dir.mkdir(parents=True, exist_ok=True) + self.history_file = str(hist_dir / "history") + else: + self.history_file = history_file + + # Detect terminal capabilities + self._color = self._detect_color_support() + + def _detect_color_support(self) -> bool: + """Check if terminal supports color.""" + if os.environ.get("NO_COLOR"): + return False + if os.environ.get("CLI_ANYTHING_NO_COLOR"): + return False + if not hasattr(sys.stdout, "isatty"): + return False + return sys.stdout.isatty() + + def _c(self, code: str, text: str) -> str: + """Apply color code if colors are supported.""" + if not self._color: + return text + return f"{code}{text}{_RESET}" + + # ── Banner ──────────────────────────────────────────────────────── + + def print_banner(self): + """Print the startup banner with branding.""" + import textwrap + + inner = 72 + + def _box_line(content: str) -> str: + """Wrap content in box drawing, padding to inner width.""" + pad = inner - _visible_len(content) + vl = self._c(_DARK_GRAY, _V_LINE) + return f"{vl}{content}{' ' * max(0, pad)}{vl}" + + def _meta_lines(label: str, value: str) -> list[str]: + """Wrap a metadata line for the banner box.""" + icon = self._c(_MAGENTA, "◇") + label_text = self._c(_DARK_GRAY, label) + prefix = f" {icon} {label_text} " + available = max(12, inner - _visible_len(prefix)) + wrapped = textwrap.wrap( + value, + width=available, + break_long_words=True, + break_on_hyphens=False, + ) or [""] + lines = [f"{prefix}{self._c(_LIGHT_GRAY, wrapped[0])}"] + continuation_prefix = " " * _visible_len(prefix) + for chunk in wrapped[1:]: + lines.append(f"{continuation_prefix}{self._c(_LIGHT_GRAY, chunk)}") + return lines + + top = self._c(_DARK_GRAY, f"{_TL}{_H_LINE * inner}{_TR}") + bot = self._c(_DARK_GRAY, f"{_BL}{_H_LINE * inner}{_BR}") + + # Title: ◆ cli-anything · Shotcut + icon = self._c(_CYAN + _BOLD, "◆") + brand = self._c(_CYAN + _BOLD, "cli-anything") + dot = self._c(_DARK_GRAY, "·") + name = self._c(self.accent + _BOLD, self.display_name) + title = f" {icon} {brand} {dot} {name}" + + ver = f" {self._c(_DARK_GRAY, f' v{self.version}')}" + tip = f" {self._c(_DARK_GRAY, ' Type help for commands, quit to exit')}" + empty = "" + + meta_lines: list[str] = [] + meta_lines.extend(_meta_lines("Install:", self.skill_install_cmd)) + meta_lines.extend(_meta_lines("Global skill:", _display_home_path(self.global_skill_path))) + print(top) + print(_box_line(title)) + print(_box_line(ver)) + for line in meta_lines: + print(_box_line(line)) + print(_box_line(empty)) + print(_box_line(tip)) + print(bot) + print() + + # ── Prompt ──────────────────────────────────────────────────────── + + def prompt(self, project_name: str = "", modified: bool = False, + context: str = "") -> str: + """Build a styled prompt string for prompt_toolkit or input(). + + Args: + project_name: Current project name (empty if none open). + modified: Whether the project has unsaved changes. + context: Optional extra context to show in prompt. + + Returns: + Formatted prompt string. + """ + parts = [] + + # Icon + if self._color: + parts.append(f"{_CYAN}◆{_RESET} ") + else: + parts.append("> ") + + # Software name + parts.append(self._c(self.accent + _BOLD, self.software)) + + # Project context + if project_name or context: + ctx = context or project_name + mod = "*" if modified else "" + parts.append(f" {self._c(_DARK_GRAY, '[')}") + parts.append(self._c(_LIGHT_GRAY, f"{ctx}{mod}")) + parts.append(self._c(_DARK_GRAY, ']')) + + parts.append(self._c(_GRAY, " ❯ ")) + + return "".join(parts) + + def prompt_tokens(self, project_name: str = "", modified: bool = False, + context: str = ""): + """Build prompt_toolkit formatted text tokens for the prompt. + + Use with prompt_toolkit's FormattedText for proper ANSI handling. + + Returns: + list of (style, text) tuples for prompt_toolkit. + """ + accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff") + tokens = [] + + tokens.append(("class:icon", "◆ ")) + tokens.append(("class:software", self.software)) + + if project_name or context: + ctx = context or project_name + mod = "*" if modified else "" + tokens.append(("class:bracket", " [")) + tokens.append(("class:context", f"{ctx}{mod}")) + tokens.append(("class:bracket", "]")) + + tokens.append(("class:arrow", " ❯ ")) + + return tokens + + def get_prompt_style(self): + """Get a prompt_toolkit Style object matching the skin. + + Returns: + prompt_toolkit.styles.Style + """ + try: + from prompt_toolkit.styles import Style + except ImportError: + return None + + accent_hex = _ANSI_256_TO_HEX.get(self.accent, "#5fafff") + + return Style.from_dict({ + "icon": "#5fdfdf bold", # cyan brand color + "software": f"{accent_hex} bold", + "bracket": "#585858", + "context": "#bcbcbc", + "arrow": "#808080", + # Completion menu + "completion-menu.completion": "bg:#303030 #bcbcbc", + "completion-menu.completion.current": f"bg:{accent_hex} #000000", + "completion-menu.meta.completion": "bg:#303030 #808080", + "completion-menu.meta.completion.current": f"bg:{accent_hex} #000000", + # Auto-suggest + "auto-suggest": "#585858", + # Bottom toolbar + "bottom-toolbar": "bg:#1c1c1c #808080", + "bottom-toolbar.text": "#808080", + }) + + # ── Messages ────────────────────────────────────────────────────── + + def success(self, message: str): + """Print a success message with green checkmark.""" + icon = self._c(_GREEN + _BOLD, "✓") + print(f" {icon} {self._c(_GREEN, message)}") + + def error(self, message: str): + """Print an error message with red cross.""" + icon = self._c(_RED + _BOLD, "✗") + print(f" {icon} {self._c(_RED, message)}", file=sys.stderr) + + def warning(self, message: str): + """Print a warning message with yellow triangle.""" + icon = self._c(_YELLOW + _BOLD, "⚠") + print(f" {icon} {self._c(_YELLOW, message)}") + + def info(self, message: str): + """Print an info message with blue dot.""" + icon = self._c(_BLUE, "●") + print(f" {icon} {self._c(_LIGHT_GRAY, message)}") + + def hint(self, message: str): + """Print a subtle hint message.""" + print(f" {self._c(_DARK_GRAY, message)}") + + def section(self, title: str): + """Print a section header.""" + print() + print(f" {self._c(self.accent + _BOLD, title)}") + print(f" {self._c(_DARK_GRAY, _H_LINE * len(title))}") + + # ── Status display ──────────────────────────────────────────────── + + def status(self, label: str, value: str): + """Print a key-value status line.""" + lbl = self._c(_GRAY, f" {label}:") + val = self._c(_WHITE, f" {value}") + print(f"{lbl}{val}") + + def status_block(self, items: dict[str, str], title: str = ""): + """Print a block of status key-value pairs. + + Args: + items: Dict of label -> value pairs. + title: Optional title for the block. + """ + if title: + self.section(title) + + max_key = max(len(k) for k in items) if items else 0 + for label, value in items.items(): + lbl = self._c(_GRAY, f" {label:<{max_key}}") + val = self._c(_WHITE, f" {value}") + print(f"{lbl}{val}") + + def progress(self, current: int, total: int, label: str = ""): + """Print a simple progress indicator. + + Args: + current: Current step number. + total: Total number of steps. + label: Optional label for the progress. + """ + pct = int(current / total * 100) if total > 0 else 0 + bar_width = 20 + filled = int(bar_width * current / total) if total > 0 else 0 + bar = "█" * filled + "░" * (bar_width - filled) + text = f" {self._c(_CYAN, bar)} {self._c(_GRAY, f'{pct:3d}%')}" + if label: + text += f" {self._c(_LIGHT_GRAY, label)}" + print(text) + + # ── Table display ───────────────────────────────────────────────── + + def table(self, headers: list[str], rows: list[list[str]], + max_col_width: int = 40): + """Print a formatted table with box-drawing characters. + + Args: + headers: Column header strings. + rows: List of rows, each a list of cell strings. + max_col_width: Maximum column width before truncation. + """ + if not headers: + return + + # Calculate column widths + col_widths = [min(len(h), max_col_width) for h in headers] + for row in rows: + for i, cell in enumerate(row): + if i < len(col_widths): + col_widths[i] = min( + max(col_widths[i], len(str(cell))), max_col_width + ) + + def pad(text: str, width: int) -> str: + t = str(text)[:width] + return t + " " * (width - len(t)) + + # Header + header_cells = [ + self._c(_CYAN + _BOLD, pad(h, col_widths[i])) + for i, h in enumerate(headers) + ] + sep = self._c(_DARK_GRAY, f" {_V_LINE} ") + header_line = f" {sep.join(header_cells)}" + print(header_line) + + # Separator + sep_parts = [self._c(_DARK_GRAY, _H_LINE * w) for w in col_widths] + sep_line = self._c(_DARK_GRAY, f" {'───'.join([_H_LINE * w for w in col_widths])}") + print(sep_line) + + # Rows + for row in rows: + cells = [] + for i, cell in enumerate(row): + if i < len(col_widths): + cells.append(self._c(_LIGHT_GRAY, pad(str(cell), col_widths[i]))) + row_sep = self._c(_DARK_GRAY, f" {_V_LINE} ") + print(f" {row_sep.join(cells)}") + + # ── Help display ────────────────────────────────────────────────── + + def help(self, commands: dict[str, str]): + """Print a formatted help listing. + + Args: + commands: Dict of command -> description pairs. + """ + self.section("Commands") + max_cmd = max(len(c) for c in commands) if commands else 0 + for cmd, desc in commands.items(): + cmd_styled = self._c(self.accent, f" {cmd:<{max_cmd}}") + desc_styled = self._c(_GRAY, f" {desc}") + print(f"{cmd_styled}{desc_styled}") + print() + + # ── Goodbye ─────────────────────────────────────────────────────── + + def print_goodbye(self): + """Print a styled goodbye message.""" + print(f"\n {_ICON_SMALL} {self._c(_GRAY, 'Goodbye!')}\n") + + # ── Prompt toolkit session factory ──────────────────────────────── + + def create_prompt_session(self): + """Create a prompt_toolkit PromptSession with skin styling. + + Returns: + A configured PromptSession, or None if prompt_toolkit unavailable. + """ + try: + from prompt_toolkit import PromptSession + from prompt_toolkit.history import FileHistory + from prompt_toolkit.auto_suggest import AutoSuggestFromHistory + from prompt_toolkit.formatted_text import FormattedText + + style = self.get_prompt_style() + + session = PromptSession( + history=FileHistory(self.history_file), + auto_suggest=AutoSuggestFromHistory(), + style=style, + enable_history_search=True, + ) + return session + except ImportError: + return None + + def get_input(self, pt_session, project_name: str = "", + modified: bool = False, context: str = "") -> str: + """Get input from user using prompt_toolkit or fallback. + + Args: + pt_session: A prompt_toolkit PromptSession (or None). + project_name: Current project name. + modified: Whether project has unsaved changes. + context: Optional context string. + + Returns: + User input string (stripped). + """ + if pt_session is not None: + from prompt_toolkit.formatted_text import FormattedText + tokens = self.prompt_tokens(project_name, modified, context) + return pt_session.prompt(FormattedText(tokens)).strip() + else: + raw_prompt = self.prompt(project_name, modified, context) + return input(raw_prompt).strip() + + # ── Toolbar builder ─────────────────────────────────────────────── + + def bottom_toolbar(self, items: dict[str, str]): + """Create a bottom toolbar callback for prompt_toolkit. + + Args: + items: Dict of label -> value pairs to show in toolbar. + + Returns: + A callable that returns FormattedText for the toolbar. + """ + def toolbar(): + from prompt_toolkit.formatted_text import FormattedText + parts = [] + for i, (k, v) in enumerate(items.items()): + if i > 0: + parts.append(("class:bottom-toolbar.text", " │ ")) + parts.append(("class:bottom-toolbar.text", f" {k}: ")) + parts.append(("class:bottom-toolbar", v)) + return FormattedText(parts) + return toolbar + + +# ── ANSI 256-color to hex mapping (for prompt_toolkit styles) ───────── + +_ANSI_256_TO_HEX = { + "\033[38;5;33m": "#0087ff", # audacity navy blue + "\033[38;5;35m": "#00af5f", # shotcut teal + "\033[38;5;39m": "#00afff", # inkscape bright blue + "\033[38;5;40m": "#00d700", # libreoffice green + "\033[38;5;55m": "#5f00af", # obs purple + "\033[38;5;69m": "#5f87ff", # kdenlive slate blue + "\033[38;5;75m": "#5fafff", # default sky blue + "\033[38;5;80m": "#5fd7d7", # brand cyan + "\033[38;5;208m": "#ff8700", # blender deep orange + "\033[38;5;214m": "#ffaf00", # gimp warm orange +} diff --git a/siyuan/agent-harness/cli_anything/siyuan/utils/siyuan_backend.py b/siyuan/agent-harness/cli_anything/siyuan/utils/siyuan_backend.py new file mode 100644 index 000000000..726b962a3 --- /dev/null +++ b/siyuan/agent-harness/cli_anything/siyuan/utils/siyuan_backend.py @@ -0,0 +1,82 @@ +"""SiYuan backend integration. + +Provides helper functions to find a running SiYuan instance and verify +connectivity. Since SiYuan has no headless CLI mode, this module handles +the connection check and configuration discovery. +""" + +import os +import subprocess +import sys +from pathlib import Path +from typing import Any + +from cli_anything.siyuan.core.client import SiYuanClient, SiYuanClientError, SiYuanConfig + + +def find_siyuan_data_dir() -> str | None: + """Try to locate the SiYuan data directory. + + Checks platform-specific default locations: + - Windows: %USERPROFILE%\\SiYuan\\data\\ + - Linux: ~/.config/siyuan/data/ + - macOS: ~/.config/siyuan/data/ + """ + home = Path.home() + candidates = [] + + if sys.platform == "win32": + candidates.append(home / "SiYuan" / "data") + else: + candidates.append(home / ".config" / "siyuan" / "data") + + for candidate in candidates: + if candidate.is_dir(): + return str(candidate.resolve()) + + return None + + +def check_siyuan_running(config: SiYuanConfig | None = None) -> dict[str, Any]: + """Check if a SiYuan instance is reachable and return its status. + + Returns: + dict with keys: connected (bool), version (str), error (str) + """ + if config is None: + from cli_anything.siyuan.core.client import load_config + config = load_config() + + client = SiYuanClient(config) + try: + version = client.get_version() + return {"connected": True, "version": version, "error": ""} + except SiYuanClientError as e: + return {"connected": False, "version": "", "error": str(e)} + + +def get_api_token_from_conf() -> str: + """Try to read the API token from SiYuan config files. + + Returns empty string if not found. + """ + home = Path.home() + candidates = [] + + if sys.platform == "win32": + candidates.append(home / "SiYuan" / "conf" / "conf.json") + else: + candidates.append(home / ".config" / "siyuan" / "conf" / "conf.json") + + import json + for conf_path in candidates: + if conf_path.is_file(): + try: + data = json.loads(conf_path.read_text(encoding="utf-8")) + token = data.get("api", {}).get("token", "") + if token: + return token + except (json.JSONDecodeError, KeyError, OSError): + continue + + return "" diff --git a/siyuan/agent-harness/setup.py b/siyuan/agent-harness/setup.py new file mode 100644 index 000000000..581e88acb --- /dev/null +++ b/siyuan/agent-harness/setup.py @@ -0,0 +1,41 @@ +from setuptools import setup, find_namespace_packages + +with open("cli_anything/siyuan/README.md", encoding="utf-8") as f: + long_description = f.read() + +setup( + name="cli-anything-siyuan", + version="1.0.0", + description="SiYuan (思源笔记) CLI — manage notebooks, documents, blocks from the terminal", + long_description=long_description, + long_description_content_type="text/markdown", + author="cli-anything", + packages=find_namespace_packages(include=["cli_anything.*"]), + package_data={ + "cli_anything.siyuan": ["skills/*.md"], + }, + include_package_data=True, + python_requires=">=3.10", + install_requires=[ + "click>=8.0", + "requests>=2.28", + ], + extras_require={ + "repl": ["prompt_toolkit>=3.0"], + "test": ["pytest>=7.0"], + }, + entry_points={ + "console_scripts": [ + "cli-anything-siyuan=cli_anything.siyuan.siyuan_cli:cli", + ], + }, + classifiers=[ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "License :: OSI Approved :: MIT License", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + ], +) diff --git a/skills/cli-anything-siyuan/SKILL.md b/skills/cli-anything-siyuan/SKILL.md new file mode 100644 index 000000000..9c594d521 --- /dev/null +++ b/skills/cli-anything-siyuan/SKILL.md @@ -0,0 +1,106 @@ +--- +name: "cli-anything-siyuan" +description: SiYuan (思源笔记) CLI — manage notebooks, documents, blocks, and search your knowledge base from the terminal. +--- + +# cli-anything-siyuan + +CLI harness for [SiYuan](https://github.com/siyuan-note/siyuan) (思源笔记), +a local-first knowledge management and note-taking application. + +This CLI connects to a running SiYuan kernel via its HTTP API +(`http://127.0.0.1:6806`) and provides structured access to notebooks, +documents, blocks, search, and export. + +## Prerequisites + +- SiYuan must be running (Settings → About shows the API token) +- Python 3.10+ +- Install: `pip install cli-anything-siyuan` or `pip install -e .` from agent-harness/ + +## Command Groups + +### notebook — Notebook management +| Subcommand | Description | +|------------|-------------| +| `list` | List all notebooks | +| `create <name>` | Create a new notebook | +| `rename <id> <name>` | Rename a notebook | +| `remove <id>` | Delete a notebook | +| `open <id>` | Open a notebook | + +### doc — Document management +| Subcommand | Description | +|------------|-------------| +| `create <notebook> <path> [--md "content"]` | Create a document | +| `list <notebook> [path]` | List documents | +| `tree <notebook> [--path / --depth]` | Show doc tree | +| `get <id>` | Get document path by ID | +| `rename <id> <title>` | Rename a document | +| `remove <id>` | Delete a document | + +### block — Block operations +| Subcommand | Description | +|------------|-------------| +| `insert <data> [--previous / --parent]` | Insert a block | +| `update <id> <data>` | Update a block | +| `delete <id>` | Delete a block | +| `get <id>` | Get block kramdown source | +| `children <id>` | Get child blocks | + +### Other commands +| Command | Description | +|---------|-------------| +| `sql <stmt>` | Execute SQL on the block database | +| `search <query>` | Full-text search across blocks | +| `export md <doc-id>` | Export document as Markdown | +| `tag list` | List all tags | +| `version` | Show SiYuan kernel version | +| `status` | Show connection and session status | +| `repl` | Start interactive REPL | + +## Agent Guidance + +- Always use `--json` for machine-readable output +- Power search: use `sql "SELECT * FROM blocks WHERE content LIKE '%keyword%'"` for SQL-level access +- Document IDs look like `20210817205410-2kvfpfn` (timestamp-based) +- API token can be found in SiYuan Settings → About +- Connection defaults: `http://127.0.0.1:6806` + +## Examples + +```bash +# List notebooks (JSON) +cli-anything-siyuan --json notebook list + +# Create a document with Markdown (--md flag, watch shell escaping) +cli-anything-siyuan doc create nb1 /projects/new --md "## Title\n\nContent" + +# Create a document — pipe Markdown via stdin (avoids shell escaping) +# PowerShell here-string (literal, no escaping needed): +@' +## Title +Content with `backticks` and (parentheses) and "quotes" +'@ | cli-anything-siyuan doc create nb1 /projects/new --md - + +# Bash heredoc: +# cat <<'EOF' | cli-anything-siyuan doc create nb1 /projects/new --md - +# ## Title +# Content with `backticks` and (parentheses) +# EOF + +# SQL search +cli-anything-siyuan sql "SELECT id, content FROM blocks WHERE content LIKE '%meeting%' LIMIT 5" + +# Export +cli-anything-siyuan export md doc123 + +# Enter REPL +cli-anything-siyuan +``` + +## Error Handling + +- Connection errors: check that SiYuan is running and the API token is correct +- API errors: returned as `{"code": N, "msg": "..."}` — check the message field +- Auth errors: verify the token in `~/.siyuan-cli.json` or `SIYUAN_TOKEN` env var