Linux下我们经常会写一些shell脚本来辅助我们学习或者工作,从而提高效率。
之前就写过相关博客:
Linux下利用shell脚本批量产生内容有规律变化的文件
csh 脚本批量处理文件并将文件扔给程序
bcsh 脚本批量处理文件并将文件扔给程序
import os
import concurrent.futures
import subprocess
def collect_comands(case_list_file):
commands = []
case_list = []
with open(case_list_file, 'r') as f:
case_list = [line.strip() for line in f.readlines()]
#print(case_list)
for case in case_list:
command = f"""
pwd;
ls;
#cp $cmd_option ./cmd
#sed -i "s/pattern/$case/g" ./cmd
# program -i input.txt -o out.txt;
"""
commands.append(command)
return commands
def run_command(cmd):
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, stderr = process.communicate()
if process.returncode != 0:
raise Exception(f"ERROR : {cmd}, stderr: {stderr.decode('utf-8')}, stdout: {stdout.decode('utf-8')}")
return stdout.decode('utf-8')
def run_command_parallel(commands, max_concurrent_commands):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_concurrent_commands) as executor:
futures = [executor.submit(run_command, cmd) for cmd in commands]
for future in concurrent.futures.as_completed(futures):
try:
result = future.result()
print(result)
except Exception as e:
print(f"error: {e}")
if __name__ == "__main__":
case_list_file = "case_list.txt"
if not os.path.exists(case_list_file) :
print(case_list_file, " is not exit, please check it.")
commands = collect_comands(case_list_file)
max_concurrent_commands = 20
run_command_parallel(commands, max_concurrent_commands)