| | 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"] |
| |
|
