# 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")