commit 13bb37585f8734594f4f6d9b0f74a2ba104f4b36 Author: 王鹏 Date: Fri Jul 17 09:14:45 2026 +0800 feat: add ProcessDock process and port manager diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 0000000..c395341 --- /dev/null +++ b/.gitattributes @@ -0,0 +1,4 @@ +* text=auto eol=lf +*.ico binary +*.png binary +*.exe binary diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..49fddb7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.venv/ +dist/ +build/ +*.spec +*.log +!ProcessDock.spec diff --git a/ProcessDock.pyw b/ProcessDock.pyw new file mode 100644 index 0000000..9ec792f --- /dev/null +++ b/ProcessDock.pyw @@ -0,0 +1,5 @@ +from app import main + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/ProcessDock.spec b/ProcessDock.spec new file mode 100644 index 0000000..86e8181 --- /dev/null +++ b/ProcessDock.spec @@ -0,0 +1,85 @@ +# -*- mode: python ; coding: utf-8 -*- +from pathlib import Path + + +ROOT = Path(SPECPATH) + +a = Analysis( + [str(ROOT / "ProcessDock.pyw")], + pathex=[str(ROOT)], + binaries=[], + datas=[], + hiddenimports=[], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[ + "PySide6.Qt3DAnimation", + "PySide6.Qt3DCore", + "PySide6.Qt3DExtras", + "PySide6.Qt3DInput", + "PySide6.Qt3DLogic", + "PySide6.Qt3DRender", + "PySide6.QtBluetooth", + "PySide6.QtCharts", + "PySide6.QtDataVisualization", + "PySide6.QtGraphs", + "PySide6.QtLocation", + "PySide6.QtMultimedia", + "PySide6.QtMultimediaWidgets", + "PySide6.QtNetworkAuth", + "PySide6.QtNfc", + "PySide6.QtOpenGL", + "PySide6.QtOpenGLWidgets", + "PySide6.QtPdf", + "PySide6.QtPdfWidgets", + "PySide6.QtPositioning", + "PySide6.QtQml", + "PySide6.QtQuick", + "PySide6.QtQuick3D", + "PySide6.QtQuickControls2", + "PySide6.QtQuickWidgets", + "PySide6.QtRemoteObjects", + "PySide6.QtScxml", + "PySide6.QtSensors", + "PySide6.QtSerialBus", + "PySide6.QtSerialPort", + "PySide6.QtSpatialAudio", + "PySide6.QtSql", + "PySide6.QtStateMachine", + "PySide6.QtTest", + "PySide6.QtTextToSpeech", + "PySide6.QtWebChannel", + "PySide6.QtWebEngineCore", + "PySide6.QtWebEngineWidgets", + "PySide6.QtWebSockets", + "PySide6.QtXml", + ], + noarchive=False, + optimize=1, +) + +pyz = PYZ(a.pure) + +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.datas, + [], + name="ProcessDock", + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + upx_exclude=[], + runtime_tmpdir=None, + console=False, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, + icon=str(ROOT / "assets" / "processdock.ico"), + version=str(ROOT / "build_config" / "windows_version_info.txt"), +) diff --git a/README.md b/README.md new file mode 100644 index 0000000..38c4de4 --- /dev/null +++ b/README.md @@ -0,0 +1,87 @@ +# ProcessDock + +ProcessDock 是一个使用 Python 和 PySide6 编写的 Windows 原生进程/端口管理工具。它读取本机的真实进程与监听端口，并提供搜索、自动刷新、进程详情和安全确认后的结束操作。 + +## 功能 + +- 查看正在运行的进程，以及 PID、CPU、内存、用户和状态 +- 汇总 TCP 监听端口与 UDP 绑定端口，并支持根据端口反查进程 +- 按进程名称、PID、端口、用户、路径或启动命令即时搜索 +- 点击表格行，在右侧查看程序路径、启动参数、启动时间和父进程 PID +- 打开程序文件位置、复制启动命令、正常结束或强制结束进程 +- 每 3 秒自动刷新，也可暂停自动刷新或手动刷新 +- 后台线程采集数据，刷新时不阻塞窗口 +- 结束前二次确认，并校验进程创建时间，避免 PID 复用导致误操作 + +## 环境要求 + +- Windows 10/11 +- Python 3.10 或更高版本（已在 Python 3.11 上验证） + +建议使用虚拟环境： + +```powershell +python -m venv .venv +.\.venv\Scripts\Activate.ps1 +python -m pip install -r requirements.txt +``` + +## 启动 + +```powershell +python app.py +``` + +不需要控制台窗口时，也可以运行 `pythonw ProcessDock.pyw`；在 Python 文件关联正常的 Windows 环境中可直接双击 `ProcessDock.pyw`。 + +若不希望启动后开启自动刷新： + +```powershell +python app.py --no-auto-refresh +``` + +部分系统进程的路径、命令行、端口或结束权限受 Windows 限制。需要管理这些进程时，可右键 PowerShell 或终端并选择“以管理员身份运行”，再启动 ProcessDock。 + +## 测试 + +```powershell +python -m unittest discover -s tests -v +``` + +## 打包为 EXE + +在 PowerShell 中运行： + +```powershell +.\build_exe.ps1 +``` + +脚本会生成图标、运行测试并构建单文件 GUI 程序，输出位置为 `dist\ProcessDock.exe`。目标电脑不需要安装 Python；首次启动单文件程序时需要解压 Qt 运行库，因此会比后续启动稍慢。 + +无窗口界面冒烟测试可以这样运行： + +```powershell +$env:QT_QPA_PLATFORM="offscreen" +python -m unittest tests.test_gui -v +``` + +## 行为说明 + +- 端口列表聚焦于服务端口：TCP 仅显示 `LISTEN`，UDP 显示已绑定的本地端口，不展示大量临时的出站连接端口。 +- CPU 使用率按所有逻辑核心归一化。Windows 使用批量性能计数器；非 Windows 兼容模式的第一次刷新用于建立采样基线。 +- “正常结束”在 Windows 上调用不带 `/F` 的 `taskkill`；“强制结束”直接终止进程。强制结束可能丢失未保存的数据。 +- ProcessDock 会保护自己的进程，防止从界面中结束自身。 +- 本工具只操作当前计算机，不连接远程服务、不上传进程数据，也不使用数据库。 + +## 项目结构 + +```text +app.py 程序入口 +ProcessDock.pyw 无控制台窗口入口 +processdock/services.py 进程采集、端口映射和系统操作 +processdock/models.py 表格模型、筛选和操作按钮代理 +processdock/ui.py PySide6 主窗口与详情面板 +processdock/styles.py GUI 主题样式 +tests/ 自动化测试 +docs/blueprint.md 需求证据与验收范围 +``` diff --git a/app.py b/app.py new file mode 100644 index 0000000..6f70fcf --- /dev/null +++ b/app.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +import argparse +import sys + +from PySide6.QtCore import Qt +from PySide6.QtGui import QFont +from PySide6.QtWidgets import QApplication + +from processdock.services import ProcessService +from processdock.ui import MainWindow + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="ProcessDock 进程与端口管理器") + parser.add_argument("--no-auto-refresh", action="store_true", help="启动时关闭自动刷新") + return parser + + +def main() -> int: + arguments = build_parser().parse_args() + QApplication.setHighDpiScaleFactorRoundingPolicy( + Qt.HighDpiScaleFactorRoundingPolicy.PassThrough + ) + application = QApplication(sys.argv) + application.setApplicationName("ProcessDock") + application.setOrganizationName("ProcessDock") + application.setStyle("Fusion") + application.setFont(QFont("Microsoft YaHei UI", 10)) + + window = MainWindow( + service=ProcessService(), auto_refresh=not arguments.no_auto_refresh + ) + window.show() + return application.exec() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/artifacts/processdock-preview.png b/artifacts/processdock-preview.png new file mode 100644 index 0000000..bbc39d0 Binary files /dev/null and b/artifacts/processdock-preview.png differ diff --git a/assets/processdock.ico b/assets/processdock.ico new file mode 100644 index 0000000..f75ceb7 Binary files /dev/null and b/assets/processdock.ico differ diff --git a/assets/processdock.png b/assets/processdock.png new file mode 100644 index 0000000..bd4fb1b Binary files /dev/null and b/assets/processdock.png differ diff --git a/build_config/windows_version_info.txt b/build_config/windows_version_info.txt new file mode 100644 index 0000000..dac9125 --- /dev/null +++ b/build_config/windows_version_info.txt @@ -0,0 +1,30 @@ +VSVersionInfo( + ffi=FixedFileInfo( + filevers=(1, 0, 0, 0), + prodvers=(1, 0, 0, 0), + mask=0x3f, + flags=0x0, + OS=0x40004, + fileType=0x1, + subtype=0x0, + date=(0, 0) + ), + kids=[ + StringFileInfo([ + StringTable( + u'080404B0', + [ + StringStruct(u'CompanyName', u'ProcessDock'), + StringStruct(u'FileDescription', u'ProcessDock 进程与端口管理器'), + StringStruct(u'FileVersion', u'1.0.0'), + StringStruct(u'InternalName', u'ProcessDock'), + StringStruct(u'LegalCopyright', u'Copyright © 2026 ProcessDock'), + StringStruct(u'OriginalFilename', u'ProcessDock.exe'), + StringStruct(u'ProductName', u'ProcessDock'), + StringStruct(u'ProductVersion', u'1.0.0') + ] + ) + ]), + VarFileInfo([VarStruct(u'Translation', [2052, 1200])]) + ] +) diff --git a/build_exe.ps1 b/build_exe.ps1 new file mode 100644 index 0000000..cd19a88 --- /dev/null +++ b/build_exe.ps1 @@ -0,0 +1,28 @@ +$ErrorActionPreference = "Stop" +Set-Location -LiteralPath $PSScriptRoot + +Write-Host "[1/4] Installing build dependencies..." +python -m pip install -r requirements-build.txt +if ($LASTEXITCODE -ne 0) { throw "Dependency installation failed." } + +Write-Host "[2/4] Generating application icon..." +python tools\generate_icon.py +if ($LASTEXITCODE -ne 0) { throw "Icon generation failed." } + +Write-Host "[3/4] Running tests..." +python -m unittest discover -s tests -v +if ($LASTEXITCODE -ne 0) { throw "Tests failed." } + +Write-Host "[4/4] Building single-file executable..." +python -m PyInstaller --noconfirm --clean ProcessDock.spec +if ($LASTEXITCODE -ne 0) { throw "PyInstaller build failed." } + +$exe = Join-Path $PSScriptRoot "dist\ProcessDock.exe" +if (-not (Test-Path -LiteralPath $exe)) { throw "Build completed without an executable." } +$file = Get-Item -LiteralPath $exe +$hash = Get-FileHash -LiteralPath $exe -Algorithm SHA256 + +Write-Host "" +Write-Host "Build succeeded: $($file.FullName)" +Write-Host "Size: $([math]::Round($file.Length / 1MB, 1)) MB" +Write-Host "SHA256: $($hash.Hash)" diff --git a/docs/blueprint.md b/docs/blueprint.md new file mode 100644 index 0000000..6f5a435 --- /dev/null +++ b/docs/blueprint.md @@ -0,0 +1,53 @@ +# ProcessDock 实现蓝图 + +## 需求证据 + +| 条目 | 分类 | 依据 | 实现决定 | +|---|---|---|---| +| 进程、端口、CPU、内存列表 | requested | 用户给出的表格字段 | 使用 `psutil` 读取本机实时数据 | +| 搜索、刷新、自动刷新 | requested | 用户给出的页头控件 | 本地代理筛选；3 秒后台刷新 | +| 点击行显示完整详情 | requested | 用户给出的详情字段 | 使用主窗口右侧详情面板 | +| 打开位置、复制命令、两种结束方式 | requested | 用户给出的详情操作 | 真实调用资源管理器、剪贴板和系统进程 API | +| 原生 GUI | requested | 用户补充“用 GUI 实现” | PySide6 Qt Widgets，不启动 Web 服务 | +| TCP 只显示监听连接 | inferred | “监听端口”及端口管理工具的核心用途 | 排除临时出站 TCP 端口，UDP 保留绑定端口 | +| 当前应用自保护与 PID 复用校验 | inferred | 降低危险操作误用风险 | 禁止结束自身；操作前核对创建时间 | + +未提供外部截图、源码或演示站点，因此不存在无法访问的外部材料，也不声明对第三方产品的像素级复刻。 + +## 页面与操作矩阵 + +| 区域 | 读取 | 可见操作 | 空/错状态 | +|---|---|---|---| +| 顶部工具栏 | 当前筛选与刷新状态 | 搜索、手动刷新、自动刷新开关 | 刷新错误显示警告条 | +| 统计卡片 | 进程数、端口数、系统 CPU/内存 | 无 | 首次加载显示占位符 | +| 进程表格 | 真实进程和监听端口快照 | 排序、选中、正常结束 | 无匹配结果页、权限警告 | +| 右侧详情 | 选中进程完整信息 | 打开位置、复制命令、正常/强制结束 | 未选择、进程已结束、权限受限 | + +## 领域对象与状态 + +应用没有数据库。核心对象是只读的 `ProcessSnapshot`，以 `(PID, create_time)` 标识一个进程实例。结束操作的状态转换为： + +```text +运行中 --正常结束请求--> 已结束 + \--请求后仍运行--> 运行中 --强制结束--> 已结束 +``` + +操作前重新读取 `create_time`。如果 PID 已被复用，则拒绝操作并要求刷新。 + +## 验收场景 + +1. 启动应用后，后台读取本机进程，窗口保持可交互。 +2. 输入进程名、PID 或已知监听端口，列表立即缩小到匹配项。 +3. 点击任意一行，右侧显示与该 PID 对应的路径、命令和资源数据。 +4. 点击复制命令后，系统剪贴板获得完整启动命令。 +5. 点击正常结束或强制结束，先显示确认框；确认后执行真实系统操作并刷新列表。 +6. 对 ProcessDock 自身，结束按钮禁用；对已结束或 PID 已复用的进程，操作被拒绝并显示明确错误。 + +## 交付范围 + +- 目标系统：Windows 10/11 +- 运行时：Python 3.10+ +- 依赖：PySide6、psutil +- 存储：无 +- 网络服务：无 +- 已知限制：受当前 Windows 用户权限约束；非 Windows 兼容模式的 CPU 首帧是采样基线；正常结束的温和程度取决于目标进程与 Windows 行为。 diff --git a/processdock/__init__.py b/processdock/__init__.py new file mode 100644 index 0000000..b3b2d89 --- /dev/null +++ b/processdock/__init__.py @@ -0,0 +1,3 @@ +"""ProcessDock - a local process and port manager.""" + +__version__ = "1.0.0" diff --git a/processdock/models.py b/processdock/models.py new file mode 100644 index 0000000..22315f4 --- /dev/null +++ b/processdock/models.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +from typing import Any + +from PySide6.QtCore import QAbstractTableModel, QModelIndex, QRect, QSortFilterProxyModel, Qt, Signal +from PySide6.QtGui import QColor, QFont, QMouseEvent, QPainter +from PySide6.QtWidgets import QStyledItemDelegate, QStyleOptionViewItem + +from processdock.services import matches_process + + +PROCESS_ROLE = Qt.ItemDataRole.UserRole + 1 +SORT_ROLE = Qt.ItemDataRole.UserRole + 2 + + +class ProcessTableModel(QAbstractTableModel): + columns = ("PID", "进程名称", "监听端口", "CPU", "内存", "用户", "状态", "操作") + + def __init__(self, parent=None) -> None: + super().__init__(parent) + self._processes: list[dict[str, Any]] = [] + + def set_processes(self, processes: list[dict[str, Any]]) -> None: + self.beginResetModel() + self._processes = list(processes) + self.endResetModel() + + @property + def processes(self) -> list[dict[str, Any]]: + return self._processes + + def rowCount(self, parent=QModelIndex()) -> int: # noqa: N802 + return 0 if parent.isValid() else len(self._processes) + + def columnCount(self, parent=QModelIndex()) -> int: # noqa: N802 + return 0 if parent.isValid() else len(self.columns) + + def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole): # noqa: N802 + if role == Qt.ItemDataRole.DisplayRole and orientation == Qt.Orientation.Horizontal: + return self.columns[section] + return None + + def data(self, index: QModelIndex, role=Qt.ItemDataRole.DisplayRole): + if not index.isValid() or not 0 <= index.row() < len(self._processes): + return None + process = self._processes[index.row()] + column = index.column() + + if role == PROCESS_ROLE: + return process + if role == SORT_ROLE: + values = ( + process["pid"], + process["name"].casefold(), + process["ports"][0] if process["ports"] else -1, + process["cpu_percent"], + process["memory_bytes"], + process["username"].casefold(), + process["status_label"], + process["pid"], + ) + return values[column] + if role == Qt.ItemDataRole.DisplayRole: + ports = ", ".join(str(port) for port in process["ports"]) or "—" + values = ( + str(process["pid"]), + process["name"], + ports, + f'{process["cpu_percent"]:.1f}%', + process["memory_text"], + process["username"], + process["status_label"], + "当前应用" if process["is_current"] else "结束", + ) + return values[column] + if role == Qt.ItemDataRole.TextAlignmentRole: + if column in (0, 2, 3, 4, 6, 7): + return Qt.AlignmentFlag.AlignCenter + return Qt.AlignmentFlag.AlignVCenter | Qt.AlignmentFlag.AlignLeft + if role == Qt.ItemDataRole.ForegroundRole: + if column == 6: + if process["status"] == "running": + return QColor("#129B72") + if process["status"] in {"stopped", "suspended"}: + return QColor("#D88918") + if column == 2 and process["ports"]: + return QColor("#5B5BD6") + if column == 7: + return QColor("#A73A48" if process["can_terminate"] else "#9AA3B2") + if role == Qt.ItemDataRole.FontRole: + font = QFont() + if column in (0, 2, 3, 4): + font.setFamily("Cascadia Mono") + font.setPointSize(9) + if column == 1: + font.setWeight(QFont.Weight.DemiBold) + return font + if role == Qt.ItemDataRole.ToolTipRole: + if column == 2 and process["port_details"]: + return "\n".join( + f'{item["protocol"]} {item["address"]}:{item["port"]}' + for item in process["port_details"] + ) + if column == 1: + return process["exe"] or process["name"] + if column == 7 and process["is_current"]: + return "ProcessDock 不允许结束自身" + return None + + +class ProcessFilterProxyModel(QSortFilterProxyModel): + def __init__(self, parent=None) -> None: + super().__init__(parent) + self._query = "" + self.setDynamicSortFilter(True) + self.setSortRole(SORT_ROLE) + + def set_query(self, query: str) -> None: + self.beginFilterChange() + self._query = query + self.endFilterChange(QSortFilterProxyModel.Direction.Rows) + + def filterAcceptsRow(self, source_row: int, source_parent: QModelIndex) -> bool: # noqa: N802 + model = self.sourceModel() + index = model.index(source_row, 0, source_parent) + process = model.data(index, PROCESS_ROLE) + return bool(process and matches_process(process, self._query)) + + def lessThan(self, left: QModelIndex, right: QModelIndex) -> bool: # noqa: N802 + left_value = self.sourceModel().data(left, SORT_ROLE) + right_value = self.sourceModel().data(right, SORT_ROLE) + return left_value < right_value + + +class ActionButtonDelegate(QStyledItemDelegate): + terminate_requested = Signal(object) + + def paint(self, painter: QPainter, option: QStyleOptionViewItem, index: QModelIndex) -> None: + process = index.data(PROCESS_ROLE) + if not process: + return super().paint(painter, option, index) + + painter.save() + selected = bool(option.state & option.state.State_Selected) + if selected: + painter.fillRect(option.rect, QColor("#EEF0FF")) + + rect = self._button_rect(option.rect) + enabled = bool(process["can_terminate"]) + fill = QColor("#FFF0F1") if enabled else QColor("#F2F4F7") + text_color = QColor("#B53C4A") if enabled else QColor("#9AA3B2") + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + painter.setPen(Qt.PenStyle.NoPen) + painter.setBrush(fill) + painter.drawRoundedRect(rect, 7, 7) + painter.setPen(text_color) + font = painter.font() + font.setPointSize(9) + font.setWeight(QFont.Weight.DemiBold) + painter.setFont(font) + text = "结束" if enabled else "当前应用" + painter.drawText(rect, Qt.AlignmentFlag.AlignCenter, text) + painter.restore() + + def editorEvent(self, event, model, option, index) -> bool: # noqa: N802 + process = index.data(PROCESS_ROLE) + if ( + process + and process["can_terminate"] + and isinstance(event, QMouseEvent) + and event.type() == QMouseEvent.Type.MouseButtonRelease + and event.button() == Qt.MouseButton.LeftButton + and self._button_rect(option.rect).contains(event.position().toPoint()) + ): + self.terminate_requested.emit(process) + return True + return super().editorEvent(event, model, option, index) + + @staticmethod + def _button_rect(cell: QRect) -> QRect: + width, height = 58, 30 + return QRect( + cell.center().x() - width // 2, + cell.center().y() - height // 2, + width, + height, + ) diff --git a/processdock/services.py b/processdock/services.py new file mode 100644 index 0000000..6025853 --- /dev/null +++ b/processdock/services.py @@ -0,0 +1,499 @@ +from __future__ import annotations + +import os +import json +import socket +import subprocess +import sys +import threading +import time +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any + +import psutil + + +STATUS_LABELS = { + # psutil exposes different status constants on each operating system, while + # the returned values are stable strings. Literal keys keep imports portable. + "running": "运行中", + "sleeping": "休眠", + "disk-sleep": "等待磁盘", + "stopped": "已停止", + "tracing-stop": "调试暂停", + "zombie": "僵尸进程", + "dead": "已结束", + "wake-kill": "唤醒终止", + "waking": "唤醒中", + "parked": "已驻留", + "idle": "空闲", + "locked": "已锁定", + "waiting": "等待中", + "suspended": "已挂起", +} + + +class ProcessDockError(RuntimeError): + """Base error with a user-facing message.""" + + +class ProcessGoneError(ProcessDockError): + pass + + +class ProcessAccessError(ProcessDockError): + pass + + +class ProtectedProcessError(ProcessDockError): + pass + + +class ProcessChangedError(ProcessDockError): + pass + + +@dataclass(frozen=True) +class ActionResult: + success: bool + message: str + still_running: bool = False + + +def format_command(arguments: list[str] | None) -> str: + if not arguments: + return "" + if os.name == "nt": + return subprocess.list2cmdline(arguments) + return " ".join(_shell_quote(argument) for argument in arguments) + + +def _shell_quote(value: str) -> str: + if not value: + return "''" + if all(character.isalnum() or character in "@%_-+=:,./" for character in value): + return value + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def humanize_bytes(size: int | float) -> str: + value = float(max(size, 0)) + for unit in ("B", "KB", "MB", "GB", "TB"): + if value < 1024 or unit == "TB": + if unit == "B": + return f"{value:.0f} {unit}" + return f"{value:.1f} {unit}" + value /= 1024 + return f"{value:.1f} TB" + + +def matches_process(process: dict[str, Any], query: str) -> bool: + normalized = query.strip().casefold() + if not normalized: + return True + values = [ + str(process.get("pid", "")), + str(process.get("name", "")), + str(process.get("username", "")), + str(process.get("exe", "")), + str(process.get("command", "")), + ",".join(str(port) for port in process.get("ports", [])), + ] + return any(normalized in value.casefold() for value in values) + + +class ProcessService: + """Reads process/port data and performs guarded local process actions.""" + + def __init__(self) -> None: + self.current_pid = os.getpid() + self._cpu_count = max(psutil.cpu_count(logical=True) or 1, 1) + self._cpu_samples: dict[int, tuple[float, float, float]] = {} + self._last_cpu: dict[int, float] = {} + self._lock = threading.RLock() + + def snapshot(self) -> dict[str, Any]: + port_map, port_warning = self._collect_ports() + warnings = [port_warning] if port_warning else [] + if os.name == "nt": + try: + processes = self._collect_windows_processes(port_map) + except (OSError, subprocess.SubprocessError, json.JSONDecodeError) as error: + warnings.append(f"快速采集不可用，已切换兼容模式：{error}") + processes = self._collect_psutil_processes(port_map) + else: + processes = self._collect_psutil_processes(port_map) + + processes.sort(key=lambda item: (item["name"].casefold(), item["pid"])) + unique_ports = { + (detail["protocol"], detail["address"], detail["port"]) + for process in processes + for detail in process["port_details"] + } + memory = psutil.virtual_memory() + summary = { + "process_count": len(processes), + "port_count": len(unique_ports), + "cpu_percent": round(float(psutil.cpu_percent(interval=None)), 1), + "memory_percent": round(float(memory.percent), 1), + "memory_used": humanize_bytes(memory.used), + "memory_total": humanize_bytes(memory.total), + } + return { + "processes": processes, + "summary": summary, + "warnings": warnings, + "refreshed_at": datetime.now().strftime("%H:%M:%S"), + } + + def _collect_psutil_processes( + self, port_map: dict[int, list[dict[str, Any]]] + ) -> list[dict[str, Any]]: + now = time.monotonic() + processes: list[dict[str, Any]] = [] + live_pids: set[int] = set() + for process in psutil.process_iter(): + try: + row = self._read_process(process, port_map.get(process.pid, []), now) + except (psutil.NoSuchProcess, psutil.ZombieProcess): + continue + except Exception: + # A single unusual system process should never break the full refresh. + continue + processes.append(row) + live_pids.add(row["pid"]) + + with self._lock: + self._cpu_samples = { + pid: sample for pid, sample in self._cpu_samples.items() if pid in live_pids + } + self._last_cpu = { + pid: value for pid, value in self._last_cpu.items() if pid in live_pids + } + return processes + + def _collect_windows_processes( + self, port_map: dict[int, list[dict[str, Any]]] + ) -> list[dict[str, Any]]: + """Use Windows CIM's bulk providers to avoid hundreds of slow per-PID calls.""" + script = r""" +[Console]::OutputEncoding = [System.Text.Encoding]::UTF8 +$processes = Get-CimInstance Win32_Process -ErrorAction Stop | ForEach-Object { + $created = 0 + if ($_.CreationDate) { + $created = [DateTimeOffset]$_.CreationDate + $created = $created.ToUnixTimeMilliseconds() / 1000 + } + [PSCustomObject]@{ + pid = [int]$_.ProcessId + ppid = [int]$_.ParentProcessId + name = [string]$_.Name + exe = [string]$_.ExecutablePath + command = [string]$_.CommandLine + created = [double]$created + } +} +$performance = Get-CimInstance Win32_PerfFormattedData_PerfProc_Process -ErrorAction Stop | + Where-Object { $_.Name -ne '_Total' } | + ForEach-Object { + [PSCustomObject]@{ + pid = [int]$_.IDProcess + cpu = [double]$_.PercentProcessorTime + memory = [int64]$_.WorkingSetPrivate + } + } +[PSCustomObject]@{ processes = @($processes); performance = @($performance) } | + ConvertTo-Json -Depth 4 -Compress +""" + completed = subprocess.run( + ["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", script], + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=8, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + if completed.returncode != 0: + detail = (completed.stderr or completed.stdout).strip() + raise OSError(detail or "PowerShell CIM 查询失败") + payload = json.loads(completed.stdout.lstrip("\ufeff")) + static_by_pid = { + int(item["pid"]): item + for item in self._ensure_list(payload.get("processes")) + if item.get("pid") is not None + } + performance_by_pid = { + int(item["pid"]): item + for item in self._ensure_list(payload.get("performance")) + if item.get("pid") is not None + } + total_memory = max(int(psutil.virtual_memory().total), 1) + processes: list[dict[str, Any]] = [] + + for pid, values in static_by_pid.items(): + if pid == 0: + continue + performance = performance_by_pid.get(pid, {}) + memory_bytes = max(int(performance.get("memory") or 0), 0) + raw_cpu = max(float(performance.get("cpu") or 0.0), 0.0) + cpu_percent = round(min(raw_cpu / self._cpu_count, 100.0), 1) + try: + username = psutil.Process(pid).username() or "—" + except (psutil.Error, OSError): + username = "—" + port_details = port_map.get(pid, []) + ports = sorted({detail["port"] for detail in port_details}) + executable = values.get("exe") or "" + command = values.get("command") or "" + create_time = float(values.get("created") or 0.0) + processes.append( + { + "pid": pid, + "name": values.get("name") or f"PID {pid}", + "ports": ports, + "port_details": port_details, + "cpu_percent": cpu_percent, + "memory_bytes": memory_bytes, + "memory_text": humanize_bytes(memory_bytes), + "memory_percent": round(memory_bytes / total_memory * 100, 1), + "username": username, + "status": "running", + "status_label": "运行中", + "exe": executable, + "command": command, + "command_parts": [], + "create_time": create_time, + "create_time_text": self._format_timestamp(create_time), + "ppid": int(values.get("ppid") or 0), + "access_limited": not executable and not command and username == "—", + "is_current": pid == self.current_pid, + "can_terminate": pid != self.current_pid, + "can_open_location": bool(executable), + } + ) + return processes + + @staticmethod + def _ensure_list(value: Any) -> list[Any]: + if value is None: + return [] + return value if isinstance(value, list) else [value] + + def _collect_ports(self) -> tuple[dict[int, list[dict[str, Any]]], str | None]: + ports: dict[int, set[tuple[str, str, int]]] = defaultdict(set) + try: + connections = psutil.net_connections(kind="inet") + except psutil.AccessDenied: + return {}, "权限不足：无法读取完整端口列表，建议以管理员身份运行。" + except OSError as error: + return {}, f"端口信息暂时不可用：{error}" + + for connection in connections: + if connection.pid is None or not connection.laddr: + continue + protocol = "TCP" if connection.type == socket.SOCK_STREAM else "UDP" + if protocol == "TCP" and connection.status != psutil.CONN_LISTEN: + continue + try: + address = connection.laddr.ip + port = int(connection.laddr.port) + except AttributeError: + address, port = connection.laddr[:2] + port = int(port) + if port > 0: + ports[connection.pid].add((protocol, str(address), port)) + + result = { + pid: [ + {"protocol": protocol, "address": address, "port": port} + for protocol, address, port in sorted( + items, key=lambda item: (item[2], item[0], item[1]) + ) + ] + for pid, items in ports.items() + } + return result, None + + def _read_process( + self, + process: psutil.Process, + port_details: list[dict[str, Any]], + sample_time: float, + ) -> dict[str, Any]: + values = process.as_dict( + attrs=[ + "pid", + "name", + "username", + "status", + "exe", + "cmdline", + "create_time", + "ppid", + "memory_info", + "memory_percent", + "cpu_times", + ], + ad_value=None, + ) + pid = int(values["pid"]) + create_time = float(values.get("create_time") or 0.0) + cpu_percent = self._calculate_cpu( + pid, values.get("cpu_times"), create_time, sample_time + ) + memory_info = values.get("memory_info") + memory_bytes = int(getattr(memory_info, "rss", 0) or 0) + command_parts = values.get("cmdline") or [] + status = str(values.get("status") or "unknown") + ports = sorted({detail["port"] for detail in port_details}) + exe = values.get("exe") or "" + username = values.get("username") or "—" + limited = not exe and not command_parts and username == "—" + + return { + "pid": pid, + "name": values.get("name") or f"PID {pid}", + "ports": ports, + "port_details": port_details, + "cpu_percent": cpu_percent, + "memory_bytes": memory_bytes, + "memory_text": humanize_bytes(memory_bytes), + "memory_percent": round(float(values.get("memory_percent") or 0.0), 1), + "username": username, + "status": status, + "status_label": STATUS_LABELS.get(status, "未知"), + "exe": exe, + "command": format_command(command_parts), + "command_parts": command_parts, + "create_time": create_time, + "create_time_text": self._format_timestamp(create_time), + "ppid": int(values.get("ppid") or 0), + "access_limited": limited, + "is_current": pid == self.current_pid, + "can_terminate": pid != self.current_pid, + "can_open_location": bool(exe), + } + + def _calculate_cpu( + self, + pid: int, + cpu_times: Any, + create_time: float, + sample_time: float, + ) -> float: + if cpu_times is None: + return 0.0 + total = float(getattr(cpu_times, "user", 0.0)) + float( + getattr(cpu_times, "system", 0.0) + ) + with self._lock: + previous = self._cpu_samples.get(pid) + result = self._last_cpu.get(pid, 0.0) + if previous is not None: + previous_time, previous_total, previous_created = previous + elapsed = sample_time - previous_time + if elapsed >= 0.2 and abs(previous_created - create_time) < 0.01: + used = max(total - previous_total, 0.0) + result = min(max(used / elapsed / self._cpu_count * 100, 0.0), 100.0) + result = round(result, 1) + self._cpu_samples[pid] = (sample_time, total, create_time) + self._last_cpu[pid] = result + return result + + @staticmethod + def _format_timestamp(timestamp: float) -> str: + if timestamp <= 0: + return "—" + try: + return datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S") + except (OSError, OverflowError, ValueError): + return "—" + + def terminate( + self, pid: int, *, force: bool, expected_create_time: float | None = None + ) -> ActionResult: + if pid == self.current_pid: + raise ProtectedProcessError("ProcessDock 正在使用该进程，不能结束自身。") + try: + process = psutil.Process(pid) + actual_create_time = process.create_time() + if ( + expected_create_time is not None + and expected_create_time > 0 + and abs(actual_create_time - expected_create_time) >= 0.01 + ): + raise ProcessChangedError("PID 已被新的进程复用，请刷新后重试。") + name = process.name() + + if force: + process.kill() + elif os.name == "nt": + completed = subprocess.run( + ["taskkill", "/PID", str(pid)], + capture_output=True, + text=True, + timeout=5, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + # Console/background processes commonly reject taskkill without /F. + # Treat that as "still running" below so the GUI can offer force, + # rather than presenting a misleading hard failure. + else: + process.terminate() + + try: + process.wait(timeout=3) + except psutil.TimeoutExpired: + return ActionResult( + False, + f"{name} 尚未退出，可以尝试强制结束。", + still_running=True, + ) + action = "强制结束" if force else "正常结束" + return ActionResult(True, f"已{action} {name}（PID {pid}）。") + except ProtectedProcessError: + raise + except ProcessChangedError: + raise + except psutil.NoSuchProcess as error: + raise ProcessGoneError("进程已经结束。") from error + except psutil.AccessDenied as error: + raise ProcessAccessError("权限不足，请尝试以管理员身份运行 ProcessDock。") from error + except subprocess.TimeoutExpired as error: + raise ProcessDockError("系统结束命令超时，请稍后重试。") from error + + def open_file_location( + self, pid: int, *, expected_create_time: float | None = None + ) -> ActionResult: + try: + process = psutil.Process(pid) + actual_create_time = process.create_time() + if ( + expected_create_time is not None + and expected_create_time > 0 + and abs(actual_create_time - expected_create_time) >= 0.01 + ): + raise ProcessChangedError("PID 已被新的进程复用，请刷新后重试。") + executable = process.exe() + if not executable: + raise ProcessDockError("该进程没有可访问的程序路径。") + path = Path(executable) + if sys.platform == "win32": + subprocess.Popen(["explorer.exe", f"/select,{path}"]) + elif sys.platform == "darwin": + subprocess.Popen(["open", "-R", str(path)]) + else: + subprocess.Popen(["xdg-open", str(path.parent)]) + return ActionResult(True, "已打开程序所在位置。") + except ProcessChangedError: + raise + except psutil.NoSuchProcess as error: + raise ProcessGoneError("进程已经结束。") from error + except psutil.AccessDenied as error: + raise ProcessAccessError("权限不足，无法读取程序路径。") from error + except OSError as error: + raise ProcessDockError(f"无法打开文件位置：{error}") from error diff --git a/processdock/styles.py b/processdock/styles.py new file mode 100644 index 0000000..2ee8dff --- /dev/null +++ b/processdock/styles.py @@ -0,0 +1,198 @@ +APP_STYLE = r""" +* { + font-family: "Microsoft YaHei UI"; + color: #202534; +} +QMainWindow, QWidget#AppRoot { + background: #F4F6FA; +} +QFrame#HeaderCard, QFrame#TableCard, QFrame#DetailCard, QFrame#SummaryCard { + background: #FFFFFF; + border: 1px solid #E6E9F0; + border-radius: 14px; +} +QLabel#BrandMark { + color: white; + background: qlineargradient(x1:0, y1:0, x2:1, y2:1, stop:0 #6868E8, stop:1 #4C4CC1); + border-radius: 11px; + font-size: 20px; + font-weight: 700; +} +QLabel#BrandTitle { + color: #171B2A; + font-size: 19px; + font-weight: 700; +} +QLabel#BrandSubtitle, QLabel#MutedLabel, QLabel#CardSubtitle, QLabel#DetailKey { + color: #80899A; +} +QLabel#BrandSubtitle { font-size: 10px; } +QLineEdit#SearchBox { + min-height: 38px; + min-width: 260px; + padding: 0 14px; + background: #F7F8FB; + border: 1px solid #DFE3EB; + border-radius: 9px; + selection-background-color: #6464D9; +} +QLineEdit#SearchBox:focus { + background: #FFFFFF; + border: 1px solid #6262D6; +} +QPushButton { + min-height: 36px; + padding: 0 16px; + border-radius: 9px; + font-weight: 600; +} +QPushButton#PrimaryButton { + color: white; + background: #5B5BD6; + border: 1px solid #5B5BD6; +} +QPushButton#PrimaryButton:hover { background: #4F4FC5; } +QPushButton#SecondaryButton { + background: #FFFFFF; + border: 1px solid #DDE1EA; +} +QPushButton#SecondaryButton:hover { background: #F6F7FB; border-color: #C9CEDA; } +QPushButton#DangerButton { + color: white; + background: #C64B59; + border: 1px solid #C64B59; +} +QPushButton#DangerButton:hover { background: #B33D4B; } +QPushButton#SoftDangerButton { + color: #B53C4A; + background: #FFF1F2; + border: 1px solid #F3CDD2; +} +QPushButton#SoftDangerButton:hover { background: #FDE5E8; } +QPushButton:disabled { + color: #A6ADBA; + background: #F2F4F7; + border-color: #E4E7EC; +} +QCheckBox { spacing: 8px; color: #596273; } +QLabel#CardIcon { + min-width: 38px; max-width: 38px; + min-height: 38px; max-height: 38px; + color: #5858CD; + background: #EEEEFF; + border-radius: 10px; + font-size: 15px; + font-weight: 700; +} +QLabel#CardValue { + color: #1C2130; + font-size: 22px; + font-weight: 700; +} +QLabel#CardTitle { color: #687183; font-size: 11px; } +QLabel#WarningBanner { + color: #8B5A10; + background: #FFF7E5; + border: 1px solid #F2D79A; + border-radius: 9px; + padding: 9px 12px; +} +QLabel#SectionTitle { + color: #202534; + font-size: 15px; + font-weight: 700; +} +QTableView { + background: #FFFFFF; + alternate-background-color: #FBFCFE; + border: none; + gridline-color: transparent; + selection-background-color: #EEF0FF; + selection-color: #202534; + outline: none; +} +QTableView::item { + border-bottom: 1px solid #EEF0F4; + padding: 0 10px; +} +QTableView::item:hover { background: #F7F8FF; } +QHeaderView::section { + color: #70798A; + background: #F7F8FB; + border: none; + border-bottom: 1px solid #E8EAF0; + padding: 9px 8px; + font-size: 10px; + font-weight: 700; +} +QScrollBar:vertical { width: 9px; background: transparent; margin: 3px; } +QScrollBar::handle:vertical { min-height: 30px; background: #CFD4DE; border-radius: 4px; } +QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { height: 0; } +QScrollBar:horizontal { height: 9px; background: transparent; margin: 3px; } +QScrollBar::handle:horizontal { min-width: 30px; background: #CFD4DE; border-radius: 4px; } +QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal { width: 0; } +QSplitter::handle { background: transparent; width: 12px; } +QLabel#ProcessAvatar { + color: #FFFFFF; + background: #5B5BD6; + border-radius: 12px; + font-size: 20px; + font-weight: 700; +} +QLabel#ProcessName { font-size: 17px; font-weight: 700; color: #1D2230; } +QLabel#PidBadge { + color: #626B7A; + background: #F0F2F6; + border-radius: 7px; + padding: 3px 7px; + font-family: "Cascadia Mono"; + font-size: 9px; +} +QLabel#DetailValue { + color: #242A38; + font-size: 11px; +} +QLabel#PortValue { + color: #4E4EC4; + background: #F0F0FF; + border-radius: 7px; + padding: 7px 9px; + font-family: "Cascadia Mono"; +} +QTextEdit#CommandBox { + color: #3D4656; + background: #F7F8FB; + border: 1px solid #E4E7ED; + border-radius: 8px; + padding: 7px; + font-family: "Cascadia Mono"; + font-size: 9px; +} +QLabel#LimitedBanner { + color: #805C1B; + background: #FFF7E7; + border-radius: 7px; + padding: 7px; +} +QLabel#EmptyIcon { + color: #6868D7; + background: #EEEEFF; + border-radius: 22px; + font-size: 23px; + font-weight: 700; +} +QLabel#EmptyTitle { color: #343A49; font-size: 14px; font-weight: 700; } +QProgressBar#LoadingBar { + min-height: 3px; max-height: 3px; + border: none; background: #EEEFF4; +} +QProgressBar#LoadingBar::chunk { background: #5B5BD6; } +QLabel#FooterStatus { color: #778092; font-size: 10px; } +QToolTip { + color: #FFFFFF; + background: #252A38; + border: none; + padding: 6px; +} +QMessageBox { background: #FFFFFF; } +""" diff --git a/processdock/ui.py b/processdock/ui.py new file mode 100644 index 0000000..209564d --- /dev/null +++ b/processdock/ui.py @@ -0,0 +1,764 @@ +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from PySide6.QtCore import QModelIndex, QObject, QRunnable, QSize, Qt, QThreadPool, QTimer, Signal, Slot +from PySide6.QtGui import QCloseEvent, QColor, QFont, QGuiApplication, QIcon, QPainter, QPixmap +from PySide6.QtWidgets import ( + QAbstractItemView, + QApplication, + QCheckBox, + QFrame, + QHBoxLayout, + QHeaderView, + QLabel, + QLineEdit, + QMainWindow, + QMessageBox, + QProgressBar, + QPushButton, + QScrollArea, + QSizePolicy, + QSplitter, + QStackedWidget, + QStyle, + QTableView, + QTextEdit, + QVBoxLayout, + QWidget, +) + +from processdock.models import ( + PROCESS_ROLE, + ActionButtonDelegate, + ProcessFilterProxyModel, + ProcessTableModel, +) +from processdock.services import ActionResult, ProcessDockError, ProcessService +from processdock.styles import APP_STYLE + + +class WorkerSignals(QObject): + result = Signal(object) + error = Signal(str) + finished = Signal() + + +class FunctionWorker(QRunnable): + def __init__(self, function: Callable[[], Any]) -> None: + super().__init__() + self.function = function + self.signals = WorkerSignals() + + @Slot() + def run(self) -> None: + try: + result = self.function() + except Exception as error: # delivered to the GUI thread + self.signals.error.emit(str(error) or error.__class__.__name__) + else: + self.signals.result.emit(result) + finally: + self.signals.finished.emit() + + +class SummaryCard(QFrame): + def __init__(self, icon: str, title: str, value: str, subtitle: str, parent=None) -> None: + super().__init__(parent) + self.setObjectName("SummaryCard") + self.setMinimumHeight(92) + layout = QHBoxLayout(self) + layout.setContentsMargins(16, 13, 16, 13) + layout.setSpacing(12) + + icon_label = QLabel(icon) + icon_label.setObjectName("CardIcon") + icon_label.setAlignment(Qt.AlignmentFlag.AlignCenter) + layout.addWidget(icon_label) + + text_layout = QVBoxLayout() + text_layout.setSpacing(1) + title_label = QLabel(title) + title_label.setObjectName("CardTitle") + self.value_label = QLabel(value) + self.value_label.setObjectName("CardValue") + self.subtitle_label = QLabel(subtitle) + self.subtitle_label.setObjectName("CardSubtitle") + text_layout.addWidget(title_label) + text_layout.addWidget(self.value_label) + text_layout.addWidget(self.subtitle_label) + layout.addLayout(text_layout, 1) + + def update_value(self, value: str, subtitle: str | None = None) -> None: + self.value_label.setText(value) + if subtitle is not None: + self.subtitle_label.setText(subtitle) + + +class DetailRow(QWidget): + def __init__(self, key: str, parent=None) -> None: + super().__init__(parent) + layout = QHBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + layout.setSpacing(12) + key_label = QLabel(key) + key_label.setObjectName("DetailKey") + key_label.setFixedWidth(76) + key_label.setAlignment(Qt.AlignmentFlag.AlignTop | Qt.AlignmentFlag.AlignLeft) + self.value_label = QLabel("—") + self.value_label.setObjectName("DetailValue") + self.value_label.setTextFormat(Qt.TextFormat.PlainText) + self.value_label.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse) + self.value_label.setWordWrap(True) + layout.addWidget(key_label) + layout.addWidget(self.value_label, 1) + + def set_value(self, value: str) -> None: + self.value_label.setText(value or "—") + + +class DetailPanel(QFrame): + open_requested = Signal(object) + copy_requested = Signal(str) + terminate_requested = Signal(object, bool) + + def __init__(self, parent=None) -> None: + super().__init__(parent) + self.setObjectName("DetailCard") + self.setMinimumWidth(350) + self.setMaximumWidth(430) + self._process: dict[str, Any] | None = None + + outer = QVBoxLayout(self) + outer.setContentsMargins(0, 0, 0, 0) + outer.setSpacing(0) + + heading = QWidget() + heading_layout = QHBoxLayout(heading) + heading_layout.setContentsMargins(18, 16, 18, 12) + title = QLabel("进程详情") + title.setObjectName("SectionTitle") + hint = QLabel("实时信息") + hint.setObjectName("MutedLabel") + heading_layout.addWidget(title) + heading_layout.addStretch() + heading_layout.addWidget(hint) + outer.addWidget(heading) + + self.stack = QStackedWidget() + outer.addWidget(self.stack, 1) + self._build_empty_page() + self._build_detail_page() + self.stack.setCurrentWidget(self.empty_page) + + def _build_empty_page(self) -> None: + self.empty_page = QWidget() + layout = QVBoxLayout(self.empty_page) + layout.setContentsMargins(34, 34, 34, 70) + layout.addStretch() + icon = QLabel("⌁") + icon.setObjectName("EmptyIcon") + icon.setFixedSize(54, 54) + icon.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.empty_title = QLabel("选择一个进程") + self.empty_title.setObjectName("EmptyTitle") + self.empty_title.setAlignment(Qt.AlignmentFlag.AlignCenter) + self.empty_hint = QLabel("点击左侧任意一行，查看路径、命令行和资源占用。") + self.empty_hint.setObjectName("MutedLabel") + self.empty_hint.setWordWrap(True) + self.empty_hint.setAlignment(Qt.AlignmentFlag.AlignCenter) + layout.addWidget(icon, alignment=Qt.AlignmentFlag.AlignCenter) + layout.addSpacing(12) + layout.addWidget(self.empty_title) + layout.addSpacing(4) + layout.addWidget(self.empty_hint) + layout.addStretch() + self.stack.addWidget(self.empty_page) + + def _build_detail_page(self) -> None: + self.detail_page = QWidget() + page_layout = QVBoxLayout(self.detail_page) + page_layout.setContentsMargins(0, 0, 0, 0) + page_layout.setSpacing(0) + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setFrameShape(QFrame.Shape.NoFrame) + detail_content = QWidget() + layout = QVBoxLayout(detail_content) + layout.setContentsMargins(18, 6, 18, 18) + layout.setSpacing(12) + + identity = QWidget() + identity_layout = QHBoxLayout(identity) + identity_layout.setContentsMargins(0, 0, 0, 6) + identity_layout.setSpacing(12) + self.avatar = QLabel("P") + self.avatar.setObjectName("ProcessAvatar") + self.avatar.setFixedSize(50, 50) + self.avatar.setAlignment(Qt.AlignmentFlag.AlignCenter) + name_layout = QVBoxLayout() + name_layout.setSpacing(4) + self.process_name = QLabel("—") + self.process_name.setObjectName("ProcessName") + self.process_name.setTextFormat(Qt.TextFormat.PlainText) + self.pid_badge = QLabel("PID —") + self.pid_badge.setObjectName("PidBadge") + self.pid_badge.setSizePolicy(QSizePolicy.Policy.Fixed, QSizePolicy.Policy.Fixed) + name_layout.addWidget(self.process_name) + name_layout.addWidget(self.pid_badge) + identity_layout.addWidget(self.avatar) + identity_layout.addLayout(name_layout, 1) + layout.addWidget(identity) + + self.limited_banner = QLabel("部分信息受系统权限限制") + self.limited_banner.setObjectName("LimitedBanner") + self.limited_banner.setWordWrap(True) + self.limited_banner.hide() + layout.addWidget(self.limited_banner) + + self.rows: dict[str, DetailRow] = {} + for key, label in ( + ("ports", "监听端口"), + ("exe", "程序路径"), + ("create_time", "启动时间"), + ("cpu", "CPU 使用率"), + ("memory", "内存占用"), + ("username", "运行用户"), + ("status", "进程状态"), + ("ppid", "父进程 PID"), + ): + row = DetailRow(label) + if key == "ports": + row.value_label.setObjectName("PortValue") + self.rows[key] = row + layout.addWidget(row) + + command_title = QLabel("启动命令") + command_title.setObjectName("DetailKey") + layout.addWidget(command_title) + self.command_box = QTextEdit() + self.command_box.setObjectName("CommandBox") + self.command_box.setReadOnly(True) + self.command_box.setMinimumHeight(76) + self.command_box.setMaximumHeight(108) + layout.addWidget(self.command_box) + layout.addStretch() + + actions = QWidget() + actions_layout = QVBoxLayout(actions) + actions_layout.setContentsMargins(18, 10, 18, 18) + actions_layout.setSpacing(8) + utility_layout = QHBoxLayout() + utility_layout.setSpacing(8) + self.open_button = QPushButton("打开文件位置") + self.open_button.setObjectName("SecondaryButton") + self.copy_button = QPushButton("复制启动命令") + self.copy_button.setObjectName("SecondaryButton") + utility_layout.addWidget(self.open_button) + utility_layout.addWidget(self.copy_button) + actions_layout.addLayout(utility_layout) + + danger_layout = QHBoxLayout() + danger_layout.setSpacing(8) + self.normal_button = QPushButton("正常结束") + self.normal_button.setObjectName("SoftDangerButton") + self.force_button = QPushButton("强制结束") + self.force_button.setObjectName("DangerButton") + danger_layout.addWidget(self.normal_button) + danger_layout.addWidget(self.force_button) + actions_layout.addLayout(danger_layout) + + self.open_button.clicked.connect(self._request_open) + self.copy_button.clicked.connect(self._request_copy) + self.normal_button.clicked.connect(lambda: self._request_terminate(False)) + self.force_button.clicked.connect(lambda: self._request_terminate(True)) + + scroll.setWidget(detail_content) + page_layout.addWidget(scroll, 1) + page_layout.addWidget(actions) + self.stack.addWidget(self.detail_page) + self.detail_scroll = scroll + + def set_process(self, process: dict[str, Any]) -> None: + self._process = process + name = process["name"] + self.avatar.setText((name[:1] or "P").upper()) + self.process_name.setText(name) + self.pid_badge.setText(f'PID {process["pid"]}') + self.limited_banner.setVisible(bool(process["access_limited"])) + + ports = "、".join(str(port) for port in process["ports"]) or "未监听端口" + port_tooltip = "\n".join( + f'{item["protocol"]} {item["address"]}:{item["port"]}' + for item in process["port_details"] + ) + self.rows["ports"].set_value(ports) + self.rows["ports"].value_label.setToolTip(port_tooltip) + self.rows["exe"].set_value(process["exe"]) + self.rows["create_time"].set_value(process["create_time_text"]) + self.rows["cpu"].set_value(f'{process["cpu_percent"]:.1f}%') + self.rows["memory"].set_value( + f'{process["memory_text"]} ({process["memory_percent"]:.1f}%)' + ) + self.rows["username"].set_value(process["username"]) + self.rows["status"].set_value(process["status_label"]) + self.rows["ppid"].set_value(str(process["ppid"]) if process["ppid"] else "—") + self.command_box.setPlainText(process["command"] or "—") + self.open_button.setEnabled(bool(process["can_open_location"])) + self.copy_button.setEnabled(bool(process["command"])) + self.normal_button.setEnabled(bool(process["can_terminate"])) + self.force_button.setEnabled(bool(process["can_terminate"])) + self.normal_button.setToolTip( + "向进程发送常规结束请求" if process["can_terminate"] else "不能结束 ProcessDock 自身" + ) + self.force_button.setToolTip("立即终止进程，未保存的数据可能丢失") + self.stack.setCurrentIndex(1) + + def clear(self, message: str = "点击左侧任意一行，查看路径、命令行和资源占用。") -> None: + self._process = None + self.empty_hint.setText(message) + self.stack.setCurrentWidget(self.empty_page) + + def set_action_busy(self, busy: bool) -> None: + if not self._process: + return + self.open_button.setEnabled(not busy and bool(self._process["can_open_location"])) + self.copy_button.setEnabled(not busy and bool(self._process["command"])) + self.normal_button.setEnabled(not busy and bool(self._process["can_terminate"])) + self.force_button.setEnabled(not busy and bool(self._process["can_terminate"])) + + def _request_open(self) -> None: + if self._process: + self.open_requested.emit(self._process) + + def _request_copy(self) -> None: + if self._process and self._process["command"]: + self.copy_requested.emit(self._process["command"]) + + def _request_terminate(self, force: bool) -> None: + if self._process: + self.terminate_requested.emit(self._process, force) + + +class MainWindow(QMainWindow): + def __init__( + self, service: ProcessService | None = None, auto_refresh: bool = True, parent=None + ) -> None: + super().__init__(parent) + self.service = service or ProcessService() + self.thread_pool = QThreadPool.globalInstance() + self._workers: set[FunctionWorker] = set() + self._refreshing = False + self._refresh_pending = False + self._closing = False + self._selected_pid: int | None = None + self._action_busy = False + + self.setWindowTitle("ProcessDock · 进程与端口管理器") + self.setWindowIcon(make_app_icon()) + self.setMinimumSize(1080, 700) + self.resize(1360, 820) + self.setStyleSheet(APP_STYLE) + self._build_ui(auto_refresh) + + self.auto_timer = QTimer(self) + self.auto_timer.setInterval(3000) + self.auto_timer.timeout.connect(self.refresh) + self.auto_checkbox.toggled.connect(self._toggle_auto_refresh) + if auto_refresh: + self.auto_timer.start() + QTimer.singleShot(0, self.refresh) + + def _build_ui(self, auto_refresh: bool) -> None: + root = QWidget() + root.setObjectName("AppRoot") + self.setCentralWidget(root) + outer = QVBoxLayout(root) + outer.setContentsMargins(20, 18, 20, 16) + outer.setSpacing(12) + + header = QFrame() + header.setObjectName("HeaderCard") + header_layout = QHBoxLayout(header) + header_layout.setContentsMargins(16, 11, 16, 11) + header_layout.setSpacing(12) + mark = QLabel("P") + mark.setObjectName("BrandMark") + mark.setFixedSize(42, 42) + mark.setAlignment(Qt.AlignmentFlag.AlignCenter) + header_layout.addWidget(mark) + brand = QVBoxLayout() + brand.setSpacing(0) + title = QLabel("ProcessDock") + title.setObjectName("BrandTitle") + subtitle = QLabel("进程与端口管理器") + subtitle.setObjectName("BrandSubtitle") + brand.addWidget(title) + brand.addWidget(subtitle) + header_layout.addLayout(brand) + header_layout.addStretch() + + self.search_box = QLineEdit() + self.search_box.setObjectName("SearchBox") + self.search_box.setPlaceholderText("搜索进程、PID、端口或路径") + self.search_box.setClearButtonEnabled(True) + self.search_box.textChanged.connect(self._filter_changed) + header_layout.addWidget(self.search_box) + + self.refresh_button = QPushButton("刷新") + self.refresh_button.setObjectName("PrimaryButton") + self.refresh_button.setIcon( + self.style().standardIcon(QStyle.StandardPixmap.SP_BrowserReload) + ) + self.refresh_button.clicked.connect(self.refresh) + header_layout.addWidget(self.refresh_button) + self.auto_checkbox = QCheckBox("自动刷新 3 秒") + self.auto_checkbox.setChecked(auto_refresh) + header_layout.addWidget(self.auto_checkbox) + outer.addWidget(header) + + cards_layout = QHBoxLayout() + cards_layout.setSpacing(10) + self.process_card = SummaryCard("进", "运行进程", "—", "正在读取") + self.port_card = SummaryCard("端", "监听端口", "—", "TCP 与 UDP") + self.cpu_card = SummaryCard("CPU", "系统 CPU", "—", "所有逻辑核心") + self.memory_card = SummaryCard("存", "系统内存", "—", "正在读取") + for card in (self.process_card, self.port_card, self.cpu_card, self.memory_card): + cards_layout.addWidget(card, 1) + outer.addLayout(cards_layout) + + self.warning_banner = QLabel() + self.warning_banner.setObjectName("WarningBanner") + self.warning_banner.setWordWrap(True) + self.warning_banner.hide() + outer.addWidget(self.warning_banner) + + splitter = QSplitter(Qt.Orientation.Horizontal) + splitter.setChildrenCollapsible(False) + outer.addWidget(splitter, 1) + + table_card = QFrame() + table_card.setObjectName("TableCard") + table_layout = QVBoxLayout(table_card) + table_layout.setContentsMargins(0, 0, 0, 0) + table_layout.setSpacing(0) + table_heading = QWidget() + table_heading_layout = QHBoxLayout(table_heading) + table_heading_layout.setContentsMargins(18, 14, 18, 11) + table_title = QLabel("正在运行的进程") + table_title.setObjectName("SectionTitle") + self.count_label = QLabel("准备读取…") + self.count_label.setObjectName("MutedLabel") + table_heading_layout.addWidget(table_title) + table_heading_layout.addStretch() + table_heading_layout.addWidget(self.count_label) + table_layout.addWidget(table_heading) + + self.loading_bar = QProgressBar() + self.loading_bar.setObjectName("LoadingBar") + self.loading_bar.setRange(0, 0) + self.loading_bar.setTextVisible(False) + self.loading_bar.hide() + table_layout.addWidget(self.loading_bar) + + self.model = ProcessTableModel(self) + self.proxy = ProcessFilterProxyModel(self) + self.proxy.setSourceModel(self.model) + self.table = QTableView() + self.table.setModel(self.proxy) + self.table.setAlternatingRowColors(True) + self.table.setSelectionBehavior(QAbstractItemView.SelectionBehavior.SelectRows) + self.table.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) + self.table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers) + self.table.setShowGrid(False) + self.table.setWordWrap(False) + self.table.setSortingEnabled(True) + self.table.sortByColumn(1, Qt.SortOrder.AscendingOrder) + self.table.verticalHeader().hide() + self.table.verticalHeader().setDefaultSectionSize(48) + header_view = self.table.horizontalHeader() + header_view.setDefaultAlignment(Qt.AlignmentFlag.AlignCenter) + header_view.setMinimumSectionSize(62) + header_view.setSectionResizeMode(QHeaderView.ResizeMode.Interactive) + header_view.setSectionResizeMode(1, QHeaderView.ResizeMode.Stretch) + widths = (78, 190, 135, 82, 105, 130, 88, 82) + for column, width in enumerate(widths): + self.table.setColumnWidth(column, width) + self.table.selectionModel().currentRowChanged.connect(self._selection_changed) + self.action_delegate = ActionButtonDelegate(self.table) + self.action_delegate.terminate_requested.connect( + lambda process: self._confirm_terminate(process, False) + ) + self.table.setItemDelegateForColumn(7, self.action_delegate) + + self.table_stack = QStackedWidget() + self.table_stack.addWidget(self.table) + self.no_results_page = self._build_no_results_page() + self.table_stack.addWidget(self.no_results_page) + table_layout.addWidget(self.table_stack, 1) + splitter.addWidget(table_card) + + self.detail_panel = DetailPanel() + self.detail_panel.open_requested.connect(self._open_location) + self.detail_panel.copy_requested.connect(self._copy_command) + self.detail_panel.terminate_requested.connect(self._confirm_terminate) + splitter.addWidget(self.detail_panel) + splitter.setStretchFactor(0, 1) + splitter.setStretchFactor(1, 0) + splitter.setSizes([910, 390]) + + footer = QHBoxLayout() + self.footer_status = QLabel("就绪") + self.footer_status.setObjectName("FooterStatus") + self.footer_time = QLabel("尚未刷新") + self.footer_time.setObjectName("FooterStatus") + footer.addWidget(self.footer_status) + footer.addStretch() + footer.addWidget(self.footer_time) + outer.addLayout(footer) + + @staticmethod + def _build_no_results_page() -> QWidget: + page = QWidget() + layout = QVBoxLayout(page) + layout.addStretch() + icon = QLabel("⌕") + icon.setObjectName("EmptyIcon") + icon.setFixedSize(54, 54) + icon.setAlignment(Qt.AlignmentFlag.AlignCenter) + title = QLabel("没有匹配的进程") + title.setObjectName("EmptyTitle") + title.setAlignment(Qt.AlignmentFlag.AlignCenter) + hint = QLabel("换一个进程名称、PID 或端口试试。") + hint.setObjectName("MutedLabel") + hint.setAlignment(Qt.AlignmentFlag.AlignCenter) + layout.addWidget(icon, alignment=Qt.AlignmentFlag.AlignCenter) + layout.addSpacing(10) + layout.addWidget(title) + layout.addWidget(hint) + layout.addStretch() + return page + + @Slot() + def refresh(self) -> None: + if self._refreshing: + self._refresh_pending = True + return + self._refreshing = True + self.refresh_button.setEnabled(False) + self.loading_bar.show() + self._set_status("正在读取进程和端口…") + + worker = FunctionWorker(self.service.snapshot) + worker.signals.result.connect(self._apply_snapshot) + worker.signals.error.connect(self._refresh_failed) + worker.signals.finished.connect(self._refresh_finished) + self._track_worker(worker) + self.thread_pool.start(worker) + + @Slot(object) + def _apply_snapshot(self, snapshot: dict[str, Any]) -> None: + if self._closing: + return + processes = snapshot["processes"] + self.model.set_processes(processes) + summary = snapshot["summary"] + self.process_card.update_value(str(summary["process_count"]), "本机可见进程") + self.port_card.update_value(str(summary["port_count"]), "TCP 监听 / UDP 绑定") + self.cpu_card.update_value(f'{summary["cpu_percent"]:.1f}%', "所有逻辑核心") + self.memory_card.update_value( + f'{summary["memory_percent"]:.1f}%', + f'{summary["memory_used"]} / {summary["memory_total"]}', + ) + warnings = snapshot.get("warnings") or [] + self.warning_banner.setText(" ".join(warnings)) + self.warning_banner.setVisible(bool(warnings)) + self.footer_time.setText(f'上次刷新 {snapshot["refreshed_at"]}') + self._set_status("数据已更新") + self._update_visible_count() + + if self._selected_pid is not None: + selected = next( + (process for process in processes if process["pid"] == self._selected_pid), + None, + ) + if selected is None: + self._selected_pid = None + self.detail_panel.clear("该进程已经结束，请重新选择。") + else: + self.detail_panel.set_process(selected) + self._reselect_process(self._selected_pid) + + @Slot(str) + def _refresh_failed(self, message: str) -> None: + self.warning_banner.setText(f"刷新失败：{message}") + self.warning_banner.show() + self._set_status("刷新失败") + + @Slot() + def _refresh_finished(self) -> None: + self._refreshing = False + self.refresh_button.setEnabled(True) + self.loading_bar.hide() + if self._refresh_pending and not self._closing: + self._refresh_pending = False + QTimer.singleShot(0, self.refresh) + + def _track_worker(self, worker: FunctionWorker) -> None: + self._workers.add(worker) + worker.signals.finished.connect(lambda current=worker: self._workers.discard(current)) + + @Slot(str) + def _filter_changed(self, query: str) -> None: + self.proxy.set_query(query) + self._update_visible_count() + + def _update_visible_count(self) -> None: + visible = self.proxy.rowCount() + total = self.model.rowCount() + self.count_label.setText( + f"{visible} / {total} 个进程" if self.search_box.text().strip() else f"{total} 个进程" + ) + self.table_stack.setCurrentWidget(self.table if visible else self.no_results_page) + + @Slot(QModelIndex, QModelIndex) + def _selection_changed(self, current: QModelIndex, previous: QModelIndex) -> None: + del previous + if not current.isValid(): + return + process = current.data(PROCESS_ROLE) + if process: + self._selected_pid = process["pid"] + self.detail_panel.set_process(process) + + def _reselect_process(self, pid: int) -> None: + for source_row, process in enumerate(self.model.processes): + if process["pid"] != pid: + continue + source_index = self.model.index(source_row, 0) + proxy_index = self.proxy.mapFromSource(source_index) + if proxy_index.isValid(): + self.table.selectRow(proxy_index.row()) + return + + @Slot(bool) + def _toggle_auto_refresh(self, enabled: bool) -> None: + if enabled: + self.auto_timer.start() + self._set_status("已开启自动刷新") + else: + self.auto_timer.stop() + self._set_status("已暂停自动刷新") + + @Slot(object, bool) + def _confirm_terminate(self, process: dict[str, Any], force: bool) -> None: + if self._action_busy or not process.get("can_terminate"): + return + name = process["name"] + pid = process["pid"] + box = QMessageBox(self) + box.setIcon(QMessageBox.Icon.Warning) + box.setWindowTitle("确认强制结束" if force else "确认结束进程") + box.setText(f"确定要结束 {name}（PID {pid}）吗？") + box.setInformativeText( + "强制结束可能导致未保存的数据丢失。" + if force + else "将先发送常规结束请求；如果进程没有响应，可再强制结束。" + ) + confirm = box.addButton( + "强制结束" if force else "正常结束", QMessageBox.ButtonRole.AcceptRole + ) + box.addButton("取消", QMessageBox.ButtonRole.RejectRole) + box.setDefaultButton(confirm) + box.exec() + if box.clickedButton() is not confirm: + return + + self._action_busy = True + if self._selected_pid == pid: + self.detail_panel.set_action_busy(True) + self._set_status(f"正在结束 {name}…") + worker = FunctionWorker( + lambda: self.service.terminate( + pid, + force=force, + expected_create_time=process.get("create_time"), + ) + ) + worker.signals.result.connect(self._termination_finished) + worker.signals.error.connect(self._action_failed) + worker.signals.finished.connect(self._action_worker_finished) + self._track_worker(worker) + self.thread_pool.start(worker) + + @Slot(object) + def _termination_finished(self, result: ActionResult) -> None: + self._set_status(result.message) + if result.still_running: + QMessageBox.warning(self, "进程仍在运行", result.message) + self.refresh() + + @Slot(str) + def _action_failed(self, message: str) -> None: + self._set_status("操作失败") + QMessageBox.critical(self, "操作失败", message) + + @Slot() + def _action_worker_finished(self) -> None: + self._action_busy = False + self.detail_panel.set_action_busy(False) + + @Slot(object) + def _open_location(self, process: dict[str, Any]) -> None: + if self._action_busy: + return + self._action_busy = True + self.detail_panel.set_action_busy(True) + worker = FunctionWorker( + lambda: self.service.open_file_location( + process["pid"], expected_create_time=process.get("create_time") + ) + ) + worker.signals.result.connect(lambda result: self._set_status(result.message)) + worker.signals.error.connect(self._action_failed) + worker.signals.finished.connect(self._action_worker_finished) + self._track_worker(worker) + self.thread_pool.start(worker) + + @Slot(str) + def _copy_command(self, command: str) -> None: + QGuiApplication.clipboard().setText(command) + self._set_status("启动命令已复制到剪贴板") + + def _set_status(self, text: str) -> None: + self.footer_status.setText(text) + + def closeEvent(self, event: QCloseEvent) -> None: # noqa: N802 + self._closing = True + self.auto_timer.stop() + self.thread_pool.clear() + super().closeEvent(event) + + +def make_app_icon() -> QIcon: + pixmap = QPixmap(64, 64) + pixmap.fill(Qt.GlobalColor.transparent) + painter = QPainter(pixmap) + painter.setRenderHint(QPainter.RenderHint.Antialiasing) + painter.setPen(Qt.PenStyle.NoPen) + painter.setBrush(QColor("#5B5BD6")) + painter.drawRoundedRect(4, 4, 56, 56, 15, 15) + painter.setPen(QColor("#FFFFFF")) + font = QFont("Arial", 29, QFont.Weight.Bold) + painter.setFont(font) + painter.drawText(pixmap.rect(), Qt.AlignmentFlag.AlignCenter, "P") + painter.end() + return QIcon(pixmap) + + +def create_window_for_testing(service: ProcessService | None = None) -> MainWindow: + """Small test hook used by the off-screen smoke check.""" + if QApplication.instance() is None: + raise RuntimeError("A QApplication must exist before creating the window.") + return MainWindow(service=service, auto_refresh=False) diff --git a/requirements-build.txt b/requirements-build.txt new file mode 100644 index 0000000..c2985bd --- /dev/null +++ b/requirements-build.txt @@ -0,0 +1,3 @@ +-r requirements.txt +pyinstaller>=6.10,<7 +Pillow>=10,<13 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6a25756 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +psutil>=5.9,<8 +PySide6>=6.7,<7 diff --git a/tests/test_gui.py b/tests/test_gui.py new file mode 100644 index 0000000..0bb6681 --- /dev/null +++ b/tests/test_gui.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import os +import unittest + +os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + +from PySide6.QtWidgets import QApplication # noqa: E402 + +from processdock.ui import MainWindow # noqa: E402 + + +class StubService: + def snapshot(self): + return {"processes": [], "summary": {}, "warnings": [], "refreshed_at": "00:00:00"} + + +class GuiSmokeTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.application = QApplication.instance() or QApplication([]) + + def setUp(self) -> None: + self.window = MainWindow(service=StubService(), auto_refresh=False) + + def tearDown(self) -> None: + self.window.close() + self.application.processEvents() + + def test_window_contains_table_search_and_detail_panel(self) -> None: + self.assertEqual(self.window.model.columnCount(), 8) + self.assertIn("端口", self.window.search_box.placeholderText()) + self.assertEqual(self.window.detail_panel.stack.currentIndex(), 0) + + def test_filter_finds_port(self) -> None: + process = { + "pid": 4420, + "name": "node.exe", + "ports": [3000], + "port_details": [{"protocol": "TCP", "address": "0.0.0.0", "port": 3000}], + "cpu_percent": 3.0, + "memory_bytes": 188743680, + "memory_text": "180.0 MB", + "memory_percent": 1.0, + "username": "wang", + "status": "running", + "status_label": "运行中", + "exe": r"D:\node\node.exe", + "command": "node server.js", + "create_time": 1.0, + "create_time_text": "2026-07-17 08:20:11", + "ppid": 100, + "access_limited": False, + "is_current": False, + "can_terminate": True, + "can_open_location": True, + } + self.window.model.set_processes([process]) + self.window.search_box.setText("3000") + self.assertEqual(self.window.proxy.rowCount(), 1) + self.window.search_box.setText("8080") + self.assertEqual(self.window.proxy.rowCount(), 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_services.py b/tests/test_services.py new file mode 100644 index 0000000..633743f --- /dev/null +++ b/tests/test_services.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +import os +import subprocess +import sys +import unittest + +from processdock.services import ( + ProcessService, + ProtectedProcessError, + format_command, + humanize_bytes, + matches_process, +) + + +class FormattingTests(unittest.TestCase): + def test_humanize_bytes(self) -> None: + self.assertEqual(humanize_bytes(0), "0 B") + self.assertEqual(humanize_bytes(1024), "1.0 KB") + self.assertEqual(humanize_bytes(512 * 1024 * 1024), "512.0 MB") + + def test_format_command_preserves_arguments(self) -> None: + command = format_command(["python", "demo app.py", "--port", "8080"]) + self.assertIn("python", command) + self.assertIn("demo app.py", command) + self.assertIn("8080", command) + + def test_searches_name_pid_port_path_and_command(self) -> None: + process = { + "pid": 9824, + "name": "java.exe", + "username": "wang", + "exe": r"D:\jdk-17\bin\java.exe", + "command": "java -jar demo.jar", + "ports": [8080, 9000], + } + for query in ("JAVA", "9824", "8080", "jdk-17", "demo.jar", "wang"): + with self.subTest(query=query): + self.assertTrue(matches_process(process, query)) + self.assertFalse(matches_process(process, "nginx")) + + +class LiveServiceTests(unittest.TestCase): + def test_snapshot_contains_current_process_and_required_fields(self) -> None: + snapshot = ProcessService().snapshot() + current = next(item for item in snapshot["processes"] if item["pid"] == os.getpid()) + required = { + "pid", + "name", + "ports", + "cpu_percent", + "memory_bytes", + "username", + "status_label", + "exe", + "command", + "create_time_text", + "ppid", + } + self.assertTrue(required.issubset(current)) + self.assertIn("process_count", snapshot["summary"]) + + def test_service_refuses_to_terminate_itself(self) -> None: + service = ProcessService() + with self.assertRaises(ProtectedProcessError): + service.terminate(os.getpid(), force=True) + + def test_force_terminates_owned_child_process(self) -> None: + child = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(30)"], + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + try: + result = ProcessService().terminate(child.pid, force=True) + self.assertTrue(result.success) + self.assertFalse(result.still_running) + finally: + if child.poll() is None: + child.kill() + child.wait() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/generate_icon.py b/tools/generate_icon.py new file mode 100644 index 0000000..e1a3ee7 --- /dev/null +++ b/tools/generate_icon.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + + +ROOT = Path(__file__).resolve().parents[1] +ASSETS = ROOT / "assets" + + +def load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: + candidates = ( + Path(r"C:\Windows\Fonts\arialbd.ttf"), + Path(r"C:\Windows\Fonts\segoeuib.ttf"), + ) + for candidate in candidates: + if candidate.exists(): + return ImageFont.truetype(str(candidate), size) + return ImageFont.load_default() + + +def main() -> None: + ASSETS.mkdir(parents=True, exist_ok=True) + size = 1024 + image = Image.new("RGBA", (size, size), (0, 0, 0, 0)) + draw = ImageDraw.Draw(image) + + # Soft shadow and the same violet visual language used by the GUI. + draw.rounded_rectangle((76, 84, 948, 956), radius=220, fill=(28, 34, 67, 52)) + draw.rounded_rectangle((52, 52, 924, 924), radius=220, fill=(91, 91, 214, 255)) + draw.rounded_rectangle((82, 82, 894, 894), radius=190, outline=(130, 130, 235, 255), width=10) + + font = load_font(540) + label = "P" + bounds = draw.textbbox((0, 0), label, font=font) + width = bounds[2] - bounds[0] + height = bounds[3] - bounds[1] + position = ((size - width) / 2 - 14, (size - height) / 2 - bounds[1] - 4) + draw.text(position, label, font=font, fill=(255, 255, 255, 255)) + + png_path = ASSETS / "processdock.png" + ico_path = ASSETS / "processdock.ico" + image.save(png_path, optimize=True) + image.save( + ico_path, + format="ICO", + sizes=[(16, 16), (24, 24), (32, 32), (48, 48), (64, 64), (128, 128), (256, 256)], + ) + print(f"Generated {ico_path}") + + +if __name__ == "__main__": + main()