交互执行命令

Author: 892768447

QProcess/InteractiveRun.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+Created on 2023/02/01
+@author: Irony
+@site: https://pyqt.site https://github.com/PyQt5
+@email: [email protected]
+@file: InteractiveRun.py
+@description:
+"""
+
+import os
+import sys
+
+try:
+    import chardet
+except ImportError:
+    print('chardet not found')
+
+try:
+    from PyQt5.QtCore import QProcess
+    from PyQt5.QtWidgets import (QApplication, QLineEdit, QPushButton,
+                                 QTextBrowser, QVBoxLayout, QWidget)
+except ImportError:
+    from PySide2.QtCore import QProcess
+    from PySide2.QtWidgets import (QApplication, QLineEdit, QPushButton,
+                                   QTextBrowser, QVBoxLayout, QWidget)
+
+
+class Window(QWidget):
+
+    def __init__(self, *args, **kwargs):
+        super(Window, self).__init__(*args, **kwargs)
+        layout = QVBoxLayout(self)
+
+        command = 'ping www.baidu.com'
+        if sys.platform.startswith('win'):
+            command = 'ping -n 5 www.baidu.com'
+        elif sys.platform.startswith('darwin') or sys.platform.startswith(
+                'linux'):
+            command = 'ping -c 5 www.baidu.com'
+        else:
+            raise RuntimeError('Unsupported platform: %s' % sys.platform)
+
+        self.cmdEdit = QLineEdit(command, self)
+        layout.addWidget(self.cmdEdit)
+
+        self.buttonRun = QPushButton('执行命令', self)
+        layout.addWidget(self.buttonRun)
+        self.buttonRun.clicked.connect(self.run_command)
+
+        self.resultView = QTextBrowser(self)
+        layout.addWidget(self.resultView)
+
+        self._cmdProcess = None
+        self._init()
+
+    def closeEvent(self, event):
+        if self._cmdProcess:
+            self._cmdProcess.writeData('exit'.encode() + os.linesep.encode())
+            self._cmdProcess.waitForFinished()
+            if self._cmdProcess:
+                self._cmdProcess.terminate()
+        super(Window, self).closeEvent(event)
+
+    def _init(self):
+        if self._cmdProcess:
+            return
+        # 打开终端shell
+        self._cmdProcess = QProcess(self)
+        self._cmdProcess.setProgram(
+            'cmd' if sys.platform.startswith('win') else 'bash')
+        # 合并输出流和错误流，只从标准输出流读取数据
+        self._cmdProcess.setProcessChannelMode(QProcess.MergedChannels)
+        self._cmdProcess.started.connect(self.on_started)
+        self._cmdProcess.finished.connect(self.on_finished)
+        self._cmdProcess.errorOccurred.connect(self.on_error)
+        self._cmdProcess.readyReadStandardOutput.connect(
+            self.on_readyReadStandardOutput)
+        self._cmdProcess.start()
+
+    def run_command(self):
+        self._init()
+        command = self.cmdEdit.text().strip()
+        if not command:
+            return
+        command = command.encode(sys.getdefaultencoding()) + os.linesep.encode(
+            sys.getdefaultencoding())
+        self._cmdProcess.writeData(command)
+
+    def on_started(self):
+        self.resultView.append('ping process started, pid: %s' %
+                               self._cmdProcess.processId())
+
+    def on_finished(self, exitCode, exitStatus):
+        print('ping process finished, exitCode: %s, exitStatus: %s' %
+              (exitCode, exitStatus))
+        self._cmdProcess.kill()
+        self._cmdProcess = None
+
+    def on_error(self, error):
+        self.resultView.append('ping process error: %s, message: %s' %
+                               (error, self._cmdProcess.errorString()))
+        self._cmdProcess.kill()
+        self._cmdProcess = None
+
+    def on_readyReadStandardOutput(self):
+        # 读取已有结果
+        result = self._cmdProcess.readAllStandardOutput().data()
+        try:
+            encoding = chardet.detect(result)
+            self.resultView.append(result.decode(encoding['encoding']))
+        except Exception:
+            self.resultView.append(result.decode('utf-8', errors='ignore'))
+
+
+if __name__ == '__main__':
+    import cgitb
+
+    cgitb.enable(format='text')
+    app = QApplication(sys.argv)
+    w = Window()
+    w.show()
+    sys.exit(app.exec_())

QProcess/README.md

@@ -2,6 +2,7 @@
 
 - 目录
   - [执行命令得到结果](#1执行命令得到结果)
+  - [交互执行命令](#2交互执行命令)
 
 ## 1、执行命令得到结果
 [运行 GetCmdResult.py](GetCmdResult.py)
@@ -17,4 +18,16 @@
 2. `waitForFinished`为同步方式，然后调用`readAll`读取所有输出
 3. 也可以绑定`finished`信号，然后通过`readAll`读取所有输出
 
-![GetCmdResult](ScreenShot/GetCmdResult.gif)
+![GetCmdResult](ScreenShot/GetCmdResult.gif)
+
+## 2、交互执行命令
+[运行 InteractiveRun.py](InteractiveRun.py)
+
+`QProcess` 也可以用于交互式执行命令，具体需要如下几步：
+
+1. 通过`setProcessChannelMode(QProcess.MergedChannels)`合并标准输出和错误输出
+2. 通过`start`启动进程
+3. 通过`readyReadStandardOutput`信号读取进程输出
+4. 通过`writeData`向进程写入数据
+
+![InteractiveRun](ScreenShot/InteractiveRun.gif)

QProcess/ScreenShot/InteractiveRun.gif

@@ -242,6 +242,7 @@
     - [串口调试小助手](QSerialPort/SerialDebugAssistant.py)
   - [QProcess](QProcess)
     - [执行命令得到结果](QProcess/GetCmdResult.py)
+    - [交互执行命令](QProcess/InteractiveRun.py)
   - [QProxyStyle](QProxyStyle)
     - [Tab文字方向](QProxyStyle/TabTextDirection.py)
     - [Tab角落控件位置占满](QProxyStyle/TabCornerWidget.py)