Python subprocess：外部命令執行與管道串接完全指南

import os

# 古老做法 ❌
os.system("echo hello world")
# 回傳值是 exit code，不是輸出內容
result = os.system("ls -la")
print(result)  # 0（成功）或其他數字（失敗）
# 但你拿不到 ls 的輸出！

os.system 的問題：

拿不到輸出 — 它只回傳 exit code
安全性差 — 直接把字串丟給 shell，容易被注入攻擊
無法控制 — 不能設超時、不能傳 stdin、不能分離 stdout/stderr

# os.popen 稍微好一點但也過時了 ❌
import os
output = os.popen("ls -la").read()
# 沒有錯誤處理、沒有 stderr、不安全

正確答案：subprocess 模組。它從 Python 3.5 開始提供了 run() 函式，涵蓋了 95% 的使用場景。

三、subprocess.run 基礎
#

最簡單的用法
#

import subprocess

# 執行命令，等它跑完
result = subprocess.run(["echo", "hello", "world"])
print(result.returncode)  # 0

注意：命令用 list 傳入，不是字串！這是最重要的觀念。

# ✅ 正確：用 list
subprocess.run(["ls", "-la", "/tmp"])

# ❌ 危險：用字串 + shell=True
subprocess.run("ls -la /tmp", shell=True)

為什麼用 list？因為每個元素是獨立的參數，不會被 shell 解釋。這避免了 shell injection 攻擊：

# 假設 filename 來自使用者輸入
filename = "important.txt; rm -rf /"

# ❌ shell=True：filename 會被 shell 解釋！
subprocess.run(f"cat {filename}", shell=True)
# 實際執行：cat important.txt; rm -rf / 😱

# ✅ list 形式：filename 只是一個參數
subprocess.run(["cat", filename])
# 實際執行：cat "important.txt; rm -rf /"
# 找不到這個檔案，安全地報錯

捕獲輸出
#

import subprocess

# capture_output=True 捕獲 stdout 和 stderr
result = subprocess.run(
    ["python3", "--version"],
    capture_output=True,
    text=True,  # 用字串而非 bytes
)

print(result.stdout)       # "Python 3.12.x\n"
print(result.stderr)       # ""（沒有錯誤）
print(result.returncode)   # 0

text=True 是個關鍵參數——沒有它，stdout 和 stderr 會是 bytes 型別：

# 沒有 text=True
result = subprocess.run(["echo", "嗨"], capture_output=True)
print(result.stdout)        # b'\xe5\x97\xa8\n'
print(type(result.stdout))  # <class 'bytes'>

# 有 text=True
result = subprocess.run(["echo", "嗨"], capture_output=True, text=True)
print(result.stdout)        # "嗨\n"
print(type(result.stdout))  # <class 'str'>

檢查回傳值
#

import subprocess

# 方法 1：手動檢查 returncode
result = subprocess.run(["ls", "/不存在的路徑"], capture_output=True, text=True)
if result.returncode != 0:
    print(f"命令失敗：{result.stderr}")

# 方法 2：用 check=True 自動拋例外
try:
    result = subprocess.run(
        ["ls", "/不存在的路徑"],
        capture_output=True,
        text=True,
        check=True,  # 非零 exit code 會拋 CalledProcessError
    )
except subprocess.CalledProcessError as e:
    print(f"命令失敗（exit code {e.returncode}）")
    print(f"stderr: {e.stderr}")

拍拍君建議：大多數時候都加 check=True。明確的例外比默默失敗好太多了。

四、CompletedProcess 物件解析
#

subprocess.run() 回傳一個 CompletedProcess 物件，包含所有你需要的資訊：

import subprocess

result = subprocess.run(
    ["git", "log", "--oneline", "-5"],
    capture_output=True,
    text=True,
)

# CompletedProcess 的屬性
print(result.args)         # ['git', 'log', '--oneline', '-5']
print(result.returncode)   # 0
print(result.stdout)       # "abc1234 最新的 commit\n..."
print(result.stderr)       # ""

你也可以把它當作布林值使用（但要小心）：

# CompletedProcess 不直接支援 bool 轉換
# 正確做法：
if result.returncode == 0:
    print("成功！")

五、實用技巧大全
#

設定工作目錄
#

import subprocess

# 在指定目錄下執行命令
result = subprocess.run(
    ["git", "status", "--short"],
    capture_output=True,
    text=True,
    cwd="/path/to/your/repo",  # 指定工作目錄
)
print(result.stdout)

設定環境變數
#

import subprocess
import os

# 方法 1：完全自訂環境變數
result = subprocess.run(
    ["printenv", "MY_VAR"],
    capture_output=True,
    text=True,
    env={"MY_VAR": "hello", "PATH": os.environ["PATH"]},
)

# 方法 2：在現有環境基礎上新增（推薦）
my_env = os.environ.copy()
my_env["MY_VAR"] = "hello"
my_env["DEBUG"] = "1"

result = subprocess.run(
    ["python3", "my_script.py"],
    capture_output=True,
    text=True,
    env=my_env,
)

超時控制
#

import subprocess

try:
    result = subprocess.run(
        ["sleep", "100"],
        timeout=5,  # 最多等 5 秒
    )
except subprocess.TimeoutExpired:
    print("命令執行超時！已被終止。")

超時功能在自動化腳本裡超重要——防止某個命令卡住導致整個 pipeline 停擺。

傳入 stdin
#

import subprocess

# 把字串當作 stdin 傳給命令
result = subprocess.run(
    ["grep", "error"],
    input="line 1: ok\nline 2: error found\nline 3: ok\n",
    capture_output=True,
    text=True,
)
print(result.stdout)  # "line 2: error found\n"

# 實用範例：用 Python 生成資料，交給外部工具處理
import json
import subprocess

data = {"name": "拍拍", "type": "penguin"}
json_str = json.dumps(data, ensure_ascii=False)

result = subprocess.run(
    ["jq", ".name"],
    input=json_str,
    capture_output=True,
    text=True,
)
print(result.stdout.strip())  # "拍拍"

六、管道串接（Pipe）
#

Shell 裡最強大的概念之一就是管道：cmd1 | cmd2 | cmd3。在 Python 裡要怎麼做？

方法 1：用 input 串接（推薦）
#

import subprocess

# Shell 等價：cat /etc/passwd | grep root | wc -l

# 第一步：讀取檔案
step1 = subprocess.run(
    ["cat", "/etc/passwd"],
    capture_output=True,
    text=True,
    check=True,
)

# 第二步：過濾
step2 = subprocess.run(
    ["grep", "root"],
    input=step1.stdout,
    capture_output=True,
    text=True,
)

# 第三步：計數
step3 = subprocess.run(
    ["wc", "-l"],
    input=step2.stdout,
    capture_output=True,
    text=True,
)

print(step3.stdout.strip())  # "2"（或其他數字）

這種方式簡單明瞭，每一步的輸出都存在 Python 變數裡，方便 debug。

方法 2：用 Popen 建立真正的管道
#

如果資料量很大，不想把全部輸出存在記憶體裡，可以用 Popen：

import subprocess

# 建立管道：ls -la | grep ".py"
p1 = subprocess.Popen(
    ["ls", "-la"],
    stdout=subprocess.PIPE,
)

p2 = subprocess.Popen(
    ["grep", r"\.py"],
    stdin=p1.stdout,
    stdout=subprocess.PIPE,
    text=True,
)

# 重要：關閉 p1 的 stdout，讓 p1 能收到 SIGPIPE
p1.stdout.close()

output = p2.communicate()[0]
print(output)

方法 3：純 Python 實現（最安全）
#

很多時候，你不需要管道——用純 Python 就能做：

import subprocess

# 取代 shell 管道：git log --oneline | grep "fix" | head -5

result = subprocess.run(
    ["git", "log", "--oneline", "-50"],
    capture_output=True,
    text=True,
    cwd="/path/to/repo",
)

# 用 Python 過濾和處理
lines = result.stdout.strip().split("\n")
fix_commits = [line for line in lines if "fix" in line.lower()][:5]

for commit in fix_commits:
    print(commit)

拍拍君強烈推薦這種方式！能用 Python 做的事，就不要用 shell 管道。 更容易測試、debug、維護。

七、即時輸出（串流處理）
#

有時候你不想等命令跑完才看到輸出——比如跑一個很長的 build 命令：

即時列印輸出
#

import subprocess

# 不捕獲輸出，直接印到終端（預設行為）
subprocess.run(["pip", "install", "requests"])
# 安裝過程會即時顯示在終端

邊跑邊處理每一行
#

import subprocess

# 用 Popen 逐行讀取
process = subprocess.Popen(
    ["ping", "-c", "5", "google.com"],
    stdout=subprocess.PIPE,
    text=True,
)

for line in process.stdout:
    line = line.strip()
    if "time=" in line:
        # 提取延遲時間
        time_part = line.split("time=")[1].split()[0]
        print(f"延遲：{time_part}")

process.wait()
print(f"Exit code: {process.returncode}")

同時處理 stdout 和 stderr
#

import subprocess
import threading
import queue

def reader(pipe, q, label):
    """在另一個 thread 讀取 pipe 的輸出"""
    for line in pipe:
        q.put((label, line.strip()))
    pipe.close()

process = subprocess.Popen(
    ["python3", "-c", """
import sys, time
for i in range(5):
    print(f"stdout: step {i}")
    print(f"stderr: warning {i}", file=sys.stderr)
    time.sleep(0.5)
"""],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
)

q = queue.Queue()
t1 = threading.Thread(target=reader, args=(process.stdout, q, "OUT"))
t2 = threading.Thread(target=reader, args=(process.stderr, q, "ERR"))
t1.start()
t2.start()

t1.join()
t2.join()

while not q.empty():
    label, line = q.get()
    prefix = "📤" if label == "OUT" else "⚠️"
    print(f"{prefix} {line}")

process.wait()

八、進階：Popen 深入理解
#

subprocess.run() 其實是 Popen 的封裝。當你需要更精細的控制時，直接用 Popen：

Popen 作為 context manager
#

import subprocess

# 用 with 語句確保 process 被正確清理
with subprocess.Popen(
    ["python3", "-c", "import time; time.sleep(2); print('done')"],
    stdout=subprocess.PIPE,
    text=True,
) as proc:
    stdout, stderr = proc.communicate(timeout=5)
    print(stdout)  # "done\n"
# with 結束時自動清理

非同步互動
#

import subprocess

# 啟動一個互動式 process
proc = subprocess.Popen(
    ["python3", "-i"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
)

# 送指令給 Python 直譯器
stdout, stderr = proc.communicate(
    input="x = 42\nprint(x * 2)\n"
)
print(stdout)  # "84\n"

run vs Popen 選擇指南
#

場景	用什麼
跑一個命令，等它結束	`subprocess.run()`
需要即時讀取輸出	`Popen` + 逐行讀取
需要管道串接	`run` + `input` 或 `Popen`
需要跟 process 互動	`Popen` + `communicate()`
背景執行	`Popen`（不呼叫 `wait()`）

九、安全性最佳實踐
#

絕對不要做的事
#

import subprocess

# ❌ 永遠不要這樣做：把使用者輸入放進 shell=True
user_input = input("請輸入檔名：")
subprocess.run(f"cat {user_input}", shell=True)
# 使用者輸入 "file.txt; rm -rf /" → 完蛋

# ❌ 不要用 shell=True 來偷懶
subprocess.run("echo hello | grep hello", shell=True)

正確的做法
#

import subprocess
import shlex

# ✅ 用 list 傳參數
user_input = input("請輸入檔名：")
subprocess.run(["cat", user_input])  # 安全！

# ✅ 如果真的需要 shell 功能，用 shlex.quote
subprocess.run(f"cat {shlex.quote(user_input)}", shell=True)

# ✅ 但更好的做法是完全不用 shell
# 用 Python 的 pathlib 處理檔案路徑
from pathlib import Path
file_path = Path(user_input)
if file_path.exists() and file_path.is_file():
    content = file_path.read_text()

什麼時候可以用 shell=True？
#

只有在你完全控制命令內容的時候：

import subprocess

# ✅ 命令是硬編碼的，沒有使用者輸入
subprocess.run("echo $HOME", shell=True)

# ✅ 需要 shell 的特殊功能（glob、重導向等）
subprocess.run("ls *.py > filelist.txt", shell=True)
# 但更好的做法：
import glob
py_files = glob.glob("*.py")

十、實戰範例：自動化部署腳本
#

把今天學的全部串起來，寫一個完整的自動化部署腳本：

"""簡易自動化部署腳本"""
import subprocess
import sys
from pathlib import Path


def run_cmd(cmd: list[str], **kwargs) -> subprocess.CompletedProcess:
    """執行命令並印出過程"""
    print(f"🔧 Running: {' '.join(cmd)}")
    result = subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        **kwargs,
    )
    if result.returncode != 0:
        print(f"❌ Failed (exit {result.returncode})")
        if result.stderr:
            print(f"   stderr: {result.stderr.strip()}")
        sys.exit(1)
    return result


def deploy(repo_path: str, branch: str = "main"):
    """部署流程"""
    cwd = Path(repo_path)

    # 1. 確認在正確的 branch
    result = run_cmd(["git", "branch", "--show-current"], cwd=str(cwd))
    current_branch = result.stdout.strip()
    print(f"📌 Current branch: {current_branch}")

    if current_branch != branch:
        print(f"⚠️ 不在 {branch} branch，切換中...")
        run_cmd(["git", "checkout", branch], cwd=str(cwd))

    # 2. 拉最新的程式碼
    print("\n📥 Pulling latest changes...")
    run_cmd(["git", "pull", "origin", branch], cwd=str(cwd))

    # 3. 安裝依賴
    print("\n📦 Installing dependencies...")
    run_cmd(["pip", "install", "-r", "requirements.txt"], cwd=str(cwd))

    # 4. 跑測試
    print("\n🧪 Running tests...")
    try:
        result = subprocess.run(
            ["python3", "-m", "pytest", "-v", "--tb=short"],
            capture_output=True,
            text=True,
            cwd=str(cwd),
            timeout=300,  # 最多 5 分鐘
        )
        if result.returncode != 0:
            print("❌ Tests failed!")
            print(result.stdout)
            sys.exit(1)
        print("✅ All tests passed!")
    except subprocess.TimeoutExpired:
        print("⏰ Tests timed out!")
        sys.exit(1)

    # 5. 重啟服務
    print("\n🔄 Restarting service...")
    run_cmd(["systemctl", "restart", "myapp"])

    # 6. 確認服務狀態
    result = run_cmd(["systemctl", "is-active", "myapp"])
    if result.stdout.strip() == "active":
        print("\n🎉 Deploy successful! Service is running.")
    else:
        print("\n❌ Service is not active!")
        sys.exit(1)


if __name__ == "__main__":
    deploy("/opt/myapp")

十一、subprocess vs 其他選擇
#

工具	適合場景	特色
`subprocess`	通用外部命令	標準庫，功能完整
`shutil`	檔案操作（複製、移動）	不需要呼叫 `cp`、`mv`
`os`	環境變數、路徑操作	`os.environ`、`os.path`
`pathlib`	檔案路徑處理	物件導向 API
`asyncio.create_subprocess_exec`	非同步場景	搭配 async/await