agents/tools/file_tools.py

"""File operation tools for reading and writing files."""

import asyncio
import glob
import os
from pathlib import Path

from .base import Tool


class FileReadTool(Tool):
    """Tool for reading files and listing directories."""

    def __init__(self):
        super().__init__(
            name="file_read",
            description="""
            Read files or list directory contents.

            Operations:
            - read: Read the contents of a file
            - list: List files in a directory
            """,
            input_schema={
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["read", "list"],
                        "description": "File operation to perform",
                    },
                    "path": {
                        "type": "string",
                        "description": "File path for read or directory path",
                    },
                    "max_lines": {
                        "type": "integer",
                        "description": "Maximum lines to read (0 means no limit)",
                    },
                    "pattern": {
                        "type": "string",
                        "description": "File pattern to match",
                    },
                },
                "required": ["operation", "path"],
            },
        )

    async def execute(
        self,
        operation: str,
        path: str,
        max_lines: int = 0,
        pattern: str = "*",
    ) -> str:
        """Execute a file read operation.

        Args:
            operation: The operation to perform (read or list)
            path: The file or directory path
            max_lines: Maximum lines to read (for read operation, 0 means no limit)
            pattern: File pattern to match (for list operation)

        Returns:
            Result of the operation as string
        """
        if operation == "read":
            return await self._read_file(path, max_lines)
        elif operation == "list":
            return await self._list_files(path, pattern)
        else:
            return f"Error: Unsupported operation '{operation}'"

    async def _read_file(self, path: str, max_lines: int = 0) -> str:
        """Read a file from disk.
        
        Args:
            path: Path to the file to read
            max_lines: Maximum number of lines to read (0 means read entire file)
        """
        try:
            file_path = Path(path)

            if not file_path.exists():
                return f"Error: File not found at {path}"
            if not file_path.is_file():
                return f"Error: {path} is not a file"

            def read_sync():
                with open(file_path, encoding="utf-8", errors="replace") as f:
                    if max_lines > 0:
                        lines = []
                        for i, line in enumerate(f):
                            if i >= max_lines:
                                break
                            lines.append(line)
                        return "".join(lines)
                    return f.read()

            return await asyncio.to_thread(read_sync)
        except Exception as e:
            return f"Error reading {path}: {str(e)}"

    async def _list_files(self, directory: str, pattern: str = "*") -> str:
        """List files in a directory."""
        try:
            dir_path = Path(directory)

            if not dir_path.exists():
                return f"Error: Directory not found at {directory}"
            if not dir_path.is_dir():
                return f"Error: {directory} is not a directory"

            def list_sync():
                search_pattern = f"{directory}/{pattern}"
                files = glob.glob(search_pattern)

                if not files:
                    return f"No files found matching {directory}/{pattern}"

                file_list = []
                for file_path in sorted(files):
                    path_obj = Path(file_path)
                    rel_path = str(file_path).replace(str(dir_path) + "/", "")

                    if path_obj.is_dir():
                        file_list.append(f"📁 {rel_path}/")
                    else:
                        file_list.append(f"📄 {rel_path}")

                return "\n".join(file_list)

            return await asyncio.to_thread(list_sync)
        except Exception as e:
            return f"Error listing files in {directory}: {str(e)}"


class FileWriteTool(Tool):
    """Tool for writing and editing files."""

    def __init__(self):
        super().__init__(
            name="file_write",
            description="""
            Write or edit files.

            Operations:
            - write: Create or completely replace a file
            - edit: Make targeted changes to parts of a file
            """,
            input_schema={
                "type": "object",
                "properties": {
                    "operation": {
                        "type": "string",
                        "enum": ["write", "edit"],
                        "description": "File operation to perform",
                    },
                    "path": {
                        "type": "string",
                        "description": "File path to write to or edit",
                    },
                    "content": {
                        "type": "string",
                        "description": "Content to write",
                    },
                    "old_text": {
                        "type": "string",
                        "description": "Text to replace (for edit operation)",
                    },
                    "new_text": {
                        "type": "string",
                        "description": "Replacement text (for edit operation)",
                    },
                },
                "required": ["operation", "path"],
            },
        )

    async def execute(
        self,
        operation: str,
        path: str,
        content: str = "",
        old_text: str = "",
        new_text: str = "",
    ) -> str:
        """Execute a file write operation.

        Args:
            operation: The operation to perform (write or edit)
            path: The file path
            content: Content to write (for write operation)
            old_text: Text to replace (for edit operation)
            new_text: Replacement text (for edit operation)

        Returns:
            Result of the operation as string
        """
        if operation == "write":
            if not content:
                return "Error: content parameter is required"
            return await self._write_file(path, content)
        elif operation == "edit":
            if not old_text or not new_text:
                return (
                    "Error: both old_text and new_text parameters "
                    "are required for edit operation"
                )
            return await self._edit_file(path, old_text, new_text)
        else:
            return f"Error: Unsupported operation '{operation}'"

    async def _write_file(self, path: str, content: str) -> str:
        """Write content to a file."""
        try:
            file_path = Path(path)
            os.makedirs(file_path.parent, exist_ok=True)

            def write_sync():
                with open(file_path, "w", encoding="utf-8") as f:
                    f.write(content)
                return (
                    f"Successfully wrote {len(content)} "
                    f"characters to {path}"
                )

            return await asyncio.to_thread(write_sync)
        except Exception as e:
            return f"Error writing to {path}: {str(e)}"

    async def _edit_file(self, path: str, old_text: str, new_text: str) -> str:
        """Make targeted changes to a file."""
        try:
            file_path = Path(path)

            if not file_path.exists():
                return f"Error: File not found at {path}"
            if not file_path.is_file():
                return f"Error: {path} is not a file"

            def edit_sync():
                try:
                    with open(
                        file_path, encoding="utf-8", errors="replace"
                    ) as f:
                        content = f.read()

                    if old_text not in content:
                        return (
                            f"Error: The specified text was not "
                            f"found in {path}"
                        )

                    # Count occurrences to warn about multiple matches
                    count = content.count(old_text)
                    if count > 1:
                        # Edit with warning about multiple occurrences
                        new_content = content.replace(old_text, new_text)
                        with open(file_path, "w", encoding="utf-8") as f:
                            f.write(new_content)
                        return (
                            f"Warning: Found {count} occurrences. "
                            f"All were replaced in {path}"
                        )
                    else:
                        # One occurrence, straightforward replacement
                        new_content = content.replace(old_text, new_text)
                        with open(file_path, "w", encoding="utf-8") as f:
                            f.write(new_content)
                        return f"Successfully edited {path}"
                except UnicodeDecodeError:
                    return f"Error: {path} appears to be a binary file"

            return await asyncio.to_thread(edit_sync)
        except Exception as e:
            return f"Error editing {path}: {str(e)}"