如果第一个脚本纯粹是命令行:它应该可以使用 python 脚本完全管理。
一般架构:
- python 脚本从第一个
subprocess
模块开始
- 从第一个脚本读取输出,直到收到要求按 Enter 的消息
- 将所有文件从源目录复制到目标目录
- 发送
\r
到第一个脚本输入
- 等待第一个脚本终止
- 退出
一般要求 :
- 第一个脚本必须是纯 CLI 脚本
- 第一个脚本必须写入标准输出/错误并从标准输入读取 - 如果它读取/写入物理终端(
/dev/tty
在 Unix/Linux 或con:
Dos/Windows 上),它将不起作用
- 处理结束必须在标准输出/错误中可识别
- 如果没有满足上述两个要求,唯一的方法就是等待一段定义的时间
可选操作:
- 如果第一个脚本中有其他交互(读取和/或写入),则需要在脚本中添加重定向,这当然是可行的,但会有点困难
配置 :
- 要运行的命令
- 指示第一个程序已完成处理的字符串(来自命令输出)
- 源目录
- 目标目录
- 要复制的文件名的模式
- 如果时间已定义且输出中没有可识别的字符串:复制前等待的延迟
像这样的脚本应该易于编写和测试,并且能够根据需要管理第一个脚本。
编辑:这里是这样一个脚本的一个例子,仍然没有超时管理。
import subprocess
import os
import shutil
import re
# default values for command execution - to be configured at installation
defCommand = "test.bat"
defEnd = "Appuyez"
defSource = "."
defDest = ".."
# BEWARE : pattern is in regex format !
defPattern="x.*\.txt"
class Launcher(object):
'''
Helper to launch a command, wait for a defined string from stderr or stdout
of the command, copy files from a source folder to a destination folder,
and write a newline to the stdin of the command.
Limits : use blocking IO without timeout'''
def __init__(self, command=defCommand, end=defEnd, source=defSource,
dest=defDest, pattern = defPattern):
self.command = command
self.end = end
self.source = source
self.dest = dest
self.pattern = pattern
def start(self):
'Actualy starts the command and copies the files'
found = False
pipes = os.pipe() # use explicit pipes to mix stdout and stderr
rx = re.compile(self.pattern)
cmd = subprocess.Popen(self.command, shell=True, stdin=subprocess.PIPE,
stdout=pipes[1], stderr=pipes[1])
os.close(pipes[1])
while True:
txt = os.read(pipes[0], 1024)
#print(txt) # for debug
if str(txt).find(self.end) != -1:
found = True
break
# only try to copy files if end string found
if found:
for file in os.listdir(self.source):
if rx.match(file):
shutil.copy(os.path.join(self.source, file), self.dest)
print("Copied : %s" % (file,))
# copy done : write the newline to command input
cmd.stdin.write(b"\n")
cmd.stdin.close()
try:
cmd.wait()
print("Command terminated with %d status" % (cmd.returncode,))
except:
print("Calling terminate ...")
cmd.terminate()
os.close(pipes[0])
# allows to use the file either as an imported module or directly as a script
if __name__ == '__main__':
# parse optional parameters
import argparse
parser = argparse.ArgumentParser(description='Launch a command and copy files')
parser.add_argument('--command', '-c', nargs = 1, default = defCommand,
help="full text of the command to launch")
parser.add_argument('--endString', '-e', nargs = 1, default = defEnd,
dest="end",
help="string that denotes that command has finished processing")
parser.add_argument('--source', '-s', nargs = 1, default = defSource,
help="source folder")
parser.add_argument('--dest', '-d', nargs = 1, default = defDest,
help = "destination folder")
parser.add_argument('--pattern', '-p', nargs = 1, default = defPattern,
help = "pattern (regex format) for files to be copied")
args = parser.parse_args()
# create and start a Launcher ...
launcher = Launcher(args.command, args.end, args.source, args.dest,
args.pattern)
launcher.start()