JOJO
a2cf547400
- 新增 modules/i18n.py：tr() + 进程级语言缓存 + modules/i18n_messages/ 域文案包自动聚合 - 新增 29 个域文案包，共 1153 条双语 key；90+ 源文件 1146 处用户可见消息 tr 化 - ui_locale 存入 personalization.json（用户级共享），前后端双向同步 - 前端匹配点双语兼容（history/shared/ChatArea/taskPolling/upload 等正则） - 修复语言判等陷阱：审批等待加稳定 code 字段；conversation.py 不存在判等改双语 helper - 边界：日志/prompt 注入/子智能体工具回填/容器内嵌脚本不迁移
244 lines
9.5 KiB
Python
244 lines
9.5 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
|
||
from modules.i18n import tr
|
||
|
||
if TYPE_CHECKING:
|
||
from modules.user_container_manager import ContainerHandle
|
||
|
||
# 临时禁用长度检查
|
||
DISABLE_LENGTH_CHECK = True
|
||
|
||
logger = setup_logger(__name__)
|
||
|
||
class ListMixin:
|
||
"""FileManager list mixin 能力 mixin。"""
|
||
|
||
def edit_lines_range(self, path: str, start_line: int, end_line: int, content: str, operation: str) -> Dict:
|
||
"""
|
||
基于行号编辑文件
|
||
|
||
Args:
|
||
path: 文件路径
|
||
start_line: 起始行号（从1开始）
|
||
end_line: 结束行号（从1开始，包含）
|
||
content: 新内容
|
||
operation: 操作类型 - "replace", "insert", "delete"
|
||
"""
|
||
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": tr("file_manager.file_not_found")}
|
||
|
||
if not full_path.is_file():
|
||
return {"success": False, "error": tr("file_manager.not_a_file")}
|
||
|
||
# 验证行号
|
||
if start_line < 1:
|
||
return {"success": False, "error": tr("file_manager.line_start_must_be_one")}
|
||
|
||
if end_line < start_line:
|
||
return {"success": False, "error": tr("file_manager.end_line_lt_start_line")}
|
||
|
||
try:
|
||
relative_path = self._relative_path(full_path)
|
||
if self._use_container():
|
||
return self._container_call("edit_lines_range", {
|
||
"path": relative_path,
|
||
"start_line": start_line,
|
||
"end_line": end_line,
|
||
"content": content,
|
||
"operation": operation
|
||
})
|
||
|
||
# 读取文件内容
|
||
with open(full_path, 'r', encoding='utf-8') as f:
|
||
lines = f.readlines()
|
||
|
||
total_lines = len(lines)
|
||
|
||
# 检查行号范围
|
||
if start_line > total_lines:
|
||
if operation == "insert":
|
||
# 插入操作允许在文件末尾后插入
|
||
lines.extend([''] * (start_line - total_lines - 1))
|
||
lines.append(content if content.endswith('\n') else content + '\n')
|
||
else:
|
||
return {"success": False, "error": tr("file_manager.edit_start_line_out_of_range", start_line=start_line, total_lines=total_lines)}
|
||
elif end_line > total_lines:
|
||
return {"success": False, "error": tr("file_manager.edit_end_line_out_of_range", end_line=end_line, total_lines=total_lines)}
|
||
else:
|
||
# 执行操作（转换为0基索引）
|
||
start_idx = start_line - 1
|
||
end_idx = end_line
|
||
|
||
if operation == "replace":
|
||
# 替换指定行范围
|
||
new_lines = content.split('\n') if '\n' in content else [content]
|
||
# 确保每行都有换行符，除了最后一行需要检查原文件格式
|
||
formatted_lines = []
|
||
for i, line in enumerate(new_lines):
|
||
if i < len(new_lines) - 1 or (end_idx < len(lines) and lines[end_idx - 1].endswith('\n')):
|
||
formatted_lines.append(line + '\n' if not line.endswith('\n') else line)
|
||
else:
|
||
formatted_lines.append(line)
|
||
|
||
lines[start_idx:end_idx] = formatted_lines
|
||
affected_lines = end_line - start_line + 1
|
||
|
||
elif operation == "insert":
|
||
# 在指定行前插入内容
|
||
new_lines = content.split('\n') if '\n' in content else [content]
|
||
formatted_lines = [line + '\n' if not line.endswith('\n') else line for line in new_lines]
|
||
lines[start_idx:start_idx] = formatted_lines
|
||
affected_lines = len(formatted_lines)
|
||
|
||
elif operation == "delete":
|
||
# 删除指定行范围
|
||
affected_lines = end_line - start_line + 1
|
||
del lines[start_idx:end_idx]
|
||
|
||
else:
|
||
return {"success": False, "error": tr("file_manager.unknown_operation", operation=operation)}
|
||
|
||
# 写回文件
|
||
with open(full_path, 'w', encoding='utf-8') as f:
|
||
f.writelines(lines)
|
||
|
||
relative_path = self._relative_path(full_path)
|
||
|
||
# 生成操作描述
|
||
if operation == "replace":
|
||
operation_desc = tr("file_manager.desc_replace_lines", start=start_line, end=end_line)
|
||
elif operation == "insert":
|
||
operation_desc = tr("file_manager.desc_insert_before", line=start_line)
|
||
elif operation == "delete":
|
||
operation_desc = tr("file_manager.desc_delete_lines", start=start_line, end=end_line)
|
||
|
||
print(f"{OUTPUT_FORMATS['file']} {operation_desc}: {relative_path}")
|
||
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"operation": operation,
|
||
"start_line": start_line,
|
||
"end_line": end_line,
|
||
"affected_lines": affected_lines,
|
||
"total_lines_after": len(lines),
|
||
"description": operation_desc
|
||
}
|
||
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def list_files(self, path: str = "") -> Dict:
|
||
"""列出目录内容"""
|
||
if path:
|
||
valid, error, full_path = self._validate_path(path)
|
||
if not valid:
|
||
return {"success": False, "error": error}
|
||
else:
|
||
full_path = self.project_path
|
||
|
||
if not full_path.exists():
|
||
return {"success": False, "error": tr("file_manager.dir_not_found")}
|
||
|
||
if not full_path.is_dir():
|
||
return {"success": False, "error": tr("file_manager.not_a_directory")}
|
||
|
||
try:
|
||
files = []
|
||
folders = []
|
||
|
||
for item in full_path.iterdir():
|
||
if item.name.startswith('.'):
|
||
continue
|
||
|
||
relative_path = self._relative_path(item)
|
||
|
||
if item.is_file():
|
||
files.append({
|
||
"name": item.name,
|
||
"path": relative_path,
|
||
"size": item.stat().st_size,
|
||
"modified": datetime.fromtimestamp(item.stat().st_mtime).isoformat()
|
||
})
|
||
elif item.is_dir():
|
||
folders.append({
|
||
"name": item.name,
|
||
"path": relative_path
|
||
})
|
||
|
||
return {
|
||
"success": True,
|
||
"path": self._relative_path(full_path) if path else ".",
|
||
"files": sorted(files, key=lambda x: x["name"]),
|
||
"folders": sorted(folders, key=lambda x: x["name"])
|
||
}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|
||
|
||
def get_file_info(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": tr("file_manager.file_not_found")}
|
||
|
||
try:
|
||
stat = full_path.stat()
|
||
relative_path = self._relative_path(full_path)
|
||
|
||
return {
|
||
"success": True,
|
||
"path": relative_path,
|
||
"name": full_path.name,
|
||
"type": "file" if full_path.is_file() else "folder",
|
||
"size": stat.st_size,
|
||
"created": datetime.fromtimestamp(stat.st_ctime).isoformat(),
|
||
"modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
|
||
"extension": full_path.suffix if full_path.is_file() else None
|
||
}
|
||
except Exception as e:
|
||
return {"success": False, "error": str(e)}
|