JOJO
f661dc6456
- Split server/chat.py, server/status.py, server/tasks.py into sub-packages
- Split utils/context_manager.py, utils/conversation_manager.py into mixin packages
- Split modules/file_manager.py, persistent_terminal.py, terminal_ops.py, mcp_client_manager.py into packages
- Split core/main_terminal_parts/context.py into base + mixins
- Split static/src/app/methods/{ui,message,upload,taskPolling,conversation} into sub-modules
- Fix flattened method exports, dynamic import paths, missing cross-module imports
- Add missing permission/network/execution mode constants and _PERMISSION_MODE_LABEL
- Update AGENTS.md, CLAUDE.md and project memories to reflect new structure
354 lines
12 KiB
Python
354 lines
12 KiB
Python
# 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 ReadMixin:
|
||
"""FileManager read mixin 能力 mixin。"""
|
||
|
||
def _read_text_lines(
|
||
self,
|
||
full_path: Path,
|
||
*,
|
||
size_limit: Optional[int] = None,
|
||
encoding: str = "utf-8",
|
||
) -> Dict:
|
||
"""读取UTF-8文本并返回行列表。"""
|
||
try:
|
||
file_size = full_path.stat().st_size
|
||
except FileNotFoundError:
|
||
return {"success": False, "error": "文件不存在"}
|
||
|
||
if size_limit and file_size > size_limit:
|
||
return {
|
||
"success": False,
|
||
"error": f"文件太大 ({file_size / 1024 / 1024:.2f}MB > {size_limit / 1024 / 1024}MB)"
|
||
}
|
||
|
||
try:
|
||
with open(full_path, 'r', encoding=encoding) as f:
|
||
lines = f.readlines()
|
||
except UnicodeDecodeError:
|
||
return {
|
||
"success": False,
|
||
"error": "文件不是 UTF-8 文本，无法直接读取，请改用 run_command 调用合适的解析工具或 Python 解释器。"
|
||
}
|
||
except Exception as e:
|
||
return {"success": False, "error": f"读取文件失败: {e}"}
|
||
|
||
content = "".join(lines)
|
||
return {
|
||
"success": True,
|
||
"content": content,
|
||
"lines": lines,
|
||
"size": file_size
|
||
}
|
||
|
||
def read_file(self, path: str) -> Dict:
|
||
"""读取文件内容（兼容旧逻辑，限制为 MAX_FILE_SIZE）。"""
|
||
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, "read")
|
||
if not ok:
|
||
return {"success": False, "error": msg}
|
||
|
||
if self._use_container():
|
||
relative_path = self._relative_path(full_path)
|
||
result = self._container_call("read_file", {
|
||
"path": relative_path,
|
||
"size_limit": MAX_FILE_SIZE
|
||
})
|
||
return result
|
||
|
||
result = self._read_text_lines(full_path, size_limit=MAX_FILE_SIZE)
|
||
if not result["success"]:
|
||
return result
|
||
|
||
relative_path = self._relative_path(full_path)
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"content": result["content"],
|
||
"size": result["size"]
|
||
}
|
||
|
||
def read_text_segment(
|
||
self,
|
||
path: str,
|
||
*,
|
||
start_line: Optional[int] = None,
|
||
end_line: Optional[int] = None,
|
||
size_limit: Optional[int] = None
|
||
) -> 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, "read")
|
||
if not ok:
|
||
return {"success": False, "error": msg}
|
||
|
||
if self._use_container():
|
||
relative_path = self._relative_path(full_path)
|
||
result = self._container_call("read_text_segment", {
|
||
"path": relative_path,
|
||
"start_line": start_line,
|
||
"end_line": end_line,
|
||
"size_limit": size_limit or READ_TOOL_MAX_FILE_SIZE
|
||
})
|
||
return result
|
||
|
||
if self._use_container():
|
||
relative_path = self._relative_path(full_path)
|
||
return self._container_call("search_text", {
|
||
"path": relative_path,
|
||
"query": query,
|
||
"max_matches": max_matches,
|
||
"context_before": context_before,
|
||
"context_after": context_after,
|
||
"case_sensitive": case_sensitive,
|
||
})
|
||
|
||
if self._use_container():
|
||
relative_path = self._relative_path(full_path)
|
||
return self._container_call("extract_segments", {
|
||
"path": relative_path,
|
||
"segments": segments,
|
||
"size_limit": size_limit or READ_TOOL_MAX_FILE_SIZE
|
||
})
|
||
|
||
result = self._read_text_lines(
|
||
full_path,
|
||
size_limit=size_limit or READ_TOOL_MAX_FILE_SIZE
|
||
)
|
||
if not result["success"]:
|
||
return result
|
||
|
||
lines = result["lines"]
|
||
total_lines = len(lines)
|
||
if total_lines == 0:
|
||
relative_path = self._relative_path(full_path)
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"content": "",
|
||
"size": result["size"],
|
||
"line_start": 0,
|
||
"line_end": 0,
|
||
"total_lines": 0,
|
||
}
|
||
start = start_line if start_line and start_line > 0 else 1
|
||
end = end_line if end_line and end_line >= start else total_lines
|
||
if start > total_lines:
|
||
return {"success": False, "error": "起始行超出文件长度"}
|
||
end = min(end, total_lines)
|
||
|
||
selected_lines = lines[start - 1 : end]
|
||
content = "".join(selected_lines)
|
||
|
||
relative_path = self._relative_path(full_path)
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"content": content,
|
||
"size": result["size"],
|
||
"line_start": start,
|
||
"line_end": end,
|
||
"total_lines": total_lines
|
||
}
|
||
|
||
def search_text(
|
||
self,
|
||
path: str,
|
||
*,
|
||
query: str,
|
||
max_matches: int,
|
||
context_before: int,
|
||
context_after: int,
|
||
case_sensitive: bool = False,
|
||
size_limit: Optional[int] = None
|
||
) -> Dict:
|
||
"""在文件中搜索关键词，返回合并后的窗口。"""
|
||
if not query:
|
||
return {"success": False, "error": "缺少搜索关键词"}
|
||
|
||
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, "read")
|
||
if not ok:
|
||
return {"success": False, "error": msg}
|
||
|
||
result = self._read_text_lines(
|
||
full_path,
|
||
size_limit=size_limit or READ_TOOL_MAX_FILE_SIZE
|
||
)
|
||
if not result["success"]:
|
||
return result
|
||
|
||
lines = result["lines"]
|
||
total_lines = len(lines)
|
||
matches = []
|
||
query_text = query if case_sensitive else query.lower()
|
||
|
||
def contains(haystack: str) -> bool:
|
||
target = haystack if case_sensitive else haystack.lower()
|
||
return query_text in target
|
||
|
||
for idx, line in enumerate(lines, start=1):
|
||
if contains(line):
|
||
window_start = max(1, idx - context_before)
|
||
window_end = min(total_lines, idx + context_after)
|
||
|
||
if matches and window_start <= matches[-1]["line_end"]:
|
||
matches[-1]["line_end"] = max(matches[-1]["line_end"], window_end)
|
||
matches[-1]["hits"].append(idx)
|
||
else:
|
||
if len(matches) >= max_matches:
|
||
break
|
||
matches.append({
|
||
"line_start": window_start,
|
||
"line_end": window_end,
|
||
"hits": [idx]
|
||
})
|
||
|
||
relative_path = self._relative_path(full_path)
|
||
for window in matches:
|
||
snippet_lines = lines[window["line_start"] - 1 : window["line_end"]]
|
||
window["snippet"] = "".join(snippet_lines)
|
||
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"size": result["size"],
|
||
"total_lines": total_lines,
|
||
"matches": matches
|
||
}
|
||
|
||
def extract_segments(
|
||
self,
|
||
path: str,
|
||
segments: List[Dict],
|
||
*,
|
||
size_limit: Optional[int] = None
|
||
) -> Dict:
|
||
"""根据多个行区间提取内容。"""
|
||
if not segments:
|
||
return {"success": False, "error": "缺少要提取的行区间"}
|
||
|
||
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, "read")
|
||
if not ok:
|
||
return {"success": False, "error": msg}
|
||
|
||
result = self._read_text_lines(
|
||
full_path,
|
||
size_limit=size_limit or READ_TOOL_MAX_FILE_SIZE
|
||
)
|
||
if not result["success"]:
|
||
return result
|
||
|
||
lines = result["lines"]
|
||
total_lines = len(lines)
|
||
extracted = []
|
||
|
||
for item in segments:
|
||
if not isinstance(item, dict):
|
||
return {"success": False, "error": "segments 数组中的每一项都必须是对象"}
|
||
start_line = item.get("start_line")
|
||
end_line = item.get("end_line")
|
||
label = item.get("label")
|
||
if start_line is None or end_line is None:
|
||
return {"success": False, "error": "所有区间都必须包含 start_line 和 end_line"}
|
||
if start_line <= 0 or end_line < start_line:
|
||
return {"success": False, "error": "行区间不合法"}
|
||
if start_line > total_lines:
|
||
return {"success": False, "error": f"区间起点 {start_line} 超出文件行数"}
|
||
end_line = min(end_line, total_lines)
|
||
snippet = "".join(lines[start_line - 1 : end_line])
|
||
extracted.append({
|
||
"label": label,
|
||
"line_start": start_line,
|
||
"line_end": end_line,
|
||
"content": snippet
|
||
})
|
||
|
||
relative_path = self._relative_path(full_path)
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"size": result["size"],
|
||
"total_lines": total_lines,
|
||
"segments": extracted
|
||
}
|