引言

在Python开发中，经常需要按特定顺序执行多个Python脚本文件。这种需求常见于：

• 数据处理流水线（数据提取 → 数据清洗 → 数据分析）
• 自动化测试套件
• 任务编排和工作流管理
• 多步骤的数据处理或机器学习流程

本教程将介绍4种在Python3中顺序执行多个.py文件的方法，并提供实际代码示例和最佳实践建议。

方法一：使用os.system()

os.system() 是最简单直接的方法，它允许你执行系统命令来运行Python脚本。

import os

# 定义要执行的脚本列表
scripts = ["step1.py", "step2.py", "step3.py"]

for script in scripts:
    # 使用系统命令执行Python脚本
    os.system(f"python3 {script}")
    
    # 检查执行状态（可选）
    if os.system(f"python3 {script}") != 0:
        print(f"脚本 {script} 执行失败!")
        break

方法二：使用subprocess模块

subprocess 模块提供了更强大、更灵活的方式来执行外部命令，是Python3推荐的方法。

import subprocess

scripts = ["step1.py", "step2.py", "step3.py"]

for script in scripts:
    try:
        # 执行脚本并等待完成
        result = subprocess.run(
            ["python3", script],
            check=True,  # 检查返回码
            capture_output=True,  # 捕获输出
            text=True  # 文本模式输出
        )
        
        # 打印输出
        print(f"执行 {script} 成功!")
        print(f"输出: {result.stdout}")
        
    except subprocess.CalledProcessError as e:
        print(f"执行 {script} 失败! 错误码: {e.returncode}")
        print(f"错误输出: {e.stderr}")
        break

方法三：将脚本作为模块导入

如果脚本被设计为模块，可以直接导入并调用其主函数。

def main():
    print("执行步骤1...")
    # 实际逻辑代码

if __name__ == "__main__":
    main()

import step1
import step2
import step3

def run_pipeline():
    step1.main()
    step2.main()
    step3.main()

if __name__ == "__main__":
    run_pipeline()

方法四：动态导入与执行

对于更高级的场景，可以使用Python的importlib模块动态导入脚本。

import importlib.util

scripts = ["step1", "step2", "step3"]  # 不带.py后缀

for script_name in scripts:
    # 动态加载模块
    spec = importlib.util.spec_from_file_location(script_name, f"{script_name}.py")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    
    # 执行main函数
    if hasattr(module, 'main'):
        module.main()
    else:
        print(f"警告: {script_name}.py 没有main函数")

方法比较与选择指南

高级技巧与最佳实践

1. 处理依赖关系

# 确保前置脚本成功执行
scripts = [
    ("step1.py", True),   # True表示必须成功
    ("step2.py", False),  # False表示可跳过失败
    ("step3.py", True)
]

for script, required in scripts:
    result = subprocess.run(["python3", script])
    
    if result.returncode != 0:
        if required:
            print(f"关键脚本 {script} 失败，终止执行")
            break
        else:
            print(f"非关键脚本 {script} 失败，继续执行")

2. 并行与顺序混合执行

from concurrent.futures import ThreadPoolExecutor

# 可以并行执行的脚本
parallel_scripts = ["data_fetch.py", "data_download.py"]

# 顺序执行的脚本
sequential_scripts = ["data_clean.py", "data_analyze.py"]

# 并行执行第一阶段
with ThreadPoolExecutor() as executor:
    executor.map(lambda s: subprocess.run(["python3", s]), parallel_scripts)

# 顺序执行第二阶段
for script in sequential_scripts:
    subprocess.run(["python3", script])

3. 使用配置文件管理执行流程

# config.yaml
scripts:
  - name: data_extraction
    file: extract.py
    args: ["--date", "2023-10-01"]
    required: true
    
  - name: data_processing
    file: process.py
    timeout: 300  # 5分钟超时

常见问题与解决方案

Q1: 脚本执行权限问题

解决方案：

• 使用 chmod +x script.py 添加执行权限
• 在Python脚本第一行添加shebang: #!/usr/bin/env python3

Q2: 脚本路径问题

解决方案：

import os
import subprocess

# 获取当前脚本所在目录
base_dir = os.path.dirname(os.path.abspath(__file__))

script_path = os.path.join(base_dir, "subfolder", "script.py")
subprocess.run(["python3", script_path])

Q3: 环境变量和依赖问题

解决方案：

• 使用虚拟环境（venv）
• 在调用subprocess时指定环境变量：

  env = {**os.environ, "CUSTOM_VAR": "value"}
  subprocess.run(["python3", "script.py"], env=env)