Spaces:
Paused
Paused
| # File 6: tools/filesystem_tools.py | |
| with open(f"{base_dir}/tools/filesystem_tools.py", "w") as f: | |
| f.write('''"""File System Tools - 8 core tools with parameter multiplication""" | |
| import os | |
| import shutil | |
| import glob | |
| import hashlib | |
| from pathlib import Path | |
| from typing import Dict, List, Optional, Union | |
| from datetime import datetime | |
| from config import Config | |
| from utils import UI, Logger, format_size | |
| class FileSystemTools: | |
| """Comprehensive file system operations with sandbox safety""" | |
| def __init__(self): | |
| self.sandbox = Config.ensure_sandbox() | |
| self.logger = Logger() | |
| def _resolve_path(self, path: str) -> str: | |
| """Resolve path within sandbox""" | |
| if os.path.isabs(path): | |
| abs_sandbox = os.path.abspath(self.sandbox) | |
| abs_path = os.path.abspath(path) | |
| if not abs_path.startswith(abs_sandbox): | |
| raise ValueError(f"Path {path} is outside sandbox") | |
| return abs_path | |
| return os.path.join(self.sandbox, path) | |
| def file_read(self, path: str, lines: str = "", encoding: str = "utf-8", | |
| mode: str = "text", search: str = "", context_lines: int = 0) -> Dict: | |
| """Read file content with multiple modes""" | |
| try: | |
| full_path = self._resolve_path(path) | |
| if not os.path.exists(full_path): | |
| return {"success": False, "error": f"File not found: {path}"} | |
| if mode == "binary": | |
| with open(full_path, "rb") as f: | |
| content = f.read() | |
| return { | |
| "success": True, | |
| "path": path, | |
| "size": len(content), | |
| "hex": content.hex()[:1000], | |
| "mode": "binary" | |
| } | |
| with open(full_path, "r", encoding=encoding, errors="replace") as f: | |
| all_lines = f.readlines() | |
| if search: | |
| matches = [] | |
| for i, line in enumerate(all_lines, 1): | |
| if search.lower() in line.lower(): | |
| start = max(0, i - context_lines - 1) | |
| end = min(len(all_lines), i + context_lines) | |
| context = "".join(all_lines[start:end]) | |
| matches.append({"line": i, "context": context}) | |
| return { | |
| "success": True, | |
| "path": path, | |
| "matches": matches, | |
| "total_matches": len(matches) | |
| } | |
| if lines: | |
| parts = lines.split("-") | |
| if len(parts) == 2: | |
| start = int(parts[0]) - 1 if parts[0] else 0 | |
| end = int(parts[1]) if parts[1] else len(all_lines) | |
| selected = all_lines[start:end] | |
| else: | |
| selected = all_lines | |
| else: | |
| selected = all_lines | |
| content = "".join(selected) | |
| self.logger.log("file_read", {"path": path, "lines": len(selected)}) | |
| return { | |
| "success": True, | |
| "path": path, | |
| "content": content, | |
| "total_lines": len(all_lines), | |
| "returned_lines": len(selected), | |
| "encoding": encoding | |
| } | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def file_write(self, path: str, content: str, encoding: str = "utf-8", | |
| append: bool = False, create_dirs: bool = True) -> Dict: | |
| """Write content to file""" | |
| try: | |
| full_path = self._resolve_path(path) | |
| if create_dirs: | |
| os.makedirs(os.path.dirname(full_path), exist_ok=True) | |
| mode = "a" if append else "w" | |
| with open(full_path, mode, encoding=encoding) as f: | |
| f.write(content) | |
| self.logger.log("file_write", {"path": path, "append": append, "size": len(content)}) | |
| return { | |
| "success": True, | |
| "path": path, | |
| "bytes_written": len(content.encode(encoding)), | |
| "mode": "append" if append else "write" | |
| } | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def file_delete(self, path: str, to_trash: bool = True, confirm: bool = True) -> Dict: | |
| """Delete file or directory""" | |
| try: | |
| full_path = self._resolve_path(path) | |
| if not os.path.exists(full_path): | |
| return {"success": False, "error": f"Path not found: {path}"} | |
| if confirm: | |
| if not UI.confirm(f"Delete {path}?"): | |
| return {"success": False, "error": "User cancelled"} | |
| if to_trash: | |
| trash_dir = os.path.join(self.sandbox, ".trash") | |
| os.makedirs(trash_dir, exist_ok=True) | |
| trash_name = f"{datetime.now().strftime(\'%Y%m%d_%H%M%S\')}_{os.path.basename(path)}" | |
| trash_path = os.path.join(trash_dir, trash_name) | |
| shutil.move(full_path, trash_path) | |
| self.logger.log("file_delete", {"path": path, "trash": trash_path}) | |
| return {"success": True, "path": path, "moved_to": trash_path} | |
| else: | |
| if os.path.isdir(full_path): | |
| shutil.rmtree(full_path) | |
| else: | |
| os.remove(full_path) | |
| self.logger.log("file_delete", {"path": path, "permanent": True}) | |
| return {"success": True, "path": path, "permanent": True} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def file_list(self, path: str = ".", pattern: str = "*", | |
| show_hidden: bool = False, sort_by: str = "name") -> Dict: | |
| """List directory contents""" | |
| try: | |
| full_path = self._resolve_path(path) | |
| if not os.path.exists(full_path): | |
| return {"success": False, "error": f"Directory not found: {path}"} | |
| items = [] | |
| for entry in os.listdir(full_path): | |
| if not show_hidden and entry.startswith("."): | |
| continue | |
| full_entry = os.path.join(full_path, entry) | |
| stat = os.stat(full_entry) | |
| if not glob.fnmatch.fnmatch(entry, pattern): | |
| continue | |
| items.append({ | |
| "name": entry, | |
| "type": "directory" if os.path.isdir(full_entry) else "file", | |
| "size": format_size(stat.st_size) if os.path.isfile(full_entry) else "-", | |
| "size_bytes": stat.st_size if os.path.isfile(full_entry) else 0, | |
| "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(), | |
| "permissions": oct(stat.st_mode)[-3:] | |
| }) | |
| if sort_by == "size": | |
| items.sort(key=lambda x: x["size_bytes"], reverse=True) | |
| elif sort_by == "date": | |
| items.sort(key=lambda x: x["modified"], reverse=True) | |
| else: | |
| items.sort(key=lambda x: x["name"].lower()) | |
| self.logger.log("file_list", {"path": path, "count": len(items)}) | |
| return { | |
| "success": True, | |
| "path": path, | |
| "items": items, | |
| "total": len(items), | |
| "directories": sum(1 for i in items if i["type"] == "directory"), | |
| "files": sum(1 for i in items if i["type"] == "file") | |
| } | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def file_search(self, path: str = ".", name_pattern: str = "", | |
| content_pattern: str = "", file_types: List[str] = None, | |
| max_depth: int = 5, max_results: int = 50) -> Dict: | |
| """Search for files by name or content""" | |
| try: | |
| full_path = self._resolve_path(path) | |
| results = [] | |
| for root, dirs, files in os.walk(full_path): | |
| depth = root[len(full_path):].count(os.sep) | |
| if depth > max_depth: | |
| del dirs[:] | |
| continue | |
| for file in files: | |
| if len(results) >= max_results: | |
| break | |
| if name_pattern and not glob.fnmatch.fnmatch(file, name_pattern): | |
| continue | |
| if file_types and not any(file.endswith(ext) for ext in file_types): | |
| continue | |
| file_path = os.path.join(root, file) | |
| rel_path = os.path.relpath(file_path, self.sandbox) | |
| result = { | |
| "path": rel_path, | |
| "name": file, | |
| "size": os.path.getsize(file_path) | |
| } | |
| if content_pattern: | |
| try: | |
| with open(file_path, "r", encoding="utf-8", errors="ignore") as f: | |
| content = f.read() | |
| if content_pattern.lower() in content.lower(): | |
| lines = content.split("\n") | |
| matches = [] | |
| for i, line in enumerate(lines, 1): | |
| if content_pattern.lower() in line.lower(): | |
| matches.append({"line": i, "text": line.strip()}) | |
| result["content_matches"] = matches | |
| result["match_count"] = len(matches) | |
| else: | |
| continue | |
| except: | |
| continue | |
| results.append(result) | |
| if len(results) >= max_results: | |
| break | |
| self.logger.log("file_search", {"path": path, "results": len(results)}) | |
| return { | |
| "success": True, | |
| "query": {"name": name_pattern, "content": content_pattern}, | |
| "results": results, | |
| "total": len(results) | |
| } | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def file_copy_move(self, source: str, destination: str, | |
| operation: str = "copy", overwrite: bool = False) -> Dict: | |
| """Copy or move files/directories""" | |
| try: | |
| src_path = self._resolve_path(source) | |
| dst_path = self._resolve_path(destination) | |
| if not os.path.exists(src_path): | |
| return {"success": False, "error": f"Source not found: {source}"} | |
| if os.path.exists(dst_path) and not overwrite: | |
| if not UI.confirm(f"{destination} exists. Overwrite?"): | |
| return {"success": False, "error": "User cancelled"} | |
| os.makedirs(os.path.dirname(dst_path), exist_ok=True) | |
| if operation == "copy": | |
| if os.path.isdir(src_path): | |
| shutil.copytree(src_path, dst_path, dirs_exist_ok=overwrite) | |
| else: | |
| shutil.copy2(src_path, dst_path) | |
| else: | |
| shutil.move(src_path, dst_path) | |
| self.logger.log(f"file_{operation}", {"source": source, "destination": destination}) | |
| return { | |
| "success": True, | |
| "operation": operation, | |
| "source": source, | |
| "destination": destination | |
| } | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def file_info(self, path: str) -> Dict: | |
| """Get detailed file information""" | |
| try: | |
| full_path = self._resolve_path(path) | |
| if not os.path.exists(full_path): | |
| return {"success": False, "error": f"Path not found: {path}"} | |
| stat = os.stat(full_path) | |
| file_hash = "" | |
| if os.path.isfile(full_path): | |
| hasher = hashlib.sha256() | |
| with open(full_path, "rb") as f: | |
| for chunk in iter(lambda: f.read(8192), b""): | |
| hasher.update(chunk) | |
| file_hash = hasher.hexdigest() | |
| info = { | |
| "success": True, | |
| "path": path, | |
| "absolute_path": full_path, | |
| "type": "directory" if os.path.isdir(full_path) else "file", | |
| "size": format_size(stat.st_size), | |
| "size_bytes": stat.st_size, | |
| "created": datetime.fromtimestamp(stat.st_ctime).isoformat(), | |
| "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(), | |
| "accessed": datetime.fromtimestamp(stat.st_atime).isoformat(), | |
| "permissions": oct(stat.st_mode)[-3:], | |
| "owner_uid": stat.st_uid, | |
| "group_gid": stat.st_gid, | |
| "sha256": file_hash | |
| } | |
| self.logger.log("file_info", {"path": path}) | |
| return info | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| ''') | |
| print("tools/filesystem_tools.py done") |