read.py 6.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227
  1. """
  2. Read Tool - 文件读取工具
  3. 参考 OpenCode read.ts 完整实现。
  4. 核心功能:
  5. - 支持文本文件、图片、PDF
  6. - 分页读取(offset/limit)
  7. - 二进制文件检测
  8. - 行长度和字节限制
  9. """
  10. import os
  11. import mimetypes
  12. from pathlib import Path
  13. from typing import Optional
  14. from agent.tools import tool, ToolResult, ToolContext
  15. # 常量(参考 opencode)
  16. DEFAULT_READ_LIMIT = 2000
  17. MAX_LINE_LENGTH = 2000
  18. MAX_BYTES = 50 * 1024 # 50KB
  19. @tool(description="读取文件内容,支持文本文件、图片、PDF 等多种格式")
  20. async def read_file(
  21. file_path: str,
  22. offset: int = 0,
  23. limit: int = DEFAULT_READ_LIMIT,
  24. context: Optional[ToolContext] = None
  25. ) -> ToolResult:
  26. """
  27. 读取文件内容
  28. 参考 OpenCode 实现
  29. Args:
  30. file_path: 文件路径(绝对路径或相对路径)
  31. offset: 起始行号(从 0 开始)
  32. limit: 读取行数(默认 2000 行)
  33. context: 工具上下文
  34. Returns:
  35. ToolResult: 文件内容
  36. """
  37. # 解析路径
  38. path = Path(file_path)
  39. if not path.is_absolute():
  40. path = Path.cwd() / path
  41. # 检查文件是否存在
  42. if not path.exists():
  43. # 尝试提供建议(参考 opencode:44-60)
  44. parent_dir = path.parent
  45. if parent_dir.exists():
  46. candidates = [
  47. f for f in parent_dir.iterdir()
  48. if path.name.lower() in f.name.lower() or f.name.lower() in path.name.lower()
  49. ][:3]
  50. if candidates:
  51. suggestions = "\n".join(str(c) for c in candidates)
  52. return ToolResult(
  53. title=f"文件未找到: {path.name}",
  54. output=f"文件不存在: {file_path}\n\n你是否想要:\n{suggestions}",
  55. error="File not found"
  56. )
  57. return ToolResult(
  58. title="文件未找到",
  59. output=f"文件不存在: {file_path}",
  60. error="File not found"
  61. )
  62. # 检测文件类型
  63. mime_type, _ = mimetypes.guess_type(str(path))
  64. mime_type = mime_type or ""
  65. # 图片文件(参考 opencode:66-91)
  66. if mime_type.startswith("image/") and mime_type not in ["image/svg+xml", "image/vnd.fastbidsheet"]:
  67. # 注意:实际项目中需要实现图片的 base64 编码
  68. # 这里简化处理
  69. return ToolResult(
  70. title=path.name,
  71. output=f"图片文件: {path.name} (MIME: {mime_type})",
  72. metadata={"mime_type": mime_type, "truncated": False}
  73. )
  74. # PDF 文件
  75. if mime_type == "application/pdf":
  76. return ToolResult(
  77. title=path.name,
  78. output=f"PDF 文件: {path.name}",
  79. metadata={"mime_type": mime_type, "truncated": False}
  80. )
  81. # 二进制文件检测(参考 opencode:156-211)
  82. if _is_binary_file(path):
  83. return ToolResult(
  84. title="二进制文件",
  85. output=f"无法读取二进制文件: {path.name}",
  86. error="Binary file"
  87. )
  88. # 读取文本文件(参考 opencode:96-143)
  89. try:
  90. with open(path, 'r', encoding='utf-8') as f:
  91. lines = f.readlines()
  92. total_lines = len(lines)
  93. end_line = min(offset + limit, total_lines)
  94. # 截取行并处理长度限制
  95. output_lines = []
  96. total_bytes = 0
  97. truncated_by_bytes = False
  98. for i in range(offset, end_line):
  99. line = lines[i].rstrip('\n\r')
  100. # 行长度限制(参考 opencode:104)
  101. if len(line) > MAX_LINE_LENGTH:
  102. line = line[:MAX_LINE_LENGTH] + "..."
  103. # 字节限制(参考 opencode:105-112)
  104. line_bytes = len(line.encode('utf-8')) + (1 if output_lines else 0)
  105. if total_bytes + line_bytes > MAX_BYTES:
  106. truncated_by_bytes = True
  107. break
  108. output_lines.append(line)
  109. total_bytes += line_bytes
  110. # 格式化输出(参考 opencode:114-134)
  111. formatted = []
  112. for idx, line in enumerate(output_lines):
  113. line_num = offset + idx + 1
  114. formatted.append(f"{line_num:5d}| {line}")
  115. output = "<file>\n" + "\n".join(formatted)
  116. last_read_line = offset + len(output_lines)
  117. has_more = total_lines > last_read_line
  118. truncated = has_more or truncated_by_bytes
  119. # 添加提示
  120. if truncated_by_bytes:
  121. output += f"\n\n(输出在 {MAX_BYTES} 字节处被截断。使用 'offset' 参数读取第 {last_read_line} 行之后的内容)"
  122. elif has_more:
  123. output += f"\n\n(文件还有更多内容。使用 'offset' 参数读取第 {last_read_line} 行之后的内容)"
  124. else:
  125. output += f"\n\n(文件结束 - 共 {total_lines} 行)"
  126. output += "\n</file>"
  127. # 预览(前 20 行)
  128. preview = "\n".join(output_lines[:20])
  129. return ToolResult(
  130. title=path.name,
  131. output=output,
  132. metadata={
  133. "preview": preview,
  134. "truncated": truncated,
  135. "total_lines": total_lines,
  136. "read_lines": len(output_lines)
  137. }
  138. )
  139. except UnicodeDecodeError:
  140. return ToolResult(
  141. title="编码错误",
  142. output=f"无法解码文件(非 UTF-8 编码): {path.name}",
  143. error="Encoding error"
  144. )
  145. except Exception as e:
  146. return ToolResult(
  147. title="读取错误",
  148. output=f"读取文件时出错: {str(e)}",
  149. error=str(e)
  150. )
  151. def _is_binary_file(path: Path) -> bool:
  152. """
  153. 检测是否为二进制文件
  154. 参考 OpenCode 实现
  155. """
  156. # 常见二进制扩展名
  157. binary_exts = {
  158. '.zip', '.tar', '.gz', '.exe', '.dll', '.so', '.class',
  159. '.jar', '.war', '.7z', '.doc', '.docx', '.xls', '.xlsx',
  160. '.ppt', '.pptx', '.odt', '.ods', '.odp', '.bin', '.dat',
  161. '.obj', '.o', '.a', '.lib', '.wasm', '.pyc', '.pyo'
  162. }
  163. if path.suffix.lower() in binary_exts:
  164. return True
  165. # 检查文件内容
  166. try:
  167. file_size = path.stat().st_size
  168. if file_size == 0:
  169. return False
  170. # 读取前 4KB
  171. buffer_size = min(4096, file_size)
  172. with open(path, 'rb') as f:
  173. buffer = f.read(buffer_size)
  174. # 检测 null 字节
  175. if b'\x00' in buffer:
  176. return True
  177. # 统计非打印字符(参考 opencode:202-210)
  178. non_printable = 0
  179. for byte in buffer:
  180. if byte < 9 or (13 < byte < 32):
  181. non_printable += 1
  182. # 如果超过 30% 是非打印字符,认为是二进制
  183. return non_printable / len(buffer) > 0.3
  184. except Exception:
  185. return False