Python实现美化版端口进程管理工具

import sys
import psutil
import os
import pyperclip
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QTableWidget, QTableWidgetItem, QMenu, QAction, QComboBox, QLineEdit, QPushButton, QMessageBox, QHeaderView
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QIcon

# 获取本机所有开放的端口及对应的进程信息
def get_open_ports():
    open_ports = []
    for conn in psutil.net_connections(kind='inet'):
        if conn.status != 'LISTEN':
            continue
        
        pid = conn.pid
        if conn.type == 1:  # TCP协议
            protocol = 'TCP'
        elif conn.type == 2:  # UDP协议
            protocol = 'UDP'
        else:
            protocol = 'N/A'
        
        try:
            process = psutil.Process(pid)
            process_name = process.name()
            exe_path = process.exe()
        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
            process_name = "N/A"
            exe_path = "N/A"
        
        open_ports.append({
            'Port': conn.laddr.port,
            'PID': pid,
            'Process Name': process_name,
            'Protocol': protocol,
            'Path': exe_path
        })
    
    return open_ports

# 根据端口号查询对应进程的信息
def search_by_port(port):
    open_ports = get_open_ports()
    for port_info in open_ports:
        if port_info['Port'] == port:
            return port_info
    return None

# 根据进程名称查询对应的端口信息
def search_by_process_name(name):
    open_ports = get_open_ports()
    result = []
    for port_info in open_ports:
        if name.lower() in port_info['Process Name'].lower():
            result.append(port_info)
    return result

# 根据PID查询对应的端口信息
def search_by_pid(pid):
    open_ports = get_open_ports()
    for port_info in open_ports:
        if port_info['PID'] == pid:
            return port_info
    return None

# 结束进程
def kill_process(pid):
    try:
        process = psutil.Process(pid)
        process.terminate()  # 发送 terminate 信号
        process.wait()  # 等待进程结束
        return True
    except (psutil.NoSuchProcess, psutil.AccessDenied):
        return False

# 定位进程文件夹路径
def open_process_folder(exe_path):
    if exe_path and os.path.exists(exe_path):
        folder_path = os.path.dirname(exe_path)
        os.startfile(folder_path)  # 打开文件夹
        return True
    return False

# 复制到剪贴板的功能
def copy_to_clipboard(text):
    pyperclip.copy(text)  # 使用 pyperclip 库复制文本

class PortProcessManager(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("端口进程管理工具")
        self.setWindowIcon(QIcon(os.path.join(os.getcwd(), 'icon.ico')))
        self.setGeometry(100, 100, 900, 600)
        #self.setWindowOpacity(0.8)  # 设置窗口透明度为0.8
        
        # 主窗口布局
        self.main_widget = QWidget()
        self.setCentralWidget(self.main_widget)
        layout = QVBoxLayout(self.main_widget)

        # 搜索框布局
        search_layout = QHBoxLayout()
        
        self.search_type_combo = QComboBox()
        self.search_type_combo.addItems(["端口号", "进程名称", "PID"])
        self.search_type_combo.setCurrentIndex(0)
        search_layout.addWidget(self.search_type_combo)
        
        self.search_input = QLineEdit()
        search_layout.addWidget(self.search_input)
        
        self.search_button = QPushButton("查  询")
        self.search_button.clicked.connect(self.search)
        search_layout.addWidget(self.search_button)
        
        self.refresh_button = QPushButton("刷新列表")
        self.refresh_button.clicked.connect(self.refresh_list)
        search_layout.addWidget(self.refresh_button)

        layout.addLayout(search_layout)
        
        # 表格展示
        self.table = QTableWidget()
        self.table.setColumnCount(5)
        self.table.setHorizontalHeaderLabels(["PID", "协议", "端口", "进程名称", "相关路径"])
        self.table.setEditTriggers(QTableWidget.NoEditTriggers)  # 禁止编辑
        self.table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)  # 列宽自适应
        layout.addWidget(self.table)

        # 右键菜单
        self.table.setContextMenuPolicy(Qt.CustomContextMenu)
        self.table.customContextMenuRequested.connect(self.open_right_menu)

        # 设置样式
        self.set_styles()

        # 初始化数据
        self.refresh_list()

    def set_styles(self):
        # 设置字体为 "Segoe UI"
        style = """
        * {
            font-family: "Segoe UI";
        }
        QPushButton {
            background-color: #2dd8e1;
            color: white;
            border: 1px solid #00a7c3;
            padding: 5px;
        }
        QHeaderView::section {
            background-color: #2dd8e1;
            color: white;
            padding: 5px;
        }
        """
        self.setStyleSheet(style)

    def refresh_list(self):
        self.table.setRowCount(0)
        open_ports = get_open_ports()
        if not open_ports:
            QMessageBox.information(self, "没有找到端口", "没有开放的端口或无法获取端口信息。")
            return
        for port_info in open_ports:
            row_position = self.table.rowCount()
            self.table.insertRow(row_position)
            self.table.setItem(row_position, 0, QTableWidgetItem(str(port_info['PID'])))
            self.table.setItem(row_position, 1, QTableWidgetItem(port_info['Protocol']))
            self.table.setItem(row_position, 2, QTableWidgetItem(str(port_info['Port'])))
            self.table.setItem(row_position, 3, QTableWidgetItem(port_info['Process Name']))
            self.table.setItem(row_position, 4, QTableWidgetItem(port_info['Path']))

    def search(self):
        search_value = self.search_input.text()
        search_type = self.search_type_combo.currentText()
        
        if not search_value:
            QMessageBox.warning(self, "输入错误", "请输入查询内容！")
            return
        
        self.table.setRowCount(0)
        if search_type == "端口号":
            try:
                port = int(search_value)
                port_info = search_by_port(port)
                if port_info:
                    self.add_row(port_info)
                else:
                    QMessageBox.information(self, "未找到", f"未找到端口 {port} 对应的进程。")
            except ValueError:
                QMessageBox.warning(self, "输入错误", "请输入有效的端口号。")
        elif search_type == "进程名称":
            result = search_by_process_name(search_value)
            if result:
                for port_info in result:
                    self.add_row(port_info)
            else:
                QMessageBox.information(self, "未找到", f"未找到进程名称包含 {search_value} 的记录。")
        elif search_type == "PID":
            try:
                pid = int(search_value)
                port_info = search_by_pid(pid)
                if port_info:
                    self.add_row(port_info)
                else:
                    QMessageBox.information(self, "未找到", f"未找到PID {pid} 对应的进程。")
            except ValueError:
                QMessageBox.warning(self, "输入错误", "请输入有效的PID。")

    def add_row(self, port_info):
        row_position = self.table.rowCount()
        self.table.insertRow(row_position)
        self.table.setItem(row_position, 0, QTableWidgetItem(str(port_info['PID'])))
        self.table.setItem(row_position, 1, QTableWidgetItem(port_info['Protocol']))
        self.table.setItem(row_position, 2, QTableWidgetItem(str(port_info['Port'])))
        self.table.setItem(row_position, 3, QTableWidgetItem(port_info['Process Name']))
        self.table.setItem(row_position, 4, QTableWidgetItem(port_info['Path']))

    def open_right_menu(self, pos):
        selected_item = self.table.itemAt(pos)
        if selected_item:
            row = selected_item.row()
            pid = self.table.item(row, 0).text()
            port = self.table.item(row, 2).text()
            process_name = self.table.item(row, 3).text()
            exe_path = self.table.item(row, 4).text()
            
            menu = QMenu(self)
            kill_action = QAction("结束进程", self)
            kill_action.triggered.connect(lambda: self.kill_process(int(pid)))
            menu.addAction(kill_action)

            folder_action = QAction("定位进程文件夹路径", self)
            folder_action.triggered.connect(lambda: self.open_folder(exe_path))
            menu.addAction(folder_action)

            copy_pid_action = QAction("复制PID", self)
            copy_pid_action.triggered.connect(lambda: copy_to_clipboard(pid))
            menu.addAction(copy_pid_action)

            copy_port_action = QAction("复制端口号", self)
            copy_port_action.triggered.connect(lambda: copy_to_clipboard(port))
            menu.addAction(copy_port_action)

            copy_process_action = QAction("复制进程名称", self)
            copy_process_action.triggered.connect(lambda: copy_to_clipboard(process_name))
            menu.addAction(copy_process_action)

            copy_path_action = QAction("复制相关路径", self)
            copy_path_action.triggered.connect(lambda: copy_to_clipboard(exe_path))
            menu.addAction(copy_path_action)

            menu.exec_(self.table.viewport().mapToGlobal(pos))

    def kill_process(self, pid):
        if kill_process(pid):
            QMessageBox.information(self, "结束进程", f"进程 (PID: {pid}) 已被成功结束。")
        else:
            QMessageBox.warning(self, "错误", "无法结束该进程，可能没有权限。")

    def open_folder(self, exe_path):
        if open_process_folder(exe_path):
            QMessageBox.information(self, "打开文件夹", "已成功定位进程文件夹。")
        else:
            QMessageBox.warning(self, "错误", "无法找到进程文件路径。")

    def keyPressEvent(self, event):
        if event.key() == Qt.Key_Return:  # 检测到回车键
            self.search()  # 执行查询操作
        else:
            super().keyPressEvent(event)

if __name__ == "__main__":
    app = QApplication(sys.argv)
    window = PortProcessManager()
    window.show()
    sys.exit(app.exec_())