如何通过nbconvert.preprocessors.ExecutePreprocessor获取执行过程中的部分输出？

要实现执行过程中实时获取输出并持续保存笔记本，你可以通过自定义预处理器继承ExecutePreprocessor，重写单元格执行的方法，在每个单元格执行完成后立即保存更新后的笔记本内容。如果要捕获进度条这类实时输出，还可以监听内核的IO消息流。

解决方案代码 #

import nbformat
from nbconvert.preprocessors import ExecutePreprocessor

class RealTimeExecutePreprocessor(ExecutePreprocessor):
    def __init__(self, save_path='executed_notebook.ipynb', *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.save_path = save_path

    def run_cell(self, cell, cell_index, resources=None):
        # 执行当前单元格
        cell, resources = super().run_cell(cell, cell_index, resources)
        
        # 执行完成后立即保存整个笔记本
        with open(self.save_path, 'w', encoding='utf-8') as f:
            nbformat.write(self.nb, f)
        
        # 可选：实时打印单元格的输出（比如进度条）
        if cell.outputs:
            for output in cell.outputs:
                if hasattr(output, 'text'):
                    print(output.text, end='')
                elif hasattr(output, 'data') and 'text/plain' in output.data:
                    print(output.data['text/plain'], end='')
        return cell, resources

# 加载笔记本
with open('./test2.ipynb') as f:
    nb = nbformat.read(f, as_version=4)

# 使用自定义预处理器执行
ep = RealTimeExecutePreprocessor(timeout=600, kernel_name='python3', save_path='executed_notebook.ipynb')
ep.preprocess(nb)

print("笔记本执行完成")

关键说明 #

• 自定义预处理器：通过继承ExecutePreprocessor并重写run_cell方法，实现每个单元格执行后的即时保存。self.nb就是当前正在执行的笔记本对象，每次单元格执行后它都会被更新。
• 实时输出捕获：代码中添加了对单元格输出的打印逻辑，能直接在控制台看到进度条这类实时输出（比如tqdm的动态更新）。
• 保存频率：当前是每个单元格执行后就保存，你可以根据需求调整，比如每5个单元格保存一次，或者添加条件判断只在特定单元格执行后保存。

注意事项 #

• 频繁写入文件可能会小幅影响执行速度，若笔记本单元格数量极多，可适当降低保存频率。
• 确保执行环境的内核依赖与笔记本要求一致，避免因缺失包导致执行中断。

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