# modules/file_manager.py - 文件管理模块（添加行编辑功能） import os import shutil from pathlib import Path import re from bisect import bisect_right from typing import Any, Optional, Dict, List, Set, Tuple, TYPE_CHECKING from datetime import datetime try: from config import ( MAX_FILE_SIZE, FORBIDDEN_PATHS, FORBIDDEN_ROOT_PATHS, OUTPUT_FORMATS, READ_TOOL_MAX_FILE_SIZE, PROJECT_MAX_STORAGE_BYTES, TERMINAL_SANDBOX_MODE, LINUX_SAFETY, ) except ImportError: # 兼容全局环境中存在同名包的情况 import sys from pathlib import Path project_root = Path(__file__).resolve().parents[1] if str(project_root) not in sys.path: sys.path.insert(0, str(project_root)) from config import ( MAX_FILE_SIZE, FORBIDDEN_PATHS, FORBIDDEN_ROOT_PATHS, OUTPUT_FORMATS, READ_TOOL_MAX_FILE_SIZE, PROJECT_MAX_STORAGE_BYTES, TERMINAL_SANDBOX_MODE, LINUX_SAFETY, ) from modules.container_file_proxy import ContainerFileProxy from modules.host_sandbox_policy import get_macos_writable_paths, get_macos_readable_paths from utils.logger import setup_logger if TYPE_CHECKING: from modules.user_container_manager import ContainerHandle # 临时禁用长度检查 DISABLE_LENGTH_CHECK = True logger = setup_logger(__name__) class CrudMixin: """FileManager crud mixin 能力 mixin。""" def create_file(self, path: str, content: str = "", file_type: str = "txt") -> Dict: """创建文件""" valid, error, full_path = self._validate_path(path) if not valid: return {"success": False, "error": error} # 添加文件扩展名 if not full_path.suffix: full_path = full_path.with_suffix(f".{file_type}") ok, msg = self._ensure_host_access(full_path, "write") if not ok: return {"success": False, "error": msg} relative_path = self._relative_path(full_path) try: # 检查是否允许在根目录创建文件 personalization_config = self._load_personalization_config() allow_root_creation = bool( personalization_config.get("allow_root_file_creation", False) ) if isinstance(personalization_config, dict) else False if full_path.parent == self.project_path and not allow_root_creation: return { "success": False, "error": "禁止在项目根目录直接创建文件，请先创建或选择合适的子目录。", "suggestion": "然后必须**重新**再次创建文件。" } if self._use_container(): result = self._container_call("create_file", { "path": relative_path, "content": "" }) if result.get("success"): print(f"{OUTPUT_FORMATS['file']} 创建文件: {relative_path}") return result # 创建父目录 full_path.parent.mkdir(parents=True, exist_ok=True) with open(full_path, 'w', encoding='utf-8') as f: f.write("") print(f"{OUTPUT_FORMATS['file']} 创建文件: {relative_path}") return { "success": True, "path": relative_path, "size": 0 } except Exception as e: return {"success": False, "error": str(e)} def delete_file(self, path: str) -> Dict: """删除文件""" valid, error, full_path = self._validate_path(path) if not valid: return {"success": False, "error": error} if not full_path.exists(): return {"success": False, "error": "文件不存在"} if not full_path.is_file(): return {"success": False, "error": "不是文件"} ok, msg = self._ensure_host_access(full_path, "write") if not ok: return {"success": False, "error": msg} try: relative_path = self._relative_path(full_path) if self._use_container(): result = self._container_call("delete_file", { "path": relative_path }) if result.get("success"): print(f"{OUTPUT_FORMATS['file']} 删除文件: {relative_path}") return result full_path.unlink() print(f"{OUTPUT_FORMATS['file']} 删除文件: {relative_path}") # 删除文件备注（如果存在） # 这需要通过context_manager处理，但file_manager没有直接访问权限 # 所以返回相对路径，让调用者处理备注删除 return { "success": True, "path": relative_path, "action": "deleted" } except Exception as e: return {"success": False, "error": str(e)} def rename_file(self, old_path: str, new_path: str) -> Dict: """重命名文件""" valid_old, error_old, full_old_path = self._validate_path(old_path) if not valid_old: return {"success": False, "error": error_old} valid_new, error_new, full_new_path = self._validate_path(new_path) if not valid_new: return {"success": False, "error": error_new} if not full_old_path.exists(): return {"success": False, "error": "原文件不存在"} if full_new_path.exists(): return {"success": False, "error": "目标文件已存在"} ok_old, msg_old = self._ensure_host_access(full_old_path, "write") if not ok_old: return {"success": False, "error": msg_old} ok_new, msg_new = self._ensure_host_access(full_new_path, "write") if not ok_new: return {"success": False, "error": msg_new} try: old_relative = self._relative_path(full_old_path) new_relative = self._relative_path(full_new_path) if self._use_container(): result = self._container_call("rename_file", { "old_path": old_relative, "new_path": new_relative }) if result.get("success"): print(f"{OUTPUT_FORMATS['file']} 重命名: {old_relative} -> {new_relative}") return result full_old_path.rename(full_new_path) print(f"{OUTPUT_FORMATS['file']} 重命名: {old_relative} -> {new_relative}") return { "success": True, "old_path": old_relative, "new_path": new_relative, "action": "renamed" } except Exception as e: return {"success": False, "error": str(e)} def create_folder(self, path: str) -> Dict: """创建文件夹""" valid, error, full_path = self._validate_path(path) if not valid: return {"success": False, "error": error} if full_path.exists(): return {"success": False, "error": "文件夹已存在"} ok, msg = self._ensure_host_access(full_path, "write") if not ok: return {"success": False, "error": msg} try: relative_path = self._relative_path(full_path) if self._use_container(): result = self._container_call("create_folder", {"path": relative_path}) if result.get("success"): print(f"{OUTPUT_FORMATS['file']} 创建文件夹: {relative_path}") return result full_path.mkdir(parents=True, exist_ok=True) print(f"{OUTPUT_FORMATS['file']} 创建文件夹: {relative_path}") return {"success": True, "path": relative_path} except Exception as e: return {"success": False, "error": str(e)} def delete_folder(self, path: str) -> Dict: """删除文件夹""" valid, error, full_path = self._validate_path(path) if not valid: return {"success": False, "error": error} if not full_path.exists(): return {"success": False, "error": "文件夹不存在"} if not full_path.is_dir(): return {"success": False, "error": "不是文件夹"} try: relative_path = self._relative_path(full_path) if self._use_container(): result = self._container_call("delete_folder", {"path": relative_path}) if result.get("success"): print(f"{OUTPUT_FORMATS['file']} 删除文件夹: {relative_path}") return result shutil.rmtree(full_path) print(f"{OUTPUT_FORMATS['file']} 删除文件夹: {relative_path}") return {"success": True, "path": relative_path} except Exception as e: return {"success": False, "error": str(e)} def write_file(self, path: str, content: str, mode: str = "w") -> Dict: """ 写入文件 Args: path: 文件路径 content: 内容 mode: 写入模式 - "w"(覆盖), "a"(追加) """ valid, error, full_path = self._validate_path(path) if not valid: return {"success": False, "error": error} ok, msg = self._ensure_host_access(full_path, "write") if not ok: return {"success": False, "error": msg} # === 新增：内容预处理和验证 === if content: # 长度检查 if not DISABLE_LENGTH_CHECK and len(content) > 9999999999: # 100KB限制 return { "success": False, "error": f"内容过长({len(content)}字符)，超过100KB限制", "suggestion": "请分块处理或使用部分修改方式" } # 检查潜在的JSON格式问题 if content.count('"') % 2 != 0: print(f"{OUTPUT_FORMATS['warning']} 检测到奇数个引号，可能存在格式问题") # 检查大量转义字符 if content.count('\\') > len(content) / 20: print(f"{OUTPUT_FORMATS['warning']} 检测到大量转义字符，建议检查内容格式") try: relative_path = self._relative_path(full_path) if PROJECT_MAX_STORAGE_BYTES and self._is_docker_mode(): current_size = self._get_project_size() existing_size = full_path.stat().st_size if full_path.exists() else 0 if mode == "a": projected_total = current_size + len(content) else: projected_total = current_size - existing_size + len(content) if projected_total > PROJECT_MAX_STORAGE_BYTES: return { "success": False, "error": "写入失败：超出项目磁盘配额", "limit_bytes": PROJECT_MAX_STORAGE_BYTES, "project_size_bytes": current_size, "attempt_size_bytes": len(content) } # 记录写入前的原始内容（用于版本控制和前端 diff 显示） original_file: Optional[str] = None try: if full_path.exists(): original_file = full_path.read_text(encoding='utf-8') except Exception: original_file = None if self._use_container(): result = self._container_call("write_file", { "path": relative_path, "content": content, "mode": mode }) if result.get("success"): action = "覆盖" if mode == "w" else "追加" print(f"{OUTPUT_FORMATS['file']} {action}文件: {relative_path}") result["original_file"] = original_file return result # 创建父目录 full_path.parent.mkdir(parents=True, exist_ok=True) with open(full_path, mode, encoding='utf-8') as f: f.write(content) action = "覆盖" if mode == "w" else "追加" print(f"{OUTPUT_FORMATS['file']} {action}文件: {relative_path}") # 读取写入后的实际内容（追加模式需要从文件读取） try: new_file = full_path.read_text(encoding='utf-8') except Exception: new_file = content return { "success": True, "path": relative_path, "size": len(content), "mode": mode, "original_file": original_file, "new_file": new_file, } except Exception as e: return {"success": False, "error": str(e)} def append_file(self, path: str, content: str) -> Dict: """追加内容到文件""" return self.write_file(path, content, mode="a") def clear_file(self, path: str) -> Dict: """清空文件内容""" return self.write_file(path, "", mode="w")