python 从外部启动编辑器并获取文本

Question

我正在编写一个 PyQt 程序，我希望允许用户启动他们喜欢的编辑器来填写 TextEdit 字段。

所以目标是在 tmp 文件上从外部启动一个编辑器（比如 vim），并在编辑器关闭时，将其上下文放入 python 变量中。

我发现了一些类似的问题，例如Opening vi from Python、call up an EDITOR (vim) from a python script、invoke an editor ( vim ) in python。但它们都以类似于git commit 命令的“阻塞”方式工作。我追求的是一种“非阻塞”方式（因为它是一个 GUI），类似于zimwiki 中的“编辑源”功能。

我目前的尝试：

import os
import tempfile
import threading
import subprocess

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        proc = subprocess.Popen(popenArgs)
        # this immediately finishes OPENING vim.
        rec=proc.wait()
        print('# <runInThread>: rec=', rec)
        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

def openEditor():

    fd, filepath=tempfile.mkstemp()
    print('filepath=',filepath)

    def cb(tmppath):
        print('# <cb>: cb tmppath=',tmppath)
        with open(tmppath, 'r') as tmp:
            lines=tmp.readlines()
            for ii in lines:
                print('# <cb>: ii',ii)
        return

    with os.fdopen(fd, 'w') as tmp:

        cmdflag='--'
        editor_cmd='vim'
        cmd=[os.environ['TERMCMD'], cmdflag, editor_cmd, filepath]
        print('#cmd = ',cmd)

        popenAndCall(cb, cmd)
        print('done')

    return


if __name__=='__main__':

    openEditor()

我认为它失败了，因为Popen.wait() 只等到编辑器打开，而不是直到它关闭。所以它没有从编辑器中捕获任何内容。

知道如何解决这个问题吗？谢谢！

编辑：

我发现这个answer 我猜是相关的。我正在尝试让os 等待process group，但它仍然无法正常工作。代码如下：

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        proc = subprocess.Popen(popenArgs, preexec_fn=os.setsid)
        pid=proc.pid
        gid=os.getpgid(pid)
        #rec=proc.wait()
        rec=os.waitid(os.P_PGID, gid, os.WEXITED | os.WSTOPPED)
        print('# <runInThread>: rec=', rec, 'pid=',pid, 'gid=',gid)

        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

我假设这个gid=os.getpgid(pid) 给了我组的ID，而os.waitid() 等待组。我也试过os.waitpid(gid, 0)，也没用。

我在正确的轨道上？

更新：

似乎对于某些有效的编辑器，例如xed。 vim 和 gvim 都失败了。

Answer 1

使用 QProcess，您可以在不阻塞 Qt 事件循环的情况下启动进程。

在这种情况下，我使用 xterm，因为我不知道在 TERMCMD 中建立了哪个终端。

from PyQt5 import QtCore, QtGui, QtWidgets


class EditorWorker(QtCore.QObject):
    finished = QtCore.pyqtSignal()

    def __init__(self, command, parent=None):
        super(EditorWorker, self).__init__(parent)
        self._temp_file = QtCore.QTemporaryFile(self)
        self._process = QtCore.QProcess(self)
        self._process.finished.connect(self.on_finished)
        self._text = ""
        if self._temp_file.open():
            program, *arguments = command
            self._process.start(
                program, arguments + [self._temp_file.fileName()]
            )

    @QtCore.pyqtSlot()
    def on_finished(self):
        if self._temp_file.isOpen():
            self._text = self._temp_file.readAll().data().decode()
            self.finished.emit()

    @property
    def text(self):
        return self._text

    def __del__(self):
        self._process.kill()


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self._button = QtWidgets.QPushButton(
            "Launch VIM", clicked=self.on_clicked
        )
        self._text_edit = QtWidgets.QTextEdit(readOnly=True)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self._button)
        lay.addWidget(self._text_edit)

    @QtCore.pyqtSlot()
    def on_clicked(self):
        worker = EditorWorker("xterm -e vim".split(), self)
        worker.finished.connect(self.on_finished)

    @QtCore.pyqtSlot()
    def on_finished(self):
        worker = self.sender()
        prev_cursor = self._text_edit.textCursor()
        self._text_edit.moveCursor(QtGui.QTextCursor.End)
        self._text_edit.insertPlainText(worker.text)
        self._text_edit.setTextCursor(prev_cursor)
        worker.deleteLater()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

我猜你的情况应该改变

"xterm -e vim".split()

到

[os.environ['TERMCMD'], "--", "vim"]

可能的命令：

- xterm -e vim
- xfce4-terminal --disable-server -x vim

更新：

实现与 pyinotify 使用的相同逻辑，即监控文件，但在本例中使用QFileSystemWatcher，这是一个多平台解决方案：

from PyQt5 import QtCore, QtGui, QtWidgets


class EditorWorker(QtCore.QObject):
    finished = QtCore.pyqtSignal()

    def __init__(self, command, parent=None):
        super(EditorWorker, self).__init__(parent)
        self._temp_file = QtCore.QTemporaryFile(self)
        self._process = QtCore.QProcess(self)
        self._text = ""
        self._watcher = QtCore.QFileSystemWatcher(self)
        self._watcher.fileChanged.connect(self.on_fileChanged)

        if self._temp_file.open():
            self._watcher.addPath(self._temp_file.fileName())

            program, *arguments = command
            self._process.start(
                program, arguments + [self._temp_file.fileName()]
            )

    @QtCore.pyqtSlot()
    def on_fileChanged(self):
        if self._temp_file.isOpen():
            self._text = self._temp_file.readAll().data().decode()
            self.finished.emit()

    @property
    def text(self):
        return self._text

    def __del__(self):
        self._process.kill()


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self._button = QtWidgets.QPushButton(
            "Launch VIM", clicked=self.on_clicked
        )
        self._text_edit = QtWidgets.QTextEdit(readOnly=True)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self._button)
        lay.addWidget(self._text_edit)

    @QtCore.pyqtSlot()
    def on_clicked(self):
        worker = EditorWorker("gnome-terminal -- vim".split(), self)
        worker.finished.connect(self.on_finished)

    @QtCore.pyqtSlot()
    def on_finished(self):
        worker = self.sender()
        prev_cursor = self._text_edit.textCursor()
        self._text_edit.moveCursor(QtGui.QTextCursor.End)
        self._text_edit.insertPlainText(worker.text)
        self._text_edit.setTextCursor(prev_cursor)
        worker.deleteLater()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

Answer 2

我转载的问题是proc是gnome-terminal进程而不是vim进程。

这里有两个适合我的选项。

1) 查找文本编辑器的进程，而不是终端的进程。使用正确的进程 ID，代码可以等待文本编辑器的进程完成。

使用 psutil（便携式）

Finds the latest editor process in the list of all running processes.

import psutil
def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        editor_cmd=popenArgs[-2]  # vim
        proc = subprocess.Popen(popenArgs)
        proc.wait()

        # Find the latest editor process in the list of all running processes
        editor_processes = []

        for p in psutil.process_iter():
            try:
                process_name = p.name()
                if editor_cmd in process_name:
                    editor_processes.append((process_name, p.pid))
            except:
                pass

        editor_proc = psutil.Process(editor_processes[-1][1])

        rec=editor_proc.wait()
        print('# <runInThread>: rec=', rec)
        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

没有 psutil（适用于 Linux，但不能移植到 Mac OS 或 Windows）

取自 https://stackoverflow.com/a/2704947/241866 和 source code of psutil。

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        editor_cmd=popenArgs[-2]  # vim
        proc = subprocess.Popen(popenArgs)
        proc.wait()

        # Find the latest editor process in the list of all running processes

        pids = [pid for pid in os.listdir('/proc') if pid.isdigit()]

        editor_processes = []
        for pid in pids:
            try:
                process_name = open(os.path.join('/proc', pid, 'cmdline'), 'rb').read().split('\0')[0]
                if editor_cmd in process_name:
                    editor_processes.append((process_name, int(pid)))
            except IOError:
                continue
        editor_proc_pid = editor_processes[-1][1]

        def pid_exists(pid):
            try:
                os.kill(pid, 0)
                return True
            except:
                return 

        while True:
            if pid_exists(editor_proc_pid):
                import time
                time.sleep(1)
            else:
                break

        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

2) 作为最后的手段，您可以在更新文本之前捕获 UI 事件：

def popenAndCall(onExit, popenArgs):

    def runInThread(onExit, popenArgs):
        tmppath=popenArgs[-1]
        proc = subprocess.Popen(popenArgs)
        # this immediately finishes OPENING vim.
        rec=proc.wait()
        raw_input("Press Enter")  # replace this with UI event
        print('# <runInThread>: rec=', rec)
        onExit(tmppath)
        os.remove(tmppath)
        return

    thread = threading.Thread(target=runInThread, args=(onExit, popenArgs))
    thread.start()
    return thread

Answer 3

我认为@eyllanesc 的解决方案与 zim 的做法非常接近（zim 使用的是GObject.spawn_async() 和GObject.child_watch_add()，我没有使用GObject 的经验，我想这相当于QProcess.start()）。但是我们遇到了一些关于某些终端（如gnome-terminal）如何处理新终端会话启动的问题。

我试图监控编辑器打开的临时文件，并在写入/保存临时文件时调用我的回调。使用pyinotify 进行监控。我试过gnome-terminal、xterm、urxvt 和普通的gvim，似乎都有效。

代码如下：

import threading
from PyQt5 import QtCore, QtGui, QtWidgets
import pyinotify


class EditorWorker(QtCore.QObject):
    file_close_sig = QtCore.pyqtSignal()
    edit_done_sig = QtCore.pyqtSignal()

    def __init__(self, command, parent=None):
        super(EditorWorker, self).__init__(parent)
        self._temp_file = QtCore.QTemporaryFile(self)
        self._process = QtCore.QProcess(self)
        #self._process.finished.connect(self.on_file_close)
        self.file_close_sig.connect(self.on_file_close)
        self._text = ""
        if self._temp_file.open():
            program, *arguments = command
            self._process.start(
                program, arguments + [self._temp_file.fileName()]
            )
            tmpfile=self._temp_file.fileName()
            # start a thread to monitor file saving/closing
            self.monitor_thread = threading.Thread(target=self.monitorFile,
                    args=(tmpfile, self.file_close_sig))
            self.monitor_thread.start()

    @QtCore.pyqtSlot()
    def on_file_close(self):
        if self._temp_file.isOpen():
            print('open')
            self._text = self._temp_file.readAll().data().decode()
            self.edit_done_sig.emit()
        else:
            print('not open')

    @property
    def text(self):
        return self._text

    def __del__(self):
        try:
            self._process.kill()
        except:
            pass

    def monitorFile(self, path, sig):

        class PClose(pyinotify.ProcessEvent):
            def my_init(self):
                self.sig=sig
                self.done=False

            def process_IN_CLOSE(self, event):
                f = event.name and os.path.join(event.path, event.name) or event.path
                self.sig.emit()
                self.done=True

        wm = pyinotify.WatchManager()
        eventHandler=PClose()
        notifier = pyinotify.Notifier(wm, eventHandler)
        wm.add_watch(path, pyinotify.IN_CLOSE_WRITE)

        try:
            while not eventHandler.done:
                notifier.process_events()
                if notifier.check_events():
                    notifier.read_events()
        except KeyboardInterrupt:
            notifier.stop()
            return


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super(Widget, self).__init__(parent)
        self._button = QtWidgets.QPushButton(
            "Launch VIM", clicked=self.on_clicked
        )
        self._text_edit = QtWidgets.QTextEdit(readOnly=True)

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self._button)
        lay.addWidget(self._text_edit)

    @QtCore.pyqtSlot()
    def on_clicked(self):
        worker = EditorWorker(["gnome-terminal", '--', "vim"], self)
        worker.edit_done_sig.connect(self.on_edit_done)

    @QtCore.pyqtSlot()
    def on_edit_done(self):
        worker = self.sender()
        prev_cursor = self._text_edit.textCursor()
        self._text_edit.moveCursor(QtGui.QTextCursor.End)
        self._text_edit.insertPlainText(worker.text)
        self._text_edit.setTextCursor(prev_cursor)
        worker.deleteLater()


if __name__ == "__main__":
    import sys

    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.resize(640, 480)
    w.show()
    sys.exit(app.exec_())

但pyinotify 仅适用于 Linux。如果您能找到跨平台解决方案（至少在 Mac 上），请告诉我。

更新：这似乎并不可靠。 pyinotify 报告文件写入，而不仅仅是文件关闭。我很沮丧。