99 lines
3.4 KiB
Python
99 lines
3.4 KiB
Python
import subprocess
|
||
import time
|
||
import os
|
||
from pathlib import Path
|
||
from typing import Dict, Any
|
||
|
||
import yaml
|
||
|
||
from .models import RunRequest, RunResult
|
||
|
||
|
||
def load_config(config_path: str | Path = None) -> Dict[str, Any]:
|
||
"""加载配置，默认使用 AGENT_CONFIG 或 ./config.yaml。"""
|
||
if config_path is None:
|
||
env_path = os.environ.get("AGENT_CONFIG")
|
||
config_path = Path(env_path) if env_path else Path(__file__).resolve().parent.parent / "config.yaml"
|
||
config_path = Path(config_path)
|
||
with open(config_path, "r", encoding="utf-8") as f:
|
||
return yaml.safe_load(f) or {}
|
||
|
||
|
||
def resolve_tool(config: Dict[str, Any], tool_name: str) -> Path:
|
||
tools = config.get("tools", {})
|
||
if tool_name not in tools:
|
||
raise ValueError(f"未知工具: {tool_name}")
|
||
tool_path = Path(tools[tool_name]["path"])
|
||
if not tool_path.exists():
|
||
raise FileNotFoundError(f"工具不存在: {tool_path}")
|
||
return tool_path
|
||
|
||
|
||
def run_tool(config: Dict[str, Any], req: RunRequest) -> RunResult:
|
||
start = time.time()
|
||
defaults = config.get("run_defaults", {})
|
||
|
||
tool_path = resolve_tool(config, req.tool)
|
||
workdir = Path(req.workdir or config.get("base_workdir", ".")).resolve()
|
||
workdir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# 1. 先构建参数列表 (List)
|
||
args_list = [subprocess.list2cmdline([str(tool_path)])] + req.args
|
||
|
||
# 2. 【关键修改】转换为安全的命令行字符串
|
||
cmd_str = " ".join(args_list)
|
||
|
||
# 处理环境
|
||
env = os.environ.copy()
|
||
if config.get("env"):
|
||
env.update({k: str(v) for k, v in config["env"].items()})
|
||
if req.env:
|
||
env.update({k: str(v) for k, v in req.env.items()})
|
||
|
||
timeout = req.timeout
|
||
capture_output = req.capture_output if req.capture_output is not None else defaults.get("capture_output", True)
|
||
strip_output = req.strip_output if req.strip_output is not None else defaults.get("strip_output", True)
|
||
|
||
print(f"Running tool in {workdir} :: timeout={timeout}s command:")
|
||
#print(f"Command: {cmd_str}") # 现在的打印结果是完全真实的执行指令
|
||
print(cmd_str)
|
||
try:
|
||
# 3. 执行：传入字符串 cmd_str
|
||
completed = subprocess.run(
|
||
cmd_str, # <--- 这里现在是字符串
|
||
cwd=str(workdir),
|
||
env=env,
|
||
timeout=timeout,
|
||
capture_output=capture_output,
|
||
text=True,
|
||
check=False,
|
||
shell=False # Windows 上执行 exe 字符串通常不需要 shell=True，除非用到管道/重定向
|
||
)
|
||
stdout = completed.stdout if capture_output else None
|
||
stderr = completed.stderr if capture_output else None
|
||
if strip_output:
|
||
stdout = stdout.strip() if stdout is not None else None
|
||
stderr = stderr.strip() if stderr is not None else None
|
||
|
||
elapsed = time.time() - start
|
||
return RunResult(
|
||
success=completed.returncode == 0,
|
||
return_code=completed.returncode,
|
||
stdout=stdout,
|
||
stderr=stderr,
|
||
elapsed=elapsed,
|
||
tool_path=str(tool_path),
|
||
workdir=str(workdir),
|
||
)
|
||
except Exception as exc:
|
||
elapsed = time.time() - start
|
||
return RunResult(
|
||
success=False,
|
||
return_code=None,
|
||
stdout=None,
|
||
stderr=str(exc),
|
||
elapsed=elapsed,
|
||
tool_path=str(tool_path),
|
||
workdir=str(workdir),
|
||
)
|