src/tools/file_tools.py

"""
File Tools for SPARKNET
Tools for file system operations
"""

from pathlib import Path
from typing import Optional
from loguru import logger
from .base_tool import BaseTool, ToolResult
import json


class FileReaderTool(BaseTool):
    """Tool for reading files."""

    def __init__(self):
        super().__init__(
            name="file_reader",
            description="Read contents of a file from the file system",
        )
        self.add_parameter("file_path", "str", "Path to the file to read", required=True)
        self.add_parameter("encoding", "str", "File encoding", required=False, default="utf-8")

    async def execute(self, file_path: str, encoding: str = "utf-8", **kwargs) -> ToolResult:
        """
        Read file contents.

        Args:
            file_path: Path to file
            encoding: File encoding

        Returns:
            ToolResult with file contents
        """
        try:
            path = Path(file_path)

            if not path.exists():
                return ToolResult(
                    success=False,
                    output=None,
                    error=f"File not found: {file_path}",
                )

            if not path.is_file():
                return ToolResult(
                    success=False,
                    output=None,
                    error=f"Path is not a file: {file_path}",
                )

            with open(path, "r", encoding=encoding) as f:
                contents = f.read()

            return ToolResult(
                success=True,
                output=contents,
                metadata={
                    "file_path": str(path.absolute()),
                    "size_bytes": len(contents),
                    "encoding": encoding,
                },
            )

        except Exception as e:
            return ToolResult(
                success=False,
                output=None,
                error=f"Error reading file: {str(e)}",
            )


class FileWriterTool(BaseTool):
    """Tool for writing files."""

    def __init__(self):
        super().__init__(
            name="file_writer",
            description="Write contents to a file",
        )
        self.add_parameter("file_path", "str", "Path to the file to write", required=True)
        self.add_parameter("content", "str", "Content to write to file", required=True)
        self.add_parameter("encoding", "str", "File encoding", required=False, default="utf-8")
        self.add_parameter("append", "bool", "Append to file instead of overwriting", required=False, default=False)

    async def execute(
        self,
        file_path: str,
        content: str,
        encoding: str = "utf-8",
        append: bool = False,
        **kwargs,
    ) -> ToolResult:
        """
        Write content to file.

        Args:
            file_path: Path to file
            content: Content to write
            encoding: File encoding
            append: Whether to append

        Returns:
            ToolResult
        """
        try:
            path = Path(file_path)

            # Create parent directories if needed
            path.parent.mkdir(parents=True, exist_ok=True)

            mode = "a" if append else "w"
            with open(path, mode, encoding=encoding) as f:
                f.write(content)

            return ToolResult(
                success=True,
                output=f"Successfully wrote to {file_path}",
                metadata={
                    "file_path": str(path.absolute()),
                    "bytes_written": len(content.encode(encoding)),
                    "mode": "append" if append else "write",
                },
            )

        except Exception as e:
            return ToolResult(
                success=False,
                output=None,
                error=f"Error writing file: {str(e)}",
            )


class FileSearchTool(BaseTool):
    """Tool for searching files."""

    def __init__(self):
        super().__init__(
            name="file_search",
            description="Search for files matching a pattern",
        )
        self.add_parameter("directory", "str", "Directory to search in", required=True)
        self.add_parameter("pattern", "str", "File pattern to match (e.g., '*.txt')", required=True)
        self.add_parameter("recursive", "bool", "Search recursively", required=False, default=True)

    async def execute(
        self,
        directory: str,
        pattern: str,
        recursive: bool = True,
        **kwargs,
    ) -> ToolResult:
        """
        Search for files.

        Args:
            directory: Directory to search
            pattern: File pattern
            recursive: Search recursively

        Returns:
            ToolResult with list of matching files
        """
        try:
            path = Path(directory)

            if not path.exists():
                return ToolResult(
                    success=False,
                    output=None,
                    error=f"Directory not found: {directory}",
                )

            if recursive:
                files = list(path.rglob(pattern))
            else:
                files = list(path.glob(pattern))

            file_paths = [str(f.absolute()) for f in files if f.is_file()]

            return ToolResult(
                success=True,
                output=file_paths,
                metadata={
                    "directory": str(path.absolute()),
                    "pattern": pattern,
                    "count": len(file_paths),
                    "recursive": recursive,
                },
            )

        except Exception as e:
            return ToolResult(
                success=False,
                output=None,
                error=f"Error searching files: {str(e)}",
            )


class DirectoryListTool(BaseTool):
    """Tool for listing directory contents."""

    def __init__(self):
        super().__init__(
            name="directory_list",
            description="List contents of a directory",
        )
        self.add_parameter("directory", "str", "Directory to list", required=True)
        self.add_parameter("include_hidden", "bool", "Include hidden files", required=False, default=False)

    async def execute(
        self,
        directory: str,
        include_hidden: bool = False,
        **kwargs,
    ) -> ToolResult:
        """
        List directory contents.

        Args:
            directory: Directory to list
            include_hidden: Include hidden files

        Returns:
            ToolResult with directory contents
        """
        try:
            path = Path(directory)

            if not path.exists():
                return ToolResult(
                    success=False,
                    output=None,
                    error=f"Directory not found: {directory}",
                )

            if not path.is_dir():
                return ToolResult(
                    success=False,
                    output=None,
                    error=f"Path is not a directory: {directory}",
                )

            items = []
            for item in path.iterdir():
                if not include_hidden and item.name.startswith("."):
                    continue

                items.append({
                    "name": item.name,
                    "path": str(item.absolute()),
                    "type": "directory" if item.is_dir() else "file",
                    "size": item.stat().st_size if item.is_file() else None,
                })

            return ToolResult(
                success=True,
                output=items,
                metadata={
                    "directory": str(path.absolute()),
                    "count": len(items),
                },
            )

        except Exception as e:
            return ToolResult(
                success=False,
                output=None,
                error=f"Error listing directory: {str(e)}",
            )