命令运行器¶

cp runner/runner.py your_project/

from runner import run, run_async, stream, stream_async

from runner import run

# 列表形式（推荐）
result = run(["echo", "hello", "world"])
print(result.stdout)      # "hello world\n"
print(result.returncode)  # 0
print(result.duration)    # 0.003

# 字符串形式（通过 shlex 自动拆分）
result = run("echo hello world")

from runner import run

result = run(
    ["python3", "-c", "import sys; print(sys.stdin.read().upper())"],
    input="hello world",
)
print(result.stdout)  # "HELLO WORLD\n"

from runner import run, CommandFailedError, CommandTimeoutError

# check=True（默认） -- 非零退出码时抛出异常
try:
    run(["false"])
except CommandFailedError as e:
    print(e.result.returncode)  # 1
    print(e.result.stderr)

# 抑制异常
result = run(["false"], check=False)
print(result.returncode)  # 1

from runner import run, CommandTimeoutError

try:
    run(["sleep", "60"], timeout=5.0, kill_delay=2.0)
except CommandTimeoutError as e:
    print(f"超时：{e.timeout} 秒")
    print(e.partial_stdout)  # 超时前的输出

from runner import run

# 向继承的环境中添加变量
result = run(["env"], env_extra={"MY_VAR": "hello"})

# 移除敏感变量
result = run(["env"], env_remove=["SECRET_KEY", "AWS_ACCESS_KEY_ID"])

# 完全替换（干净环境）
result = run(["env"], env={"PATH": "/usr/bin", "HOME": "/tmp"})

# 三者可组合：从 env 开始，移除后再添加
result = run(
    ["env"],
    env={"PATH": "/usr/bin"},
    env_remove=["OLDVAR"],
    env_extra={"NEWVAR": "value"},
)

from runner import run

# 流式读取 stdout，同时仍捕获完整输出
result = run(
    ["make", "build"],
    on_stdout=lambda line: print(f"[build] {line}", end=""),
    on_stderr=lambda line: print(f"[ERR]   {line}", end=""),
)
# result.stdout 包含完整的捕获输出

from runner import stream

# 逐行迭代 stdout
with stream(["make", "build"]) as proc:
    for line in proc.iter_lines():
        print(f"[build] {line}", end="")
print(f"退出码: {proc.returncode}")

# 交错读取 stdout 和 stderr
with stream(["make", "build"]) as proc:
    for source, line in proc.iter_any():
        prefix = "[OUT]" if source == "stdout" else "[ERR]"
        print(f"{prefix} {line}", end="")

import asyncio
from runner import run_async, stream_async

async def main():
    # 基本异步执行
    result = await run_async("ls -la")
    print(result.stdout)

    # 异步流式输出
    async with stream_async(["make", "build"]) as proc:
        async for line in proc.aiter_lines():
            print(line, end="")

asyncio.run(main())

from runner import run, CommandBlockedError

# 仅允许特定命令
try:
    run(["rm", "-rf", "/"], allowed_commands=["echo", "cat", "ls"])
except CommandBlockedError as e:
    print(f"{e.command} 被阻止: {e.reason}")

# 阻止危险命令
run(["echo", "safe"], blocked_commands=["rm", "dd", "mkfs"])

from runner import shell_split, shell_quote, which

# 解析 shell 字符串为参数列表
args = shell_split('git commit -m "hello world"')
# ['git', 'commit', '-m', 'hello world']

# 引号转义以安全插入 shell
quoted = shell_quote("hello world", "it's fine")
# "'hello world' \"it's fine\""

# 在 PATH 中定位命令
path = which("git")
# '/usr/bin/git'