| from __future__ import annotations |
|
|
| import os |
| from typing import Annotated, Optional |
|
|
| import gradio as gr |
|
|
| from app import _log_call_end, _log_call_start, _truncate_for_log |
| from ._docstrings import autodoc |
| from ._core import obsidian_sandbox, OBSIDIAN_ROOT |
|
|
|
|
| TOOL_SUMMARY = ( |
| "Browse and search the Obsidian vault in read-only mode. " |
| "Actions: list, read, info, search, help. " |
| "All paths resolve within the vault root. Start paths with '/' (e.g., /Notes)." |
| ) |
|
|
| HELP_TEXT = ( |
| "Obsidian Vault — actions and usage\n\n" |
| "Root: Tools/Obsidian (override with OBSIDIAN_VAULT_ROOT). " |
| "Start paths with '/' to reference the vault root (e.g., /Projects/note.md). " |
| "Absolute paths are disabled unless UNSAFE_ALLOW_ABS_PATHS=1.\n\n" |
| "Actions and fields:\n" |
| "- list: path='/' (default), recursive=false, show_hidden=false, max_entries=20\n" |
| "- read: path (e.g., /Projects/note.md), offset=0, max_chars=4000 (shows next_cursor when truncated)\n" |
| "- info: path\n" |
| "- search: path (note or folder), query text in the Search field, recursive=false, show_hidden=false, max_entries=20, case_sensitive=false, offset=0\n" |
| "- help: show this guide\n\n" |
| "Errors are returned as JSON with fields: {status:'error', code, message, path?, hint?, data?}.\n\n" |
| "Examples:\n" |
| "- list current: action=list, path='/'\n" |
| "- read note: action=read, path='/Projects/note.md', max_chars=500\n" |
| "- show metadata: action=info, path='/Inbox'\n" |
| "- search notes: action=search, path='/Projects', query='deadline', recursive=true, max_entries=100\n" |
| "- case-sensitive search: action=search, query='TODO', case_sensitive=true\n" |
| "- page search results: action=search, query='TODO', offset=20\n" |
| ) |
|
|
|
|
| |
| _sandbox = obsidian_sandbox |
| ROOT_DIR = OBSIDIAN_ROOT |
|
|
|
|
| |
| def _resolve_path(path: str) -> tuple[str, str]: |
| return _sandbox.resolve_path(path) |
|
|
|
|
| def _display_path(abs_path: str) -> str: |
| return _sandbox.display_path(abs_path) |
|
|
|
|
| def _err(code: str, message: str, *, path: str | None = None, hint: str | None = None, data: dict | None = None) -> str: |
| return _sandbox.err(code, message, path=path, hint=hint, data=data) |
|
|
|
|
| @autodoc(summary=TOOL_SUMMARY) |
| def Obsidian_Vault( |
| action: Annotated[str, "Operation to perform: 'list', 'read', 'info', 'search', 'help'."], |
| path: Annotated[str, "Target path, relative to the vault root." ] = "/", |
| query: Annotated[Optional[str], "Text to search for when action=search."] = None, |
| recursive: Annotated[bool, "Recurse into subfolders when listing/searching."] = False, |
| show_hidden: Annotated[bool, "Include hidden files when listing/searching."] = False, |
| max_entries: Annotated[int, "Max entries to list or matches to return (for list/search)."] = 20, |
| offset: Annotated[int, "Start offset when reading files."] = 0, |
| max_chars: Annotated[int, "Max characters to return when reading (0 = full file)."] = 4000, |
| case_sensitive: Annotated[bool, "Match case when searching text."] = False, |
| ) -> str: |
| _log_call_start( |
| "Obsidian_Vault", |
| action=action, |
| path=path, |
| query=query, |
| recursive=recursive, |
| show_hidden=show_hidden, |
| max_entries=max_entries, |
| offset=offset, |
| max_chars=max_chars, |
| case_sensitive=case_sensitive, |
| ) |
| action = (action or "").strip().lower() |
| if action not in {"list", "read", "info", "search", "help"}: |
| result = _err( |
| "invalid_action", |
| "Invalid action.", |
| hint="Choose from: list, read, info, search, help.", |
| ) |
| _log_call_end("Obsidian_Vault", _truncate_for_log(result)) |
| return result |
|
|
| if action == "help": |
| result = HELP_TEXT |
| _log_call_end("Obsidian_Vault", _truncate_for_log(result)) |
| return result |
|
|
| abs_path, err = _resolve_path(path) |
| if err: |
| _log_call_end("Obsidian_Vault", _truncate_for_log(err)) |
| return err |
|
|
| try: |
| if action == "list": |
| if not os.path.exists(abs_path): |
| result = _err("path_not_found", f"Path not found: {_display_path(abs_path)}", path=_display_path(abs_path)) |
| else: |
| result = _sandbox.list_dir(abs_path, show_hidden=show_hidden, recursive=recursive, max_entries=max_entries) |
| elif action == "read": |
| result = _sandbox.read_file(abs_path, offset=offset, max_chars=max_chars) |
| elif action == "search": |
| query_text = query or "" |
| if query_text.strip() == "": |
| result = _err( |
| "missing_search_query", |
| "Search query is required for the search action.", |
| hint="Provide text in the Search field to look for.", |
| ) |
| else: |
| result = _sandbox.search_text( |
| abs_path, |
| query_text, |
| recursive=recursive, |
| show_hidden=show_hidden, |
| max_results=max_entries, |
| case_sensitive=case_sensitive, |
| start_index=offset, |
| ) |
| else: |
| result = _sandbox.info(abs_path) |
| except Exception as exc: |
| result = _err("exception", "Unhandled error during operation.", data={"error": _sandbox.safe_err(exc)}) |
|
|
| _log_call_end("Obsidian_Vault", _truncate_for_log(result)) |
| return result |
|
|
|
|
| def build_interface() -> gr.Interface: |
| return gr.Interface( |
| fn=Obsidian_Vault, |
| inputs=[ |
| gr.Radio( |
| label="Action", |
| choices=["list", "read", "info", "search", "help"], |
| value="help", |
| info="Operation to perform", |
| ), |
| gr.Textbox(label="Path", placeholder="/ or /Notes/todo.md", max_lines=1, value="/", info="Target path (relative to vault root)"), |
| gr.Textbox(label="Search text", lines=3, placeholder="Text to search for...", info="Text to search for (Search only)"), |
| gr.Checkbox(label="Recursive", value=False, info="Recurse into subfolders (List/Search)"), |
| gr.Checkbox(label="Show hidden", value=False, info="Include hidden files (List/Search)"), |
| gr.Slider(minimum=10, maximum=5000, step=10, value=20, label="Max entries / matches", info="Max entries to list or matches to return (List/Search)"), |
| gr.Slider(minimum=0, maximum=1_000_000, step=100, value=0, label="Offset", info="Start offset (Read/Search)"), |
| gr.Slider(minimum=0, maximum=100_000, step=500, value=4000, label="Max chars", info="Max characters to return (Read, 0=all)"), |
| gr.Checkbox(label="Case sensitive search", value=False, info="Match case (Search)"), |
| ], |
| outputs=gr.Textbox(label="Result", lines=20), |
| title="Obsidian Vault", |
| description=( |
| "<div style=\"text-align:center; overflow:hidden;\">Explore and search notes in the vault without modifying them." "</div>" |
| ), |
| api_description=TOOL_SUMMARY, |
| flagging_mode="never", |
| submit_btn="Run", |
| ) |
|
|
|
|
| __all__ = ["Obsidian_Vault", "build_interface"] |
|
|
