如何基于不同CSV文件与参数批量运行Python项目？

当然可以实现批量自动化运行，下面提供两种实用方案，根据你的实际情况选择：

方案1：不修改原project.py，通过子进程批量调用 #

如果不想改动现有project.py的代码，使用subprocess模块批量启动进程运行脚本是最直接的方式。

实现代码 #

import subprocess

# 定义所有任务的参数配置，每个元素对应一个任务的输入、参数、输出信息
tasks = [
    {"input_file": "aaa.csv", "parm1": 7, "output_file": "a_res.csv"},
    {"input_file": "bbb.csv", "parm1": 4, "output_file": "b_res.csv"},
    {"input_file": "ccc.csv", "parm1": 2, "output_file": "c_res.csv"},
    {"input_file": "ddd.csv", "parm1": 6, "output_file": "d_res.csv"},
    # 可继续添加更多任务
]

# 循环执行每个任务
for task in tasks:
    # 构造命令行参数（需确保project.py支持命令行传参）
    cmd = [
        "python", "project.py",
        "--input", task["input_file"],
        "--parm1", str(task["parm1"]),
        "--output", task["output_file"]
    ]
    print(f"正在执行任务：处理{task['input_file']}，参数parm1={task['parm1']}")
    # 执行命令并等待任务完成
    subprocess.run(cmd, check=True)
    print(f"任务完成：结果已保存到{task['output_file']}\n")

配套修改（若project.py未支持命令行传参） #

给project.py添加命令行参数解析逻辑：

# project.py 新增代码
import argparse
import pandas as pd

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True, help="输入CSV文件路径")
    parser.add_argument("--parm1", type=int, required=True, help="参数parm1")
    parser.add_argument("--output", required=True, help="输出CSV文件路径")
    args = parser.parse_args()

    # 原有业务逻辑，示例如下
    df = pd.read_csv(args.input)
    df["processed"] = df["target_column"] * args.parm1
    df.to_csv(args.output, index=False)

if __name__ == "__main__":
    main()

方案2：重构project.py为函数调用，提升运行效率 #

如果可以修改原项目代码，把核心处理逻辑封装成函数，直接在批量脚本中导入调用，避免启动多个子进程，运行效率更高。

步骤1：重构project.py #

将核心逻辑封装为可调用的函数：

# project.py
import pandas as pd

def process_data(input_file, parm1, output_file):
    # 原有业务处理逻辑
    df = pd.read_csv(input_file)
    df["processed"] = df["target_column"] * parm1
    df.to_csv(output_file, index=False)

if __name__ == "__main__":
    # 保留原命令行调用入口，兼容之前的使用方式
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", required=True)
    parser.add_argument("--parm1", type=int, required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    process_data(args.input, args.parm1, args.output)

步骤2：编写批量运行脚本 #

# batch_run.py
from project import process_data

# 定义任务列表
tasks = [
    ("aaa.csv", 7, "a_res.csv"),
    ("bbb.csv", 4, "b_res.csv"),
    ("ccc.csv", 2, "c_res.csv"),
    ("ddd.csv", 6, "d_res.csv"),
    # 可继续添加任务
]

for input_file, parm1, output_file in tasks:
    print(f"正在处理{input_file}，parm1={parm1}")
    process_data(input_file, parm1, output_file)
    print(f"处理完成，结果保存到{output_file}\n")

两种方案均能实现批量自动化运行：方案1无需修改原代码，适合快速落地；方案2运行效率更高，适合处理大量文件的场景。

内容的提问来源于stack exchange，提问作者Lkmuraz