Source code for folderbot.tools.folder_tools

"""FolderTools class for interacting with the configured folder."""

from __future__ import annotations

import logging
import time
from typing import TYPE_CHECKING, Any

from ..bot import BotContext
from ..config import Config
from ..calendar_store import CalendarStore
from ..todo_store import TodoStore
from .activity_log import ActivityLogger
from .base import ToolResult
from .loader import load_custom_tools
from .registry import FolderServices, folder_bot
from .scheduler_tools import SchedulerServices
from .upload_tools import UploadServices

# Import tools to register them with folder_bot
from . import list_files as _list_files  # noqa: F401
from . import read_file as _read_file  # noqa: F401
from . import read_files as _read_files  # noqa: F401
from . import search_files as _search_files  # noqa: F401
from . import write_file as _write_file  # noqa: F401
from . import activity_log as _activity_log  # noqa: F401
from . import file_notifications as _file_notifications  # noqa: F401
from . import web_search as _web_search  # noqa: F401
from . import web_fetch as _web_fetch  # noqa: F401
from . import utils as _utils  # noqa: F401
from . import scheduler_tools as _scheduler_tools  # noqa: F401
from . import upload_tools as _upload_tools  # noqa: F401
from . import token_usage as _token_usage  # noqa: F401
from . import token_stats as _token_stats  # noqa: F401
from . import topic_tools as _topic_tools  # noqa: F401
from . import todo as _todo  # noqa: F401
from . import plot as _plot  # noqa: F401
from . import calendar as _calendar  # noqa: F401

if TYPE_CHECKING:
    from ..scheduler.scheduler import TaskScheduler
    from ..session_manager import SessionManager

logger = logging.getLogger(__name__)

# Tools that require user confirmation before use
TOOLS_REQUIRING_CONFIRMATION = ["write_file"]




class FolderTools:
    """Tools for interacting with the configured folder.

    This class serves as the main coordinator for tool execution. All tools
    (file tools, utility tools, web tools, scheduler tools) are registered
    on the single folder_bot instance and executed through this class.
    """



    def __init__(self, config: Config):
        self.config = config
        self._root = config.root_folder.resolve()
        self._scheduler: TaskScheduler | None = None
        self._session_manager: SessionManager | None = None
        self._upload_services: UploadServices | None = None
        self._custom_tools = load_custom_tools(self._root, tools_config=config.tools)
        self._activity_logger = ActivityLogger(self._root)
        self._todo_store = TodoStore(config.todo_path)
        self._calendar_store = CalendarStore(config.db_path)

        # Create folder services for the new pattern
        self._folder_services = FolderServices(
            root=self._root,
            config=config,
            activity_logger=self._activity_logger,
            session_manager=None,  # Set later
        )

        if self._custom_tools:
            logger.info("Custom tools loaded from .folderbot/tools")

        # Check if web tools are available
        from .web_search import is_available as web_search_available
        from .web_fetch import is_available as web_fetch_available

        if web_search_available() or web_fetch_available():
            logger.info("Web tools available")




    def set_scheduler(self, scheduler: TaskScheduler) -> None:
        """Set the scheduler reference for tool execution."""
        self._scheduler = scheduler




    def set_session_manager(self, session_manager: SessionManager) -> None:
        """Set the session manager reference for user preferences."""
        self._session_manager = session_manager
        # Update folder services with session manager
        self._folder_services = FolderServices(
            root=self._root,
            config=self.config,
            activity_logger=self._activity_logger,
            session_manager=session_manager,
        )




    def set_upload_services(self, upload_services: UploadServices) -> None:
        """Set the upload services for upload tools."""
        self._upload_services = upload_services




    def create_context(self, user_id: int = 0, chat_id: int = 0) -> BotContext:
        """Create a BotContext with services configured.

        This is the preferred way to create a context for tool execution.
        The context includes FolderServices and SchedulerServices which
        provide access to folder operations and task scheduling.
        """
        context = BotContext(query="", user_id=user_id)
        context.services["folder"] = self._folder_services
        if self._scheduler is not None:
            context.services["scheduler"] = SchedulerServices(
                scheduler=self._scheduler,
                chat_id=chat_id,
            )
        if self._upload_services is not None:
            context.services["uploads"] = self._upload_services
        if self._session_manager is not None:
            context.services["session"] = self._session_manager
        context.services["todo"] = self._todo_store
        context.services["calendar"] = self._calendar_store
        return context




    def get_tool_definitions(self) -> list[dict[str, Any]]:
        """Return tool definitions for the Claude API."""
        # All tools (sync and async) are registered on folder_bot
        definitions = folder_bot.get_tools_schema()

        # Add custom tool definitions if available
        if self._custom_tools:
            custom_defs = self._custom_tools.get_tool_definitions()
            definitions.extend(custom_defs)

        return definitions




    def get_tools_requiring_confirmation(self) -> list[str]:
        """Return names of tools that require user confirmation before use."""
        return TOOLS_REQUIRING_CONFIRMATION




    def execute(
        self, tool_name: str, tool_input: dict[str, Any], user_id: int = 0
    ) -> ToolResult:
        """Execute a tool synchronously (wraps async execute for compatibility)."""
        import asyncio

        context = self.create_context(user_id)
        try:
            loop = asyncio.get_running_loop()
        except RuntimeError:
            loop = None

        if loop and loop.is_running():
            # We're already in an async context, can't use asyncio.run()
            # Create a new task and run it
            import concurrent.futures

            with concurrent.futures.ThreadPoolExecutor() as pool:
                future = pool.submit(
                    asyncio.run,
                    self._execute_tool_internal(tool_name, tool_input, context),
                )
                return future.result()
        else:
            return asyncio.run(
                self._execute_tool_internal(tool_name, tool_input, context)
            )




    def execute_direct(
        self, tool_name: str, tool_input: dict[str, Any], user_id: int = 0
    ) -> ToolResult:
        """Execute a tool directly (alias for execute)."""
        return self.execute(tool_name, tool_input, user_id)


    async def _execute_tool_internal(
        self,
        tool_name: str,
        tool_input: dict[str, Any],
        context: BotContext,
    ) -> ToolResult:
        """Internal async tool execution."""
        start_time = time.time()
        user_id = context.user_id

        # Try folder_bot registry (all tools are async)
        if folder_bot.has_tool(tool_name):
            try:
                result = await folder_bot.execute_tool(tool_name, tool_input, context)
                if not isinstance(result, ToolResult):
                    result = ToolResult(
                        content=result.model_dump_json()
                        if hasattr(result, "model_dump_json")
                        else str(result)
                    )
                self._log_tool_call(tool_name, tool_input, result, start_time, user_id)
                return result
            except Exception as e:
                result = ToolResult(content=f"Tool execution error: {e}", is_error=True)
                self._log_tool_call(tool_name, tool_input, result, start_time, user_id)
                return result

        # Try custom tools if available
        if self._custom_tools:
            try:
                custom_result = self._custom_tools.execute(tool_name, tool_input)
                self._log_tool_call(
                    tool_name, tool_input, custom_result, start_time, user_id
                )
                return custom_result
            except Exception as e:
                error_result = ToolResult(
                    content=f"Custom tool execution error: {e}", is_error=True
                )
                self._log_tool_call(
                    tool_name, tool_input, error_result, start_time, user_id
                )
                return error_result

        return ToolResult(content=f"Unknown tool: {tool_name}", is_error=True)

    def _log_tool_call(
        self,
        tool_name: str,
        tool_input: dict[str, Any],
        result: ToolResult,
        start_time: float,
        user_id: int,
    ) -> None:
        """Log a tool call to the activity log."""
        duration_ms = int((time.time() - start_time) * 1000)
        self._activity_logger.log_tool_call(
            tool_name=tool_name,
            tool_input=tool_input,
            result=result.content,
            is_error=result.is_error,
            user_id=user_id,
            duration_ms=duration_ms,
        )



    async def execute_async(
        self,
        tool_name: str,
        tool_input: dict[str, Any],
        context: BotContext,
        chat_id: int = 0,
    ) -> ToolResult:
        """Execute a tool asynchronously with the given context.

        Args:
            tool_name: Name of the tool to execute
            tool_input: Input parameters for the tool
            context: BotContext with user info and services
            chat_id: Telegram chat ID (for scheduler tools)

        Returns:
            ToolResult with the execution result
        """
        start_time = time.time()
        user_id = context.user_id

        # Ensure services have the chat_id
        if "scheduler" in context.services:
            context.services["scheduler"].chat_id = chat_id
        if "uploads" in context.services:
            context.services["uploads"].chat_id = chat_id

        # Try folder_bot registry (all tools are async)
        if folder_bot.has_tool(tool_name):
            try:
                logger.debug(f"Executing tool {tool_name} with input: {tool_input}")
                result = await folder_bot.execute_tool(tool_name, tool_input, context)
                # Ensure we return a ToolResult
                if not isinstance(result, ToolResult):
                    result = ToolResult(
                        content=result.model_dump_json()
                        if hasattr(result, "model_dump_json")
                        else str(result)
                    )
                logger.debug(
                    f"Tool {tool_name} returned: {len(result.content)} chars, "
                    f"is_error={result.is_error}"
                )
                self._log_tool_call(tool_name, tool_input, result, start_time, user_id)
                return result
            except Exception as e:
                logger.exception(f"Tool {tool_name} execution error")
                result = ToolResult(content=f"Tool execution error: {e}", is_error=True)
                self._log_tool_call(tool_name, tool_input, result, start_time, user_id)
                return result

        # Try custom tools if available
        if self._custom_tools:
            try:
                custom_result = self._custom_tools.execute(tool_name, tool_input)
                self._log_tool_call(
                    tool_name, tool_input, custom_result, start_time, user_id
                )
                return custom_result
            except Exception as e:
                error_result = ToolResult(
                    content=f"Custom tool execution error: {e}", is_error=True
                )
                self._log_tool_call(
                    tool_name, tool_input, error_result, start_time, user_id
                )
                return error_result

        return ToolResult(content=f"Unknown tool: {tool_name}", is_error=True)


    @property
    def activity_logger(self) -> ActivityLogger:
        """Expose activity logger for external use."""
        return self._activity_logger