Coverage for tools / filesystem.py: 29%

1from pathlib import Path 

2from pydantic import BaseModel, Field 

3from qrclaw.tools.registry import register 

4from qrclaw.logger import get_logger 

5 

6logger = get_logger("qrclaw.tools.filesystem") 

7 

8class ReadFileArgs(BaseModel): 

9 path: str = Field(description="文件的绝对路径，例如 /tmp/test.txt 或 C:\\Users\\test.txt") 

10 

11class WriteFileArgs(BaseModel): 

12 path: str = Field(description="文件的绝对路径，文件不存在会自动创建") 

13 content: str = Field(description="要写入的文本内容") 

14 

15class ListDirectoryArgs(BaseModel): 

16 path: str = Field(description="要列出的目录路径，例如 /tmp 或 ~/Documents") 

17 

18@register(description="读取本地文件的内容", args_model=ReadFileArgs) 

19def read_file(path: str) -> str: 

20 logger.debug(f"读取文件: {path}") 

21 try: 

22 p = Path(path).expanduser().resolve() 

23 content = p.read_text(encoding="utf-8") 

24 logger.info(f"读取文件成功: {p}, 大小: {len(content)} 字符") 

25 return content 

26 except FileNotFoundError: 

27 error_msg = f"错误：文件不存在 {path}" 

28 logger.warning(error_msg) 

29 return error_msg 

30 except Exception as e: 

31 error_msg = f"错误：{e}" 

32 logger.error(f"读取文件失败: {path}, 错误: {e}", exc_info=True) 

33 return error_msg 

34 

35@register(description="把内容写入本地文件，文件不存在会自动创建，已存在则覆盖", args_model=WriteFileArgs, confirm=True) 

36def write_file(path: str, content: str) -> str: 

37 logger.debug(f"写入文件: {path}, 内容长度: {len(content)}") 

38 try: 

39 p = Path(path).expanduser().resolve() 

40 p.parent.mkdir(parents=True, exist_ok=True) 

41 p.write_text(content, encoding="utf-8") 

42 logger.info(f"写入文件成功: {p}") 

43 return f"已写入：{p}" 

44 except Exception as e: 

45 error_msg = f"错误：写入失败 {e}" 

46 logger.error(f"写入文件失败: {path}, 错误: {e}", exc_info=True) 

47 return error_msg 

48 

49@register(description="列出目录下的文件和子目录，LLM用此工具了解目录结构", args_model=ListDirectoryArgs) 

50def list_directory(path: str) -> str: 

51 logger.debug(f"列出目录: {path}") 

52 try: 

53 p = Path(path).expanduser().resolve() 

54 if not p.is_dir(): 

55 error_msg = f"错误：{path} 不是一个目录" 

56 logger.warning(error_msg) 

57 return error_msg 

58 entries = sorted(p.iterdir(), key=lambda e: (e.is_file(), e.name)) 

59 lines = [] 

60 for entry in entries: 

61 tag = "[文件]" if entry.is_file() else "[目录]" 

62 lines.append(f"{tag} {entry.name}") 

63 result = "\n".join(lines) if lines else "(空目录)" 

64 logger.info(f"列出目录成功: {p}, 共 {len(entries)} 项") 

65 return result 

66 except Exception as e: 

67 error_msg = f"错误：{e}" 

68 logger.error(f"列出目录失败: {path}, 错误: {e}", exc_info=True) 

69 return error_msg