""" 文件操作工具 提供安全的文件读写、目录操作、搜索功能 """ import os import shutil import glob as glob_module from typing import Optional, List, Dict, Any from pathlib import Path class FileToolConfig: """文件工具配置""" # 允许访问的基础目录(限制在项目内) ALLOWED_BASE_DIRS = [ "account", # 用户工作区 "temp", # 临时文件 ] MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB MAX_SEARCH_RESULTS = 100 def _resolve_safe_path(base_path: str, relative_path: str) -> str: """ 解析安全的文件路径 确保路径不会超出基础目录 """ # 规范化路径 full_path = os.path.normpath(os.path.join(base_path, relative_path)) # 检查是否在允许的基础目录内 path_parts = Path(full_path).parts if len(path_parts) < 2: raise ValueError("Invalid path: too short") base_dir = path_parts[0] if base_dir not in FileToolConfig.ALLOWED_BASE_DIRS and not base_dir.endswith(".py"): # 允许 account 下的子目录 if len(path_parts) >= 2 and path_parts[0] != "account": raise ValueError(f"Path not in allowed directories: {base_dir}") return full_path def read_file(file_path: str, encoding: str = "utf-8") -> Dict[str, Any]: """ 读取文件内容 Args: file_path: 文件路径 encoding: 文件编码 Returns: 文件内容 """ try: # 安全检查 full_path = _resolve_safe_path("", file_path) if not os.path.exists(full_path): return { "success": False, "error": f"File not found: {file_path}" } if not os.path.isfile(full_path): return { "success": False, "error": f"Not a file: {file_path}" } # 检查文件大小 file_size = os.path.getsize(full_path) if file_size > FileToolConfig.MAX_FILE_SIZE: return { "success": False, "error": f"File too large: {file_size} bytes (max {FileToolConfig.MAX_FILE_SIZE})" } # 读取内容 with open(full_path, "r", encoding=encoding, errors="replace") as f: content = f.read() return { "success": True, "content": content, "file_path": file_path, "size": file_size, "encoding": encoding } except ValueError as e: return { "success": False, "error": str(e) } except Exception as e: return { "success": False, "error": f"Read error: {str(e)}" } def write_file(file_path: str, content: str, encoding: str = "utf-8") -> Dict[str, Any]: """ 写入文件内容 Args: file_path: 文件路径 content: 文件内容 encoding: 文件编码 Returns: 写入结果 """ try: # 安全检查 full_path = _resolve_safe_path("", file_path) # 检查内容大小 if len(content.encode(encoding)) > FileToolConfig.MAX_FILE_SIZE: return { "success": False, "error": f"Content too large: {len(content)} bytes" } # 确保目录存在 os.makedirs(os.path.dirname(full_path), exist_ok=True) # 写入内容 with open(full_path, "w", encoding=encoding) as f: f.write(content) return { "success": True, "file_path": file_path, "bytes_written": len(content.encode(encoding)) } except ValueError as e: return { "success": False, "error": str(e) } except Exception as e: return { "success": False, "error": f"Write error: {str(e)}" } def list_dir(dir_path: str = ".") -> Dict[str, Any]: """ 列出目录内容 Args: dir_path: 目录路径 Returns: 目录内容列表 """ try: full_path = _resolve_safe_path("", dir_path) if not os.path.exists(full_path): return { "success": False, "error": f"Directory not found: {dir_path}" } if not os.path.isdir(full_path): return { "success": False, "error": f"Not a directory: {dir_path}" } items = [] for item in os.listdir(full_path): item_path = os.path.join(full_path, item) is_dir = os.path.isdir(item_path) try: size = 0 if is_dir else os.path.getsize(item_path) except: size = 0 items.append({ "name": item, "type": "directory" if is_dir else "file", "size": size }) return { "success": True, "path": dir_path, "items": items, "count": len(items) } except ValueError as e: return { "success": False, "error": str(e) } except Exception as e: return { "success": False, "error": f"List error: {str(e)}" } def delete_file(file_path: str) -> Dict[str, Any]: """ 删除文件或目录 Args: file_path: 文件或目录路径 Returns: 删除结果 """ try: full_path = _resolve_safe_path("", file_path) if not os.path.exists(full_path): return { "success": False, "error": f"Path not found: {file_path}" } # 删除 if os.path.isfile(full_path): os.remove(full_path) elif os.path.isdir(full_path): shutil.rmtree(full_path) return { "success": True, "file_path": file_path, "deleted": True } except ValueError as e: return { "success": False, "error": str(e) } except Exception as e: return { "success": False, "error": f"Delete error: {str(e)}" } def search_files( directory: str, pattern: str = "*", content_pattern: Optional[str] = None, file_only: bool = True ) -> Dict[str, Any]: """ 搜索文件 Args: directory: 搜索目录 pattern: 文件名匹配模式 (glob) content_pattern: 文件内容匹配模式 (可选) file_only: 是否只返回文件 Returns: 搜索结果 """ try: full_path = _resolve_safe_path("", directory) if not os.path.exists(full_path) or not os.path.isdir(full_path): return { "success": False, "error": f"Invalid directory: {directory}" } results = [] # 按文件名搜索 for match in glob_module.glob(os.path.join(full_path, "**", pattern), recursive=True): if file_only and os.path.isdir(match): continue rel_path = os.path.relpath(match, full_path) # 如果没有内容搜索,直接添加 if not content_pattern: results.append({ "path": rel_path, "name": os.path.basename(match), "type": "directory" if os.path.isdir(match) else "file" }) continue # 内容搜索 if os.path.isfile(match): try: # 检查文件大小 if os.path.getsize(match) > FileToolConfig.MAX_FILE_SIZE: continue with open(match, "r", encoding="utf-8", errors="ignore") as f: content = f.read() if content_pattern.lower() in content.lower(): results.append({ "path": rel_path, "name": os.path.basename(match), "type": "file", "match": content_pattern }) except: continue # 限制结果数量 if len(results) > FileToolConfig.MAX_SEARCH_RESULTS: results = results[:FileToolConfig.MAX_SEARCH_RESULTS] return { "success": True, "directory": directory, "pattern": pattern, "results": results, "count": len(results) } except ValueError as e: return { "success": False, "error": str(e) } except Exception as e: return { "success": False, "error": f"Search error: {str(e)}" } # 工具定义 READ_FILE_TOOL = { "name": "read_file", "description": "Read the contents of a file from the filesystem.", "parameters": { "type": "object", "properties": { "file_path": { "type": "string", "description": "The path to the file to read" }, "encoding": { "type": "string", "description": "File encoding (default: utf-8)", "default": "utf-8" } }, "required": ["file_path"] } } WRITE_FILE_TOOL = { "name": "write_file", "description": "Write content to a file. Creates the file if it doesn't exist, overwrites if it does.", "parameters": { "type": "object", "properties": { "file_path": { "type": "string", "description": "The path to the file to write" }, "content": { "type": "string", "description": "The content to write to the file" }, "encoding": { "type": "string", "description": "File encoding (default: utf-8)", "default": "utf-8" } }, "required": ["file_path", "content"] } } LIST_DIR_TOOL = { "name": "list_dir", "description": "List the contents of a directory.", "parameters": { "type": "object", "properties": { "dir_path": { "type": "string", "description": "The path to the directory to list (default: current directory)", "default": "." } } } } DELETE_FILE_TOOL = { "name": "delete_file", "description": "Delete a file or directory.", "parameters": { "type": "object", "properties": { "file_path": { "type": "string", "description": "The path to the file or directory to delete" } }, "required": ["file_path"] } } SEARCH_FILES_TOOL = { "name": "search_files", "description": "Search for files by name pattern or content.", "parameters": { "type": "object", "properties": { "directory": { "type": "string", "description": "The directory to search in" }, "pattern": { "type": "string", "description": "Glob pattern for file names (e.g., '*.py', '*.txt')", "default": "*" }, "content_pattern": { "type": "string", "description": "Optional: search for files containing this text in their content" }, "file_only": { "type": "boolean", "description": "Only return files, not directories", "default": True } }, "required": ["directory"] } }