"""File Operations Tools - read/write/list/delete files.""" from __future__ import annotations import os import glob from typing import Dict, Any, List from pathlib import Path from .base import Tool, ToolResult, ToolContext, ToolCategory, ToolSafety class FileReadTool(Tool): """Đọc nội dung file.""" category = ToolCategory.FILE safety = ToolSafety.SAFE @property def name(self) -> str: return "file_read" @property def description(self) -> str: return "Đọc nội dung file text. Hỗ trợ offset/limit cho file lớn." @property def parameters(self) -> Dict[str, Any]: return { "type": "object", "properties": { "path": {"type": "string", "description": "Đường dẫn file"}, "offset": {"type": "integer", "description": "Dòng bắt đầu (0-indexed)", "default": 0}, "limit": {"type": "integer", "description": "Số dòng tối đa", "default": 2000}, "encoding": {"type": "string", "default": "utf-8"}, }, "required": ["path"], } def validate_args(self, args: Dict[str, Any]) -> Any: path = args.get("path") if not path: return "Missing required arg: path" full_path = os.path.join(os.getcwd(), path) if not os.path.isabs(path) else path if not os.path.exists(full_path): return f"File not found: {path}" return None def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult: path = args["path"] offset = args.get("offset", 0) limit = args.get("limit", 2000) encoding = args.get("encoding", "utf-8") full_path = path if os.path.isabs(path) else os.path.join(context.working_dir, path) try: with open(full_path, "r", encoding=encoding) as f: lines = f.readlines() total_lines = len(lines) selected = lines[offset:offset + limit] content = "".join(selected) return ToolResult( success=True, output=content, metadata={ "path": full_path, "total_lines": total_lines, "shown_lines": len(selected), "offset": offset, }, ) except Exception as e: return ToolResult(success=False, error=str(e), return_code=1) class FileWriteTool(Tool): """Ghi nội dung vào file (overwrite hoặc append).""" category = ToolCategory.FILE safety = ToolSafety.MODERATE requires_confirmation = True @property def name(self) -> str: return "file_write" @property def description(self) -> str: return "Ghi content vào file. Hỗ trợ append mode." @property def parameters(self) -> Dict[str, Any]: return { "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"}, "append": {"type": "boolean", "default": False}, "create_dirs": {"type": "boolean", "default": True}, "encoding": {"type": "string", "default": "utf-8"}, }, "required": ["path", "content"], } def validate_args(self, args: Dict[str, Any]) -> Any: if not args.get("path"): return "Missing required arg: path" if "content" not in args: return "Missing required arg: content" return None def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult: path = args["path"] content = args["content"] append = args.get("append", False) create_dirs = args.get("create_dirs", True) encoding = args.get("encoding", "utf-8") full_path = path if os.path.isabs(path) else os.path.join(context.working_dir, path) try: if create_dirs: os.makedirs(os.path.dirname(full_path) or ".", exist_ok=True) mode = "a" if append else "w" with open(full_path, mode, encoding=encoding) as f: f.write(content) return ToolResult( success=True, output=f"Wrote {len(content)} chars to {full_path}", artifacts=[full_path], metadata={"bytes": len(content.encode(encoding)), "append": append}, ) except Exception as e: return ToolResult(success=False, error=str(e), return_code=1) class FileListTool(Tool): """Liệt kê files trong thư mục.""" category = ToolCategory.FILE safety = ToolSafety.SAFE @property def name(self) -> str: return "file_list" @property def description(self) -> str: return "Liệt kê files trong thư mục với glob pattern." @property def parameters(self) -> Dict[str, Any]: return { "type": "object", "properties": { "path": {"type": "string", "default": "."}, "pattern": {"type": "string", "default": "*"}, "recursive": {"type": "boolean", "default": False}, "include_hidden": {"type": "boolean", "default": False}, }, } def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult: path = args.get("path", ".") pattern = args.get("pattern", "*") recursive = args.get("recursive", False) include_hidden = args.get("include_hidden", False) full_path = path if os.path.isabs(path) else os.path.join(context.working_dir, path) if recursive: glob_pattern = os.path.join(full_path, "**", pattern) files = glob.glob(glob_pattern, recursive=True) else: glob_pattern = os.path.join(full_path, pattern) files = glob.glob(glob_pattern) if not include_hidden: files = [f for f in files if not os.path.basename(f).startswith(".")] files_sorted = sorted(files) output_lines = [] for f in files_sorted: try: if os.path.isdir(f): output_lines.append(f" 📁 {f}/") else: size = os.path.getsize(f) output_lines.append(f" 📄 {f} ({size} bytes)") except OSError: output_lines.append(f" ❓ {f}") return ToolResult( success=True, output="\n".join(output_lines) or "(empty)", metadata={"count": len(files_sorted), "path": full_path}, ) class FileDeleteTool(Tool): """Xóa file hoặc thư mục.""" category = ToolCategory.FILE safety = ToolSafety.DESTRUCTIVE requires_confirmation = True @property def name(self) -> str: return "file_delete" @property def description(self) -> str: return "Xóa file hoặc thư mục. CẢNH BÁO: không thể undo!" @property def parameters(self) -> Dict[str, Any]: return { "type": "object", "properties": { "path": {"type": "string"}, "recursive": {"type": "boolean", "default": False}, }, "required": ["path"], } def execute(self, args: Dict[str, Any], context: ToolContext) -> ToolResult: import shutil path = args["path"] recursive = args.get("recursive", False) full_path = path if os.path.isabs(path) else os.path.join(context.working_dir, path) if not os.path.exists(full_path): return ToolResult(success=False, error=f"Not found: {full_path}", return_code=1) try: if os.path.isdir(full_path): if not recursive: return ToolResult( success=False, error="Is a directory. Use recursive=True to delete.", return_code=1, ) shutil.rmtree(full_path) else: os.remove(full_path) return ToolResult( success=True, output=f"Deleted: {full_path}", metadata={"deleted_path": full_path}, ) except Exception as e: return ToolResult(success=False, error=str(e), return_code=1)