GStreamer Python绑定：Ctrl+C时正确发送EOS事件生成可播放MP4

我可以通过以下命令成功从videotestsrc录制视频：

gst-launch-1.0 -e videotestsrc ! videoconvert ! x264enc ! mp4mux ! filesink location=output.mp4

运行该命令并按下Ctrl+C后，能得到可播放的output.mp4文件。

但用Python Gst绑定复现此行为时，无法正确处理Ctrl+C事件：

import gi
from time import sleep

gi.require_version("Gst", "1.0")
from gi.repository import Gst, GLib  # noqa


def bus_call(bus, message, loop):
    t = message.type
    if t == Gst.MessageType.EOS:
        print("EOS")
        loop.quit()
    elif t == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        loop.quit()
    return True


def main():
    Gst.init(None)
    loop = GLib.MainLoop()

    pipeline = Gst.ElementFactory.make("pipeline", None)
    src = Gst.ElementFactory.make("videotestsrc", None)
    convert = Gst.ElementFactory.make("videoconvert", None)
    encode = Gst.ElementFactory.make("x264enc")
    mux = Gst.ElementFactory.make("mp4mux")
    sink = Gst.ElementFactory.make("filesink", None)
    sink.set_property("location", "output.mp4")

    pipeline.add(src, convert, encode, mux, sink)
    src.link(convert)
    convert.link(encode)
    encode.link(mux)
    mux.link(sink)

    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", bus_call, loop)

    ret = pipeline.set_state(Gst.State.PLAYING)
    if ret == Gst.StateChangeReturn.FAILURE:
        pipeline.set_state(Gst.State.NULL)
        return

    try:
        loop.run()
    except KeyboardInterrupt:
        pipeline.send_event(Gst.Event.new_eos())
    finally:
        pipeline.set_state(Gst.State.NULL)
        loop.quit()


if __name__ == "__main__":
    main()

运行该程序并按下Ctrl+C后，生成的output.mp4无法播放。请问如何在Python代码中正确处理Ctrl+C事件？

命令行里的-e参数会在收到中断信号时，先向管道发送EOS事件，然后**等待管道完全处理完EOS（包括容器完成文件收尾）**后才退出。而你的Python代码在发送EOS后，立刻在finally块里将管道设为NULL状态并终止主循环，此时mp4mux还没来得及写入MP4文件必需的索引和文件头，导致文件损坏无法播放。

需要等待EOS事件处理完成后，再终止主循环和清理管道状态。修改逻辑如下：

1. 发送EOS后，不要立刻终止主循环，让它继续运行直到bus_call收到EOS消息再退出
2. 调整finally块的逻辑，避免提前清理管道

修改后的代码：

import gi
import signal

gi.require_version("Gst", "1.0")
from gi.repository import Gst, GLib  # noqa


def bus_call(bus, message, loop):
    t = message.type
    if t == Gst.MessageType.EOS:
        print("EOS received, terminating...")
        loop.quit()
    elif t == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print(f"Error: {err.message}, Debug: {debug}")
        loop.quit()
    return True


def handle_interrupt(sig, frame):
    global pipeline
    if pipeline:
        print("Received Ctrl+C, sending EOS...")
        pipeline.send_event(Gst.Event.new_eos())


def main():
    global pipeline
    Gst.init(None)
    loop = GLib.MainLoop()

    pipeline = Gst.ElementFactory.make("pipeline", None)
    src = Gst.ElementFactory.make("videotestsrc", None)
    convert = Gst.ElementFactory.make("videoconvert", None)
    encode = Gst.ElementFactory.make("x264enc")
    mux = Gst.ElementFactory.make("mp4mux")
    sink = Gst.ElementFactory.make("filesink", None)
    sink.set_property("location", "output.mp4")

    pipeline.add(src, convert, encode, mux, sink)
    src.link(convert)
    convert.link(encode)
    encode.link(mux)
    mux.link(sink)

    bus = pipeline.get_bus()
    bus.add_signal_watch()
    bus.connect("message", bus_call, loop)

    signal.signal(signal.SIGINT, handle_interrupt)

    ret = pipeline.set_state(Gst.State.PLAYING)
    if ret == Gst.StateChangeReturn.FAILURE:
        pipeline.set_state(Gst.State.NULL)
        return

    try:
        loop.run()
    finally:
        pipeline.set_state(Gst.State.NULL)
        print("Pipeline cleaned up")


if __name__ == "__main__":
    main()

• 使用signal模块注册SIGINT信号处理函数，替代try-except捕获KeyboardInterrupt，避免主循环被提前打断
• 发送EOS后，让主循环继续运行，直到bus_call收到EOS消息再调用loop.quit()
• finally块只在主循环退出后执行管道清理，确保mp4mux有足够时间完成文件收尾

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