JARVIS/backend/app/agents/tools/builtins/file_tools.py

"""文件操作工具 - Phase 6.4"""

import os
from typing import Any

from app.agents.tools.base import ExternalTool, ReadTool, WriteTool
from app.agents.tools.manifest import (
    PermissionClass,
    SideEffectScope,
    ToolCategory,
)


class GlobTool(ReadTool):
    """文件路径匹配工具

    使用 glob 模式查找文件。
    """

    def __init__(self, root_dir: str = "."):
        super().__init__(
            name="glob",
            description="使用 glob 模式查找文件路径",
            permission_class=PermissionClass.READ,
            side_effect_scope=SideEffectScope.NONE,
            tags=["file", "search", "glob"],
        )
        self.root_dir = root_dir

    def get_parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "pattern": {
                    "type": "string",
                    "description": "Glob 模式，如 **/*.py",
                },
                "root_dir": {
                    "type": "string",
                    "description": "搜索根目录（可选）",
                },
            },
            "required": ["pattern"],
        }

    def get_return_schema(self) -> dict[str, Any]:
        return {
            "type": "array",
            "items": {"type": "string"},
        }

    async def execute(self, pattern: str, root_dir: str | None = None) -> list[str]:
        import glob as glob_module

        root = root_dir or self.root_dir
        return glob_module.glob(pattern, root_dir=root, recursive=True)


class GrepTool(ReadTool):
    """文件内容搜索工具

    在文件中搜索匹配的行。
    """

    def __init__(self):
        super().__init__(
            name="grep",
            description="在文件中搜索匹配的文本行",
            permission_class=PermissionClass.READ,
            side_effect_scope=SideEffectScope.NONE,
            tags=["file", "search", "text"],
        )

    def get_parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "pattern": {
                    "type": "string",
                    "description": "正则表达式模式",
                },
                "paths": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "要搜索的文件路径列表",
                },
                "case_sensitive": {
                    "type": "boolean",
                    "description": "是否区分大小写",
                },
            },
            "required": ["pattern", "paths"],
        }

    def get_return_schema(self) -> dict[str, Any]:
        return {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "file": {"type": "string"},
                    "line": {"type": "integer"},
                    "content": {"type": "string"},
                },
            },
        }

    async def execute(
        self, pattern: str, paths: list[str], case_sensitive: bool = True
    ) -> list[dict[str, Any]]:
        import re

        flags = 0 if case_sensitive else re.IGNORECASE
        regex = re.compile(pattern, flags)
        results = []

        for path in paths:
            if not os.path.isfile(path):
                continue

            try:
                with open(path, "r", encoding="utf-8") as f:
                    for line_num, line in enumerate(f, 1):
                        if regex.search(line):
                            results.append(
                                {
                                    "file": path,
                                    "line": line_num,
                                    "content": line.rstrip(),
                                }
                            )
            except (UnicodeDecodeError, PermissionError):
                continue

        return results


class ReadFileTool(ReadTool):
    """文件读取工具"""

    def __init__(self):
        super().__init__(
            name="read_file",
            description="读取文件内容",
            permission_class=PermissionClass.READ,
            side_effect_scope=SideEffectScope.NONE,
            tags=["file", "read"],
        )

    def get_parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "文件路径",
                },
                "limit": {
                    "type": "integer",
                    "description": "最大行数",
                },
                "offset": {
                    "type": "integer",
                    "description": "起始行号",
                },
            },
            "required": ["path"],
        }

    def get_return_schema(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "content": {"type": "string"},
                "lines": {"type": "integer"},
            },
        }

    async def execute(self, path: str, limit: int | None = None, offset: int = 0) -> dict[str, Any]:
        if not os.path.isfile(path):
            raise FileNotFoundError(f"File not found: {path}")

        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()

        total_lines = len(lines)
        start = max(0, offset)
        end = len(lines) if limit is None else min(start + limit, len(lines))

        content = "".join(lines[start:end])

        return {
            "content": content,
            "lines": total_lines,
            "truncated": limit is not None and end < len(lines),
        }


class WriteFileTool(WriteTool):
    """文件写入工具"""

    def __init__(self):
        super().__init__(
            name="write_file",
            description="写入文件内容",
            permission_class=PermissionClass.WRITE,
            side_effect_scope=SideEffectScope.LOCAL_STATE,
            requires_confirmation=True,
            tags=["file", "write"],
        )

    def get_parameters(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "path": {
                    "type": "string",
                    "description": "文件路径",
                },
                "content": {
                    "type": "string",
                    "description": "文件内容",
                },
                "append": {
                    "type": "boolean",
                    "description": "是否追加模式",
                },
            },
            "required": ["path", "content"],
        }

    def get_return_schema(self) -> dict[str, Any]:
        return {
            "type": "object",
            "properties": {
                "success": {"type": "boolean"},
                "bytes_written": {"type": "integer"},
            },
        }

    async def execute(self, path: str, content: str, append: bool = False) -> dict[str, Any]:
        mode = "a" if append else "w"

        # 确保目录存在
        directory = os.path.dirname(path)
        if directory and not os.path.exists(directory):
            os.makedirs(directory, exist_ok=True)

        with open(path, mode, encoding="utf-8") as f:
            bytes_written = f.write(content)

        return {
            "success": True,
            "bytes_written": bytes_written,
        }