Python多管道实现yt-dlp连续直播时视频尾部数据截断问题求助

问题根因 #

你的代码出现末尾截断的核心原因是：yt-dlp输出为空时你直接终止了处理逻辑，此时转码用的ffmpeg（encoder进程）的输入还没收到EOF信号，不会主动刷出内部缓存的最后几秒数据，同时你也没读取encoder进程剩余的输出就直接销毁了进程，导致末尾数据丢失。

修正后的实现代码 #

from subprocess import Popen, PIPE, DEVNULL

COPY_BUFSIZE = 65424

playlist = [
    {
        # 15秒测试视频
        "url": "https://youtube.com/watch?v=QiInzFHIDp4"
    },
    {
        # 15秒测试视频
        "url": "https://youtube.com/watch?v=QiInzFHIDp4"
    },
    {
        # 15秒测试视频
        "url": "https://youtube.com/watch?v=QiInzFHIDp4"
    },
]

if __name__ == "__main__":
    stream_cmd = [
        "ffmpeg", "-loglevel", "error",
        "-hide_banner", "-re", "-i", "-",
        "-c:v", "libx264",
        "-f", "flv",
        "-b:v", "3000k", "-minrate", "3000k",
        "-maxrate", "3000k", "-bufsize", "3000k",
        "-r", "25", "-pix_fmt", "yuv420p",
        "rtmp://127.0.0.1/live/H1P_x5WPF"
    ]
    print(f'推流命令:\n"{" ".join(stream_cmd)}"')

    encoder_cmd = [
        "ffmpeg", "-re", "-i", "-", "-f", "mpegts",
        "-c", "copy", "-"
    ]
    print(f'转码命令:\n"{" ".join(encoder_cmd)}"')

    stream_p = Popen(stream_cmd, stdin=PIPE, stderr=DEVNULL)

    for video in playlist:
        # 对齐原bash命令的yt-dlp参数
        yt_dlp_cmd = [
            "yt-dlp", "-q", "--buffer-size", "16k",
            video["url"],
            "-o", "-"
        ]

        print("当前播放: " + video["url"])

        with Popen(yt_dlp_cmd, stdout=PIPE, stderr=DEVNULL) as yt_dlp_p:
            with Popen(encoder_cmd, stdin=PIPE, stdout=PIPE, stderr=DEVNULL) as encoder_p:
                # 第一阶段：全量传输yt-dlp的输出到转码进程
                while True:
                    yt_dlp_buf = yt_dlp_p.stdout.read(COPY_BUFSIZE)
                    if not yt_dlp_buf:
                        break
                    encoder_p.stdin.write(yt_dlp_buf)
                
                # 核心修复：关闭转码进程输入，发送EOF信号触发转码进程刷新所有缓存数据
                encoder_p.stdin.close()
                # 第二阶段：读取转码进程所有剩余输出，写入最终推流进程
                while True:
                    encoder_buf = encoder_p.stdout.read(COPY_BUFSIZE)
                    if not encoder_buf:
                        break
                    stream_p.stdin.write(encoder_buf)
                
                # 等待进程正常退出，确保所有资源释放、数据处理完成
                encoder_p.wait()
                yt_dlp_p.wait()
    
    # 所有视频处理完成后关闭推流进程输入，让其正常退出
    stream_p.stdin.close()
    stream_p.wait()

关键改动说明 #

1. 拆分处理逻辑为两个阶段：第一阶段全量传输yt-dlp的输出到转码进程，第二阶段专门清空转码进程的剩余缓存
2. 读完yt-dlp所有数据后主动关闭转码进程的stdin，给ffmpeg发送EOF信号，ffmpeg收到后会把内部缓存的所有帧全部输出
3. 新增循环读取转码进程的所有剩余输出，全部写入最终的推流进程，确保没有数据残留
4. 对齐原bash命令的yt-dlp缓冲参数，减少缓冲不一致导致的问题
5. 补充进程等待逻辑，避免出现僵尸进程，同时确保所有数据处理完成再进入下一个视频的处理

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