怎么批量自动运行多个Python脚本文件

批量自动运行多个 Python 脚本，核心方案是建立统一的调度入口：通过一个主控 Python 脚本利用 subprocess 模块对目标脚本进行进程级隔离调用，再借助操作系统自带的定时任务工具（如 Linux 的 Crontab 或 Windows 任务计划程序）实现周期性无人值守执行。本文将从主控脚本编写、目录动态扫描、多进程并发加速以及系统定时调度四个维度，展开完整可落地的操作流程与工程避坑指南。

一、为什么采用独立进程的主控调度方式

许多初学者在需要运行多个脚本时，习惯使用 import 导入目标模块并调用函数，或者使用 exec(open("script.py").read())。但这两种方式在批量处理时容易引发全局命名空间污染、内存未释放以及某个脚本抛出异常导致整个流程中断的问题。

更为稳健的标准工程做法是进程隔离：使用 Python 内置的 subprocess 模块作为调度器，将每一个待执行的文件当作独立的子进程启动。子进程之间内存互相隔离，执行完毕立即释放资源，某一个脚本失败不会直接导致后续脚本崩溃，且能准确捕获每个脚本的标准输出与返回状态码。

二、使用主控脚本按顺序批量执行

当多个脚本之间存在先后执行依赖（例如：先抓取数据、再清洗入库、最后发送报表）时，应该采用顺序串行执行。此时主控脚本逐一启动子进程，并在前一个脚本正常退出后才继续下一个。

基础串行调度代码

import sys
import subprocess
from pathlib import Path

# 获取当前 Python 解释器的绝对路径，确保在同一虚拟环境中运行
PYTHON_EXE = sys.executable

# 待执行的脚本清单（支持相对路径或绝对路径）
SCRIPTS = [
    "tasks/01_download_files.py",
    "tasks/02_process_data.py",
    "tasks/03_generate_report.py"
]

def run_all_scripts():
    print(f"[START] 批量运行开始，使用解释器: {PYTHON_EXE}")
    
    for script_rel_path in SCRIPTS:
        script_path = Path(script_rel_path).resolve()
        
        if not script_path.exists():
            print(f"[ERROR] 文件不存在: {script_path}，已跳过")
            continue
            
        print(f"\n---> 开始执行: {script_path.name}")
        try:
            # check=True 会在脚本返回非零退出码时抛出 CalledProcessError
            result = subprocess.run(
                [PYTHON_EXE, str(script_path)],
                check=True,
                text=True,
                capture_output=False # 直接将子进程输出打印到终端
            )
            print(f"[SUCCESS] {script_path.name} 执行完成")
        except subprocess.CalledProcessError as e:
            print(f"[FAILED] {script_path.name} 执行失败，退出码: {e.returncode}")
            # 根据业务需求，决定是中断流程还是继续执行后续脚本
            break

if __name__ == "__main__":
    run_all_scripts()

三、动态扫描指定目录批量运行

如果待执行的脚本较多，或者脚本会动态增加，逐个在代码中硬编码文件名会变得难以维护。此时可以通过文件命名规范（如 step_01_*.py, step_02_*.py），由调度脚本自动扫描目录并按名称自然排序依次执行。

动态扫描并执行示例

import sys
import subprocess
from pathlib import Path

PYTHON_EXE = sys.executable
TASKS_DIR = Path("./scripts_folder")

def run_scripts_in_folder():
    if not TASKS_DIR.is_dir():
        print(f"目录不存在: {TASKS_DIR}")
        return

    # 扫描所有 .py 文件，排除以双下划线开头的文件，并按字母/序号排序
    script_files = sorted([
        f for f in TASKS_DIR.glob("*.py") 
        if not f.name.startswith("__")
    ])
    
    print(f"找到 {len(script_files)} 个待执行脚本:")
    for f in script_files:
        print(f" - {f.name}")

    for script in script_files:
        print(f"\n[RUNNING] 正在执行: {script.name}...")
        # cwd 参数非常关键：将工作目录设为脚本所在目录，防止相对路径报错
        res = subprocess.run(
            [PYTHON_EXE, script.name],
            cwd=str(TASKS_DIR),
            text=True
        )
        if res.returncode != 0:
            print(f"[FAIL] {script.name} 退出异常，状态码: {res.returncode}")
        else:
            print(f"[OK] {script.name} 运行成功")

if __name__ == "__main__":
    run_scripts_in_folder()

四、使用多进程并发加速互不依赖的脚本

在批量处理场景中，如果各个脚本之间没有先后依赖关系（例如分别清洗不同城市的数据、同时从不同数据源爬取信息），串行执行会导致整体耗时随脚本数量线性增加。此时可以使用 concurrent.futures.ThreadPoolExecutor 或 ProcessPoolExecutor 启动多进程并发调度。

并发调度代码示例

import sys
import subprocess
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor, as_completed

PYTHON_EXE = sys.executable

SCRIPTS = [
    "tasks/site_a_crawler.py",
    "tasks/site_b_crawler.py",
    "tasks/site_c_crawler.py",
    "tasks/site_d_crawler.py"
]

def execute_single(script_name):
    """单个脚本的执行包装函数"""
    path = Path(script_name).resolve()
    res = subprocess.run(
        [PYTHON_EXE, str(path)],
        capture_output=True,
        text=True
    )
    return script_name, res.returncode, res.stdout, res.stderr

def run_concurrently(max_workers=3):
    print(f"启动并发调度器，最大并发数: {max_workers}")
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        future_map = {executor.submit(execute_single, s): s for s in SCRIPTS}
        
        for future in as_completed(future_map):
            script_name = future_map[future]
            try:
                name, code, stdout, stderr = future.result()
                if code == 0:
                    print(f"[OK] {name} 执行完成")
                else:
                    print(f"[ERR] {name} 失败，错误输出:\n{stderr}")
            except Exception as exc:
                print(f"[EXC] 调度 {script_name} 时产生异常: {exc}")

if __name__ == "__main__":
    run_concurrently(max_workers=4)

五、配置系统定时任务实现全自动无人值守

编写好主控调度脚本后，最后一步是将主控脚本交由操作系统级调度服务托管，让其每天定时、周期性或在特定事件触发时自动执行。

1. Linux / macOS 环境：Crontab

在 Linux 服务器中，Crontab 是最轻量且通用的定时任务工具。在终端中输入 crontab -e 打开配置，追加一行定时配置：

# 示例：每天凌晨 02:30 自动批量执行所有脚本，并追加日志到 runner.log
30 2 * * * /home/user/myenv/bin/python /home/user/project/main_runner.py >> /home/user/project/runner.log 2>&1

2. Windows 环境：任务计划程序（Task Scheduler）

1. 按下快捷键 Win + R，输入 taskschd.msc 并回车打开“任务计划程序”。
2. 在右侧操作栏点击“创建基本任务”，输入名称（例如 AutoBatchPython）。
3. 选择触发器频率：如“每天”或“每周”，并指定开始时间点。
4. 操作选择“启动程序”：
   • 程序或脚本：填入虚拟环境或 Python 安装目录下的 python.exe 绝对路径（例如 D:\envs\work\Scripts\python.exe）。
   • 添加参数：填入主控脚本名称，如 main_runner.py。
   • 起始于：填入项目根目录绝对路径（如 D:\projects\batch_runner\），这样脚本内部使用的相对路径就会以该目录为基准。
5. 点击完成，可在列表中右键该任务点击“运行”进行单次手动测试。

六、生产环境关键避坑与健壮性检查

批量自动运行通常处于无人值守状态，因此在上线前必须做好以下三项防错设计：

• 工作路径漂移（Working Directory）：子进程默认继承当前主程序的 cwd。在 subprocess.run 中显式指定 cwd=os.path.dirname(script_path)，避免脚本内部读取本地配置文件时发生 FileNotFoundError。
• 错误隔离与警报机制：记录每个子脚本的退出码。在关键业务流中，一旦前序脚本返回非 0 状态码，主控程序应终止后续环节，并可通过邮件、钉钉/飞书机器人发送告警。
• 并发锁与防重入机制：当脚本执行时间较长且触发频率较高时，可能会发生上一次批量任务未结束、下一次又被触发的冲突。可以在主控脚本启动时创建锁文件（如 app.lock），执行完毕后删除，若锁文件存在则自动退出本次触发。

总结

批量自动运行多个 Python 脚本文件的核心不在于复杂的第三方框架，而在于建立清晰的分层管理：

1. 底层实现：采用 subprocess.run 进行子进程隔离，保证各个脚本互不污染、独立退出。
2. 中层调度：编写主控脚本统一控制是串行还是并发，自动扫描目录并规范命名排序。
3. 顶层自动：结合系统的 Crontab 或任务计划程序，绑定虚拟环境解释器的绝对路径，实现全自动定时值守。

按照这一架构构建自动化流水线，可以确保脚本在长期运行中的稳定性、可维护性与扩展性。