read.py 6.8 KB

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