在Python自动化脚本开发中,我们常常需要通过Pexpect启动外部子进程并捕获其输出,但默认获取的输出没有时间标记,当输出内容较多时很难对应到具体的产生时间,给问题排查带来不便。结合Logging模块可以很方便地为每一行子进程输出添加时间戳,让日志信息更完整。

环境准备
首先需要确保Python环境中已经安装了Pexpect库,如果没有安装可以通过pip命令进行安装:
pip install pexpect
Logging模块是Python标准库的一部分,不需要额外安装,直接导入即可使用。
核心实现思路
整体的实现逻辑可以分为三个部分:
- 配置Logging模块,设置日志的输出格式,包含时间戳、日志级别、具体内容等信息
- 使用Pexpect启动目标子进程,实时读取子进程的输出内容
- 将读取到的每一行子进程输出通过Logging模块进行输出,自动带上配置好的时间戳
完整代码示例
下面是一个完整的示例代码,实现了启动一个持续输出内容的子进程,并为每一行输出添加时间戳:
import pexpect
import logging
import time
# 配置Logging模块
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
def run_subprocess_with_timestamp():
# 启动子进程,这里以执行一个持续打印时间的shell命令为例
# 实际使用时可以替换为需要启动的任何子进程命令
child = pexpect.spawn('bash -c "while true; do echo test_output; sleep 1; done"', encoding='utf-8')
try:
# 循环读取子进程的输出
while True:
# 读取一行输出,设置超时时间
line = child.readline(timeout=2)
if not line:
break
# 去除行尾的换行符
line = line.strip()
# 通过Logging输出,自动带上时间戳
logging.info(f"子进程输出: {line}")
except pexpect.TIMEOUT:
# 超时说明子进程暂时没有输出,继续循环
pass
except pexpect.EOF:
# 子进程结束
logging.info("子进程执行结束")
finally:
# 关闭子进程
child.close()
if __name__ == "__main__":
run_subprocess_with_timestamp()
代码细节说明
Logging配置部分
在basicConfig方法中,我们通过format参数设置了日志的输出格式,其中%(asctime)s就是自动生成的时间戳占位符,datefmt参数用来指定时间戳的显示格式,这里设置为年-月-日 时:分:秒的形式,也可以根据需求调整。
Pexpect子进程读取部分
使用pexpect.spawn启动子进程,这里示例中使用了一个持续每秒输出test_output的shell命令,实际使用时可以替换为比如ping ipipp.com或者其他需要执行的外部命令。readline方法用来读取子进程的一行输出,设置了2秒的超时时间,如果超时没有读到内容会抛出pexpect.TIMEOUT异常,我们捕获这个异常后继续循环即可。
输出处理部分
读取到子进程的输出后,先去除行尾的换行符,然后通过logging.info方法输出,此时Logging会自动按照之前配置的格式添加时间戳,不需要我们手动处理时间相关的逻辑。
常见问题处理
子进程输出编码问题
如果子进程输出包含中文或者其他非ASCII字符,需要在spawn方法中设置encoding='utf-8',避免出现编码错误。如果子进程使用的是其他编码,需要对应调整该参数。
子进程异常退出
当子进程异常退出时,Pexpect会抛出pexpect.EOF异常,我们在代码中捕获这个异常后可以做对应的收尾处理,比如记录子进程退出的日志,释放相关资源等。
时间戳格式调整
如果需要调整时间戳的格式,只需要修改basicConfig中的datefmt参数即可,比如设置为'%Y/%m/%d %H:%M:%S'就可以得到斜杠分隔的日期格式。
扩展场景
如果需要将子进程的输出同时保存到文件和终端,可以在Logging配置中添加FileHandler,示例代码如下:
import pexpect
import logging
# 创建Logger对象
logger = logging.getLogger('subprocess_logger')
logger.setLevel(logging.INFO)
# 创建控制台处理器
console_handler = logging.StreamHandler()
# 创建文件处理器,将日志保存到文件
file_handler = logging.FileHandler('subprocess_output.log', encoding='utf-8')
# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S')
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
# 添加处理器到Logger
logger.addHandler(console_handler)
logger.addHandler(file_handler)
def run_subprocess_with_timestamp_extended():
child = pexpect.spawn('bash -c "while true; do echo extended_test; sleep 1; done"', encoding='utf-8')
try:
while True:
line = child.readline(timeout=2)
if not line:
break
line = line.strip()
logger.info(f"子进程输出: {line}")
except pexpect.TIMEOUT:
pass
except pexpect.EOF:
logger.info("子进程执行结束")
finally:
child.close()
if __name__ == "__main__":
run_subprocess_with_timestamp_extended()
这样子进程的输出就会同时显示在终端和subprocess_output.log文件中,并且每一行都带有时间戳,方便后续查看和分析。