subprocess 模块

import subprocess

# 执行 'ls -la' 命令
# args 传列表：第一个是命令，后面是参数
# 默认行为：stdout/stderr 直接打印到终端，run() 阻塞直到命令结束
result = subprocess.run(["ls", "-la"])

# returncode：0 表示成功，非 0 表示失败
print(f"命令执行完毕，退出码: {result.returncode}")

import subprocess

# capture_output=True：将 stdout 和 stderr 重定向到管道
# text=True：将字节流解码为字符串（否则返回 bytes）
result = subprocess.run(
    ["ls", "-la"],
    capture_output=True,  # 捕获 stdout + stderr
    text=True             # 以文本模式返回（等同于 encoding="utf-8"）
)

# result.stdout：标准输出（字符串）
# result.stderr：标准错误（字符串）
# result.returncode：退出码
print("--- STDOUT ---")
print(result.stdout)
print("--- STDERR ---")
print(result.stderr)
print(f"退出码: {result.returncode}")

import subprocess

# ---- FileNotFoundError：命令不存在 ----
try:
    subprocess.run(["non_existent_command"], check=True)
except FileNotFoundError as e:
    print(f"错误：命令未找到 - {e}")

# ---- CalledProcessError：命令执行失败（非零退出码） ----
try:
    # check=True：退出码非 0 时自动抛 CalledProcessError
    subprocess.run(
        ["ls", "/non_existent_directory"],
        check=True,
        capture_output=True,
        text=True
    )
except subprocess.CalledProcessError as e:
    print(f"错误：命令执行失败，退出码 {e.returncode}")
    print(f"STDERR: {e.stderr.strip()}")
    print(f"STDOUT: {e.stdout.strip()}")

subprocess.run(
    args,                  # 命令列表，如 ["ls", "-la"]
    *, 
    stdin=None,            # 子进程的标准输入来源
    input=None,            # 直接传入数据到 stdin（与 stdin 互斥）
    stdout=None,           # 子进程标准输出去向
    stderr=None,           # 子进程标准错误去向
    capture_output=False,  # = stdout=PIPE, stderr=PIPE 的快捷方式
    shell=False,           # 是否通过 shell 执行（⚠️ 安全风险）
    cwd=None,              # 子进程工作目录
    timeout=None,          # 超时秒数，超时抛 TimeoutExpired
    check=False,           # 非零退出码时抛 CalledProcessError
    encoding=None,         # 编码，如 "utf-8"
    errors=None,           # 编码错误处理策略，如 "replace"
    text=None,             # True = 文本模式（旧名 universal_newlines）
    env=None,              # 子进程环境变量字典
    **other_popen_kwargs   # 传递给 Popen 的其他参数
)

class subprocess.Popen:
    """底层子进程控制器。run() 的内部实现就是 Popen + communicate()。"""

    # ---- 构造参数（与 run() 大体相同，以下为重点差异） ----
    # stdin/stdout/stderr: 可设为 subprocess.PIPE 创建管道
    # bufsize: 管道缓冲区大小，-1=系统默认，0=无缓冲（二进制模式），1=行缓冲（文本模式）

    # ---- 核心属性 ----
    # .pid: 子进程 ID
    # .returncode: 退出码（未退出时为 None）

    # ---- 核心方法 ----
    # .poll()              → 检查是否退出，返回 returncode 或 None
    # .wait(timeout=None)  → 阻塞等待退出，超时抛 TimeoutExpired
    # .communicate(input=None, timeout=None) → 发送输入 + 读取全部输出 + 等待退出
    # .send_signal(sig)    → 发送信号
    # .terminate()         → 发 SIGTERM（优雅终止）
    # .kill()              → 发 SIGKILL（强制杀死）

import subprocess

# 启动子进程，将 stdout 重定向到管道
# bufsize=1 + text=True：启用行缓冲，readline() 每读到一行就返回
process = subprocess.Popen(
    ["ping", "-c", "5", "example.com"],
    stdout=subprocess.PIPE,   # 创建 stdout 管道
    stderr=subprocess.PIPE,   # 创建 stderr 管道
    text=True,                # 文本模式
    bufsize=1                 # 行缓冲（text 模式下生效）
)

print("--- 实时读取 stdout ---")
while True:
    # readline() 阻塞等待一行输出
    output = process.stdout.readline()

    # 两个退出条件：
    # 1. readline() 返回空字符串（管道已关闭）
    # 2. poll() 不再返回 None（进程已退出）
    if output == '' and process.poll() is not None:
        break
    if output:
        # strip() 去掉行尾换行
        print(output.strip()

# 确保进程已结束，获取最终退出码
return_code = process.wait()
print(f"\n进程已结束，退出码: {return_code}")

import subprocess

# ---- 模拟 shell 的 'ls -la | grep .py' ----

# 第一步：启动 ls 进程，stdout 接管道
ls_process = subprocess.Popen(
    ["ls", "-la"],
    stdout=subprocess.PIPE,  # ls 的输出通过管道传出
    text=True
)

# 第二步：启动 grep 进程，stdin 接 ls 的 stdout 管道
grep_process = subprocess.Popen(
    ["grep", ".py"],
    stdin=ls_process.stdout,  # grep 的输入 = ls 的输出
    stdout=subprocess.PIPE,   # grep 的输出也接管道
    text=True
)

# 第三步：关闭 ls_process.stdout 的引用
# 原因：ls_process.stdout 被 grep_process 接管后，
# 父进程不再需要它。如果不关闭，ls 进程在 grep 还没读完时
# 可能无法正常退出（因为管道的写入端还有一个引用）
ls_process.stdout.close()

# 第四步：从管道链末端读取最终输出
# communicate() 会读取全部 stdout/stderr 并等待进程退出
output, errors = grep_process.communicate()
print(output)

# 第五步（可选）：检查两个进程的退出码
ls_exit = ls_process.wait()
grep_exit = grep_process.returncode
print(f"ls 退出码: {ls_exit}, grep 退出码: {grep_exit}")

import subprocess
import time
import signal

# ---- 启动 ----
print("启动一个长时间运行的进程...")
process = subprocess.Popen(["sleep", "60"])
print(f"进程已启动，PID: {process.pid}")

# ---- 监控 ----
time.sleep(5)  # 等待 5 秒

if process.poll() is None:
    print("进程仍在运行。现在尝试终止它...")
    # ---- 优雅终止 (SIGTERM) ----
    # SIGTERM 是"请退出"信号，进程可以捕获它做清理工作
    process.terminate()
else:
    print("进程已经结束。")

# ---- 等待终止 ----
try:
    # 给进程 5 秒时间优雅退出
    exit_code = process.wait(timeout=5)
    print(f"进程已成功终止，退出码: {exit_code}")
except subprocess.TimeoutExpired:
    print("终止超时，进程拒绝退出，强制杀死！")
    # ---- 强制杀死 (SIGKILL) ----
    # SIGKILL 无法被捕获，操作系统直接终止进程
    process.kill()
    exit_code = process.wait()  # 仍需 wait() 收割僵尸进程
    print(f"进程已被强制杀死，退出码: {exit_code}")

import subprocess

# ---- 模拟 'cat file.txt | sort | uniq -c' ----

# 进程 A：cat（读取文件内容）
p_cat = subprocess.Popen(
    ["cat", "/etc/hosts"],    # macOS/Linux 系统文件，一定存在
    stdout=subprocess.PIPE,
    text=True
)

# 进程 B：sort（排序）
p_sort = subprocess.Popen(
    ["sort"],
    stdin=p_cat.stdout,       # 输入来自 cat
    stdout=subprocess.PIPE,
    text=True
)

# 进程 C：uniq -c（去重并计数）
p_uniq = subprocess.Popen(
    ["uniq", "-c"],
    stdin=p_sort.stdout,      # 输入来自 sort
    stdout=subprocess.PIPE,
    text=True
)

# 关闭中间管道的父进程引用
p_cat.stdout.close()
p_sort.stdout.close()

# 从管道链末端读取
output, _ = p_uniq.communicate()
print(output)

# 等待所有进程结束
p_cat.wait()
p_sort.wait()
print(f"cat: {p_cat.returncode}, sort: {p_sort.returncode}, uniq: {p_uniq.returncode}")

import subprocess

# ---- 模式 1：run() 捕获输出（最常用） ----
result = subprocess.run(["ls"], capture_output=True, text=True)
print(result.stdout)

# ---- 模式 2：Popen 单进程管道（实时读取） ----
p = subprocess.Popen(["ls"], stdout=subprocess.PIPE, text=True)
for line in p.stdout:
    print(line.strip()
p.wait()

# ---- 模式 3：管道链（已在 3.6 节详述） ----
# 见 3.6 节

# ---- 模式 4：stderr 合并到 stdout ----
# 场景：不区分标准输出和错误，统一处理
result = subprocess.run(
    ["ls", "/nonexistent"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,  # stderr 合并到 stdout
    text=True
)
print(result.stdout)  # 包含 stdout + stderr

# ---- 模式 5：丢弃输出（静默执行） ----
# 场景：只关心命令是否成功，不关心输出
result = subprocess.run(
    ["ls"],
    stdout=subprocess.DEVNULL,  # 丢弃 stdout
    stderr=subprocess.DEVNULL,  # 丢弃 stderr
    check=True
)
print("命令执行成功")

需要执行子进程？
├── 只需执行命令、拿结果 → subprocess.run()
├── 需要实时读取输出 → subprocess.Popen + stdout=PIPE
├── 需要管道链 (A | B | C) → subprocess.Popen（多实例）
├── 需要并行多个子进程 → subprocess.Popen（多实例）
└── 遗留代码中 os.system() → 迁移到 subprocess.run()

import subprocess
import shlex

# ---- 攻击场景 ----
filename = "file.txt; rm -rf /"  # 恶意用户输入

# 危险！shell=True 会把分号解释为命令分隔符
# 实际执行：cat file.txt → rm -rf /
subprocess.run(f"cat {filename}", shell=True)  # 💀 命令注入！

# ---- 防御 1：列表传参（首选） ----
# shell=False（默认）时，filename 整体被视为一个参数
# cat 会尝试打开名为 "file.txt; rm -rf /" 的文件（当然不存在）
subprocess.run(["cat", filename])  # ✅ 安全

# ---- 防御 2：shlex.quote()（必须 shell=True 时） ----
# shlex.quote() 会在特殊字符外加引号，使其失去 shell 语法含义
quoted = shlex.quote(filename)
# quoted = "'file.txt; rm -rf /'"  → shell 把它当字符串，不当命令
subprocess.run(f"cat {quoted}", shell=True)  # ✅ 安全但非首选

# ---- 防御 3：shlex.split()（把命令字符串安全拆成列表） ----
cmd_string = "ls -la 'a directory with spaces'"
args = shlex.split(cmd_string)
# args = ['ls', '-la', 'a directory with spaces']
subprocess.run(args)  # ✅ 安全，无需 shell=True

import subprocess

# ---- run() 的 timeout 参数 ----
# 超时后，run() 会先 kill 子进程，再抛 TimeoutExpired
try:
    result = subprocess.run(
        ["sleep", "100"],
        timeout=3  # 3 秒超时
    )
except subprocess.TimeoutExpired as e:
    print(f"命令超时！已终止进程")
    print(f"退出码: {e.returncode}")  # None（进程被 kill）
    # e.stdout / e.stderr：超时前已捕获的输出

# ---- Popen 的超时处理（更精细） ----
process = subprocess.Popen(["sleep", "100"])
try:
    exit_code = process.wait(timeout=3)
except subprocess.TimeoutExpired:
    print("等待超时，尝试优雅终止...")
    process.terminate()  # SIGTERM
    try:
        process.wait(timeout=2)  # 再给 2 秒
        print("进程已优雅退出")
    except subprocess.TimeoutExpired:
        print("进程拒绝退出，强制杀死")
        process.kill()  # SIGKILL
        process.wait()  # 必须收割僵尸进程

import subprocess
import signal
import os

# ---- 信号层次：从温和到暴力 ----
# SIGTERM (15)："请退出"，进程可捕获并做清理
# SIGINT  (2) ："Ctrl+C"，进程可捕获
# SIGKILL (9) ："强制终止"，进程无法捕获，OS 直接杀

process = subprocess.Popen(["sleep", "100"])

# 发送 SIGTERM（等同 process.terminate()）
process.send_signal(signal.SIGTERM)

# 等一下看进程是否退出
try:
    process.wait(timeout=3)
except subprocess.TimeoutExpired:
    # 发送 SIGKILL（等同 process.kill()）
    process.send_signal(signal.SIGKILL)
    process.wait()

# ---- 跨平台注意 ----
# Windows 上：
# - terminate() 调用 TerminateProcess()
# - kill() 与 terminate() 行为相同
# - send_signal() 仅支持 SIGTERM 和 SIGKILL
# POSIX 上：
# - terminate() 发送 SIGTERM
# - kill() 发送 SIGKILL
# - send_signal() 可发送任何信号

import subprocess
import shutil
from typing import Optional

def run_git(*args: str, cwd: Optional[str] = None) -> str:
    """安全的 Git 命令封装。

    Args:
        *args: Git 子命令和参数，如 "log", "--oneline", "-10"
        cwd: 工作目录

    Returns:
        命令的标准输出（已去除首尾空白）

    Raises:
        FileNotFoundError: Git 未安装
        CalledProcessError: Git 命令执行失败
    """
    # 先检查 git 是否可用
    if not shutil.which("git"):
        raise FileNotFoundError("Git 未安装或不在 PATH 中")

    result = subprocess.run(
        ["git", *args],       # 列表传参，安全
        capture_output=True,
        text=True,
        check=True,           # 非零退出码抛异常
        cwd=cwd,
        timeout=30            # 30 秒超时
    )
    return result.stdout.strip()

# 使用示例
try:
    # 查看最近 5 条提交
    log = run_git("log", "--oneline", "-5")
    print(log)

    # 查看当前分支
    branch = run_git("branch", "--show-current")
    print(f"当前分支: {branch}")
except subprocess.CalledProcessError as e:
    print(f"Git 命令失败: {e.stderr}")

import subprocess
import sys
import re

def run_with_progress(command: list[str], pattern: str = r"(\d+)%") -> None:
    """运行命令并实时显示进度。

    Args:
        command: 命令列表
        pattern: 匹配进度百分点的正则表达式
    """
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,  # stderr 合并到 stdout
        text=True,
        bufsize=1                  # 行缓冲
    )

    last_percent = 0
    while True:
        line = process.stdout.readline()
        if line == '' and process.poll() is not None:
            break
        if line:
            # 尝试从输出中提取进度
            match = re.search(pattern, line)
            if match:
                percent = int(match.group(1)
                if percent > last_percent:
                    last_percent = percent
                    # 在同一行更新进度条
                    bar_len = 40
                    filled = int(bar_len * percent / 100)
                    bar = '█' * filled + '░' * (bar_len - filled)
                    sys.stdout.write(f"\r[{bar}] {percent}%")
                    sys.stdout.flush()
            else:
                # 非进度行，直接打印
                print(line.strip()

    process.wait()
    if last_percent > 0:
        print()  # 进度条换行
    print(f"完成，退出码: {process.returncode}")

# 使用示例：用 curl 下载（curl 会输出进度信息）
# run_with_progress(["curl", "-O", "https://example.com/largefile.zip"])

import subprocess
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def ping_host(host: str, count: int = 3) -> dict:
    """Ping 指定主机并返回结果。"""
    try:
        result = subprocess.run(
            ["ping", "-c", str(count), "-W", "2", host],
            capture_output=True,
            text=True,
            timeout=count * 3 + 2  # 合理超时
        )
        return {
            "host": host,
            "alive": result.returncode == 0,
            "output": result.stdout if result.returncode == 0 else result.stderr
        }
    except subprocess.TimeoutExpired:
        return {"host": host, "alive": False, "output": "超时"}
    except FileNotFoundError:
        return {"host": host, "alive": False, "output": "ping 命令不可用"}

# 并行 ping 多台主机
hosts = ["example.com", "google.com", "github.com", "nonexistent.test"]

start = time.time()
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = {executor.submit(ping_host, h): h for h in hosts}
    for future in as_completed(futures):
        result = future.result()
        status = "✅ 在线" if result["alive"] else "❌ 离线"
        print(f"{result['host']}: {status}")

print(f"总耗时: {time.time() - start:.2f}s")

import subprocess

# 场景：子进程需要交互式输入（如确认提示）
# 使用 communicate() 一次性发送所有输入

# 模拟一个需要确认的命令
# 这里用 Python 脚本模拟交互式程序
script = """
name = input("请输入你的名字: ")
age = input("请输入你的年龄: ")
print(f"你好, {name}! 你 {age} 岁了。")
"""

process = subprocess.Popen(
    ["/usr/bin/env", "python3", "-c", script],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# communicate() 一次性发送所有输入
# 注意：输入之间用换行分隔，模拟用户按回车
output, errors = process.communicate(input="张三\n25\n", timeout=10)
print(output)

# ---- 重要：communicate() 只能调用一次！ ----
# 如果需要多次交互，必须用 process.stdin.write() + process.stdout.read()
# 但这很容易死锁，推荐使用 pexpect 等第三方库

# 攻击向量一览
user_input = "file.txt; rm -rf /"        # 命令分隔
user_input = "file.txt && rm -rf /"      # 条件执行
user_input = "$(rm -rf /)"               # 命令替换
user_input = "`rm -rf /`"                # 反引号替换
user_input = "file.txt | rm -rf /"       # 管道注入
user_input = "file.txt\nrm -rf /"        # 换行注入

try:
    process.wait(timeout=5)
except subprocess.TimeoutExpired:
    process.kill()    # 必须手动 kill！
    process.wait()    # 必须收割！

process = subprocess.Popen(
    ["large_output_command"],
    stdout=subprocess.PIPE,
    text=True
)

# 流式处理，不占大量内存
for line in process.stdout:
    process_line(line)  # 处理完即丢弃

process.wait()

# ---- 问题：Windows 上很多命令输出 GBK 编码 ----
# 默认 text=True 使用 locale.getpreferredencoding()
# Windows 中文系统上是 GBK (cp936)

# ---- 解决方案 1：显式指定编码 ----
result = subprocess.run(
    ["some_command"],
    capture_output=True,
    encoding="gbk",        # 显式指定子进程的编码
    errors="replace"       # 无法解码的字符用 � 替代
)

# ---- 解决方案 2：二进制模式 + 手动解码 ----
result = subprocess.run(
    ["some_command"],
    capture_output=True
    # 不设 text=True，返回 bytes
)
try:
    text = result.stdout.decode("utf-8")
except UnicodeDecodeError:
    text = result.stdout.decode("gbk", errors="replace")

# ---- 解决方案 3：环境变量强制 UTF-8 ----
import os
env = os.environ.copy()
env["PYTHONIOENCODING"] = "utf-8"
env["LANG"] = "en_US.UTF-8"
result = subprocess.run(["some_command"], env=env, capture_output=True, text=True)

import shutil
import subprocess

# 安全做法：先检查命令是否存在
git_path = shutil.which("git")
if git_path:
    subprocess.run([git_path, "--version"])
else:
    print("Git 未安装")

import subprocess
import sys

# 关键：用 sys.executable 确保使用同一个 Python 解释器
# 不要硬编码 "python" 或 "python3"
result = subprocess.run(
    [sys.executable, "my_script.py", "--arg", "value"],
    capture_output=True,
    text=True,
    check=True
)
print(result.stdout)