Coverage for merco/tools/file_tools.py: 32%

1"""文件操作工具 — 流式行读，支持 head/tail/翻页""" 

2 

3import logging 

4from collections import deque 

5from pathlib import Path 

6from .base import BaseTool 

7 

8logger = logging.getLogger("merco.tools.file") 

9 

10_DEFAULT_LIMIT = 500 # 默认读取上限（行），覆盖大多数源文件，大文件用 offset 翻页 

11 

12 

13class ReadFile(BaseTool): 

14 """流式行读取 — 大文件不爆内存，默认上限防上下文溢出""" 

15 

16 name = "read_file" 

17 description = ( 

18 "读取文件内容，默认返回前 500 行。大文件请用 offset/limit 翻页。\n" 

19 "- offset: 起始行号（1-indexed），默认第 1 行\n" 

20 "- limit: 读取行数，默认 500，设 0 表示不设上限（慎用）\n" 

21 "- head: 读前 N 行，等价 offset=1, limit=N\n" 

22 "- tail: 读最后 N 行" 

23 ) 

24 toolset = "file" 

25 parameters = { 

26 "type": "object", 

27 "properties": { 

28 "path": { 

29 "type": "string", 

30 "description": "文件路径", 

31 }, 

32 "offset": { 

33 "type": "integer", 

34 "description": "起始行号（1-indexed），默认 1", 

35 }, 

36 "limit": { 

37 "type": "integer", 

38 "description": "读取行数，默认 500。设 0 取消上限（仅小文件）", 

39 }, 

40 "head": { 

41 "type": "integer", 

42 "description": "读前 N 行，等价 offset=1, limit=N", 

43 }, 

44 "tail": { 

45 "type": "integer", 

46 "description": "读最后 N 行", 

47 }, 

48 }, 

49 "required": ["path"], 

50 } 

51 

52 async def execute(self, path: str, limit: int = None, offset: int = None, 

53 head: int = None, tail: int = None) -> dict: 

54 file_path = Path(path) 

55 if not file_path.exists(): 

56 return {"error": f"文件 '{path}' 不存在"} 

57 if not file_path.is_file(): 

58 return {"error": f"'{path}' 不是文件"} 

59 

60 # ── 参数解析 ── 

61 if tail is not None: 

62 return self._read_tail(file_path, tail) 

63 

64 if head is not None: 

65 offset = 1 

66 limit = head 

67 

68 start_line = offset or 1 

69 if start_line < 1: 

70 start_line = 1 

71 

72 if limit is None: 

73 limit = _DEFAULT_LIMIT 

74 elif limit == 0: 

75 limit = None # 0 = 不设上限 

76 elif limit < 0: 

77 limit = _DEFAULT_LIMIT 

78 

79 return self._read_by_lines(file_path, start_line, limit) 

80 

81 # ── 流式行读（不一次性加载全文件） ── 

82 

83 def _read_by_lines(self, file_path: Path, start_line: int, limit: int | None) -> dict: 

84 """从 start_line 开始逐行读取，读到 limit 行或 EOF 即停""" 

85 try: 

86 with open(file_path, "r", encoding="utf-8", errors="replace") as f: 

87 # 跳过 start_line - 1 行 

88 for _ in range(start_line - 1): 

89 if not f.readline(): 

90 return { 

91 "content": "", 

92 "hint": f"offset {start_line} 超出文件范围", 

93 } 

94 

95 lines: list[str] = [] 

96 for line in f: 

97 lines.append(line) 

98 if limit is not None and len(lines) >= limit: 

99 break 

100 

101 # 探测是否还有更多内容 

102 has_more = bool(f.readline()) if limit is not None else False 

103 except OSError as e: 

104 return {"error": f"读取失败: {e}"} 

105 

106 content = "".join(lines) 

107 end_line = start_line + len(lines) - 1 

108 mtime = file_path.stat().st_mtime 

109 

110 hint = "" if not has_more else ( 

111 f"已返回 {start_line}-{end_line} 行，文件未完。" 

112 f"用 offset={end_line + 1} 继续翻页。" 

113 ) 

114 

115 return { 

116 "content": content, 

117 "start_line": start_line, 

118 "end_line": end_line, 

119 "has_more": has_more, 

120 "hint": hint, 

121 "mtime": mtime, 

122 } 

123 

124 # ── 读尾部 N 行 ── 

125 

126 def _read_tail(self, file_path: Path, n: int) -> dict: 

127 """读取文件最后 N 行 — 用固定大小的双端队列，内存 O(n)""" 

128 try: 

129 with open(file_path, "r", encoding="utf-8", errors="replace") as f: 

130 last_n = deque(f, maxlen=n) 

131 except OSError as e: 

132 return {"error": f"读取失败: {e}"} 

133 

134 content = "".join(last_n) 

135 mtime = file_path.stat().st_mtime 

136 

137 return { 

138 "content": content, 

139 "lines": len(last_n), 

140 "hint": f"文件最后 {len(last_n)} 行", 

141 "mtime": mtime, 

142 } 

143 

144 

145class WriteFile(BaseTool): 

146 """写入文件内容""" 

147 

148 name = "write_file" 

149 description = ( 

150 "创建新文件或完全覆盖已有文件。\n" 

151 "⚠️ 仅用于新建文件。修改已有文件请用 edit_file（SEARCH/REPLACE），" 

152 "它会展示 diff 并等待确认。" 

153 ) 

154 toolset = "file" 

155 parameters = { 

156 "type": "object", 

157 "properties": { 

158 "path": {"type": "string", "description": "文件路径"}, 

159 "content": {"type": "string", "description": "写入内容"}, 

160 }, 

161 "required": ["path", "content"], 

162 } 

163 

164 async def execute(self, path: str, content: str) -> dict: 

165 file_path = Path(path) 

166 file_path.parent.mkdir(parents=True, exist_ok=True) 

167 file_path.write_text(content) 

168 return {"success": True, "path": str(path)} 

169 

170 

171from .registry import tool_registry # noqa: E402 — 模块末尾自注册 

172tool_registry.register(ReadFile()) 

173tool_registry.register(WriteFile())