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