python - 如何使 python 脚本在 bash 和 python 中都可以通过管道传输

Question

总结:我想在命令行上编写类似于 bash 脚本的 python 脚本，但我也想在 python 中轻松地将它们组合在一起。我遇到麻烦的地方是使后者发生的胶水。

想象一下，我写了两个脚本，script1.py 和 script2.py，我可以像这样将它们组合在一起:

echo input_string | ./script1.py -a -b | ./script2.py -c -d

如何从另一个 python 文件中获取此行为？这是我知道的方式，但我不喜欢:

arg_string_1 = convert_to_args(param_1, param_2)
arg_string_2 = convert_to_args(param_3, param_4)
output_string = subprocess.check_output("echo " + input_string + " | ./script1.py " + arg_string_1 + " | ./script2.py " + arg_string_2)

如果我不想利用多线程，我可以这样做(？):

input1  = StringIO(input_string)
output1 = StringIO()
script1.main(param_1, param_2, input1, output1)
input2  = StringIO(output1.get_value())
output2 = StringIO()
script2.main(param_3, param_4, input2, output2)

---

这是我尝试过的方法，但我在编写胶水时遇到了困难。我很感激学习如何完成下面的方法，或者提出更好的设计/方法的建议!

我的方法:我编写的 script1.py 和 script2.py 看起来像:

#!/usr/bin/python3

... # import sys and define "parse_args"

def main(param_1, param_2, input, output):
   for line in input:
     ...
     print(stuff, file=output)

if __name__ == "__main__":
  parameter_1, parameter_2 = parse_args(sys.argv)
  main(parameter_1, parameter_2, sys.stdin, sys.stdout)

然后我想写这样的东西，但不知道如何完成:

pipe_out, pipe_in = ????
output = StringIO()
thread_1 = Thread(target=script1.main, args=(param_1, param_2, StreamIO(input_string), pipe_out))
thread_2 = Thread(target=script2.main, args=(param_3, param_4, pipe_in, output)
thread_1.start()
thread_2.start()
thread_1.join()
thread_2.join()
output_str = output.get_value()

Answer 1

对于“管道输入”，使用 sys.stdin 和 readlines() 方法。 (使用方法 read() 将一次读取一个字符。)

要将信息从一个线程传递到另一个线程，您可以使用Queue。您必须定义一种方式来表示数据结束。在我的示例中，由于在线程之间传递的所有数据都是 str，我只是使用 None 对象来表示数据结束(因为它不能出现在传输的数据中)。

也可以使用更多的线程，或者在线程中使用不同的函数。

为了简单起见，我没有在示例中包含 sys.argv。修改它以获取参数 (parameter1, ...) 应该很容易。

import sys
from threading import Thread
from Queue import Queue
import fileinput

def stdin_to_queue( output_queue ):
  for inp_line in sys.stdin.readlines():     # input one line at at time                                                
    output_queue.put( inp_line, True, None )  # blocking, no timeout
  output_queue.put( None, True, None )    # signal the end of data                                                  


def main1(input_queue, output_queue, arg1, arg2):
  do_loop = True
  while do_loop:
    inp_data = input_queue.get(True)
    if inp_data is None:
      do_loop = False
      output_queue.put( None, True, None )  # signal end of data                                                    
    else:
      out_data = arg1 + inp_data.strip('\r\n').upper() + arg2 #  or whatever transformation...                                    
      output_queue.put( out_data, True, None )

def queue_to_stdout(input_queue):
  do_loop = True
  while do_loop:
    inp_data = input_queue.get(True)
    if inp_data is None:
      do_loop = False
    else:
      sys.stdout.write( inp_data )


def main():
  q12 = Queue()
  q23 = Queue()
  q34 = Queue()
  t1 = Thread(target=stdin_to_queue, args=(q12,) )
  t2 = Thread(target=main1, args=(q12,q23,'(',')') )
  t3 = Thread(target=main1, args=(q23,q34,'[',']') )
  t4 = Thread(target=queue_to_stdout, args=(q34,))
  t1.start()
  t2.start()
  t3.start()
  t4.start()


main()

最后，我用一个文本文件测试了这个程序(python2)。

head sometextfile.txt | python script.py