Coverage for tools / shell.py: 23%

1import os 

2import subprocess 

3from pydantic import BaseModel, Field 

4from qrclaw.tools.registry import register 

5from qrclaw.logger import get_logger 

6 

7logger = get_logger("qrclaw.tools.shell") 

8 

9 

10class RunShellArgs(BaseModel): 

11 command: str = Field(description="要执行的 shell 命令，例如 ls -la 或 python3 hello.py") 

12 

13 

14@register(description="在本地执行 shell 命令，返回输出结果，超时 30 分钟", args_model=RunShellArgs, confirm=True) 

15def run_shell(command: str) -> str: 

16 logger.debug(f"执行 shell 命令: {command}") 

17 

18 # 获取当前 workspace，强制在 workspace 目录下执行 

19 cwd = None 

20 try: 

21 from qrclaw.agent import get_workspace 

22 ws = get_workspace() 

23 if ws: 

24 cwd = str(ws.root) 

25 logger.debug(f"Shell 命令将在 workspace 目录下执行: {cwd}") 

26 except Exception: 

27 pass # 如果获取不到 workspace，使用当前目录 

28 

29 try: 

30 result = subprocess.run( 

31 command, 

32 shell=True, 

33 capture_output=True, 

34 timeout=1800, # 30 分钟 = 1800 秒 

35 cwd=cwd, # 强制工作目录 

36 ) 

37 encoding = "gbk" if os.name == "nt" else "utf-8" 

38 stdout = result.stdout.decode(encoding, errors="replace").strip() 

39 stderr = result.stderr.decode(encoding, errors="replace").strip() 

40 output = stdout 

41 if stderr: 

42 output += f"\n[stderr] {stderr}" 

43 

44 if result.returncode == 0: 

45 logger.info(f"命令执行成功: {command}") 

46 else: 

47 logger.warning(f"命令执行失败 (exit code {result.returncode}): {command}") 

48 

49 # 如果被强制在 workspace 执行，提示用户 

50 if cwd and result.returncode == 0: 

51 # 检查命令是否尝试访问外部路径 

52 if "/" in command and not command.startswith("/"): 

53 # 命令中有路径但不是绝对路径，可能在当前目录 

54 pass 

55 

56 return output or "(无输出)" 

57 except subprocess.TimeoutExpired: 

58 error_msg = "错误：命令执行超时（30分钟）" 

59 logger.warning(f"命令超时: {command}") 

60 return error_msg 

61 except Exception as e: 

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

63 logger.error(f"命令执行失败: {command}, 错误: {e}", exc_info=True) 

64 return error_msg