如何在Python中实现类似Perl的命令行工具链式文件操作

问题背景 #

我们正在将基础设施中的Perl代码迁移到Python，遇到了外部命令管道IO的差异问题：

• Perl可以通过open结合管道符号|，直接将外部命令的输出作为文件句柄读取（如调用read_file处理各种格式文件），或把输出通过管道传给gzip等命令；
• 但Python直接用open()调用命令会抛出FileNotFoundError，因为open()仅用于操作文件，无法解析执行外部命令。

Perl中的实现示例 #

读取外部命令输出（read_file）

my $fh = IO::File->new();
$fh->open("read_file $file_name |") or abend_pgm("open: $!");
while ( my $line = <$fh> ) {
  # 处理每行内容
}

写入到外部命令管道（gzip）

open (my $fh, "| /usr/bin/gzip -c > $file_name ");
# 写入内容到管道

Python的替代实现 #

Python没有直接对应Perl中open管道的语法，必须使用subprocess模块来实现类似功能，以下是两种场景的具体实现：

1. 读取外部命令的输出（模拟read_file调用） #

推荐使用subprocess.Popen，可以像操作文件句柄一样逐行读取命令输出，同时支持安全的参数传递：

安全模式（推荐，避免Shell注入）

将命令和参数以列表形式传入，无需依赖Shell解析：

import subprocess

file_name = "/ds/tmp/177598.TEST_20240225.dat.gz"
# 命令和参数拆分列表
cmd = ["/ds/CENTOS/common/bin/read_file", file_name]

# 打开管道，text=True让输出为字符串（而非字节流）
with subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True) as proc:
    # 逐行读取输出，和Perl的while循环逻辑一致
    for line in proc.stdout:
        # 处理每行内容，示例：打印行（end=''避免重复换行）
        print(line, end='')

Shell解析模式（适用于含管道/通配符的复杂命令）

如果命令需要Shell解析（如包含重定向、管道），可以设置shell=True，但需注意输入的安全性：

import subprocess

file_name = "/ds/tmp/177598.TEST_20240225.dat.gz"
cmd = f"/ds/CENTOS/common/bin/read_file {file_name}"

with subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True, shell=True) as proc:
    for line in proc.stdout:
        # 处理逻辑
        pass

2. 写入到外部命令管道（模拟gzip压缩写入） #

同样使用subprocess.Popen，通过stdin=subprocess.PIPE将内容传入外部命令：

Shell解析模式（对应Perl的写法）

import subprocess

file_name = "output.gz"
cmd = f"/usr/bin/gzip -c > {file_name}"

with subprocess.Popen(cmd, stdin=subprocess.PIPE, text=True, shell=True) as proc:
    # 写入文本内容到管道
    proc.stdin.write("待压缩的文本内容\n")
    proc.stdin.write("另一行文本\n")
    # 关闭输入流，触发gzip完成压缩
    proc.stdin.close()
    # 等待命令执行完毕
    proc.wait()

安全模式（避免Shell依赖）

直接将gzip的输出写入文件，无需Shell处理重定向：

import subprocess

file_name = "output.gz"

# 打开输出文件，同时启动gzip命令
with open(file_name, 'wb') as out_file, subprocess.Popen(
    ["/usr/bin/gzip", "-c"],
    stdin=subprocess.PIPE,
    stdout=out_file
) as proc:
    # 写入字节内容（若用text=True则可写入字符串）
    proc.stdin.write(b"待压缩的二进制内容\n")
    proc.stdin.close()
    proc.wait()

关键结论 #

• Python没有类似Perl中open()直接处理管道命令的语法，必须通过subprocess模块实现；
• popen()是Python早期的底层方法，现在推荐使用更灵活、安全的subprocess.Popen或更高层的subprocess.run/subprocess.check_output等函数；
• 优先使用列表形式传递命令参数（非Shell模式），避免Shell注入风险，仅在命令需要Shell解析时使用shell=True。

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