feat: add ProcessDock process and port manager

This commit is contained in:
王鹏
2026-07-17 09:14:45 +08:00
commit 13bb37585f
22 changed files with 2201 additions and 0 deletions

4
.gitattributes vendored Normal file
View File

@@ -0,0 +1,4 @@
* text=auto eol=lf
*.ico binary
*.png binary
*.exe binary

9
.gitignore vendored Normal file
View File

@@ -0,0 +1,9 @@
__pycache__/
*.py[cod]
.pytest_cache/
.venv/
dist/
build/
*.spec
*.log
!ProcessDock.spec

5
ProcessDock.pyw Normal file
View File

@@ -0,0 +1,5 @@
from app import main
if __name__ == "__main__":
raise SystemExit(main())

85
ProcessDock.spec Normal file
View File

@@ -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"),
)

87
README.md Normal file
View File

@@ -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 需求证据与验收范围
```

39
app.py Normal file
View File

@@ -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())

Binary file not shown.

After

Width:  |  Height:  |  Size: 163 KiB

BIN
assets/processdock.ico Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 26 KiB

BIN
assets/processdock.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 13 KiB

View File

@@ -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])])
]
)

28
build_exe.ps1 Normal file
View File

@@ -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)"

53
docs/blueprint.md Normal file
View File

@@ -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 行为。

3
processdock/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""ProcessDock - a local process and port manager."""
__version__ = "1.0.0"

187
processdock/models.py Normal file
View File

@@ -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,
)

499
processdock/services.py Normal file
View File

@@ -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

198
processdock/styles.py Normal file
View File

@@ -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; }
"""

764
processdock/ui.py Normal file
View File

@@ -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)

3
requirements-build.txt Normal file
View File

@@ -0,0 +1,3 @@
-r requirements.txt
pyinstaller>=6.10,<7
Pillow>=10,<13

2
requirements.txt Normal file
View File

@@ -0,0 +1,2 @@
psutil>=5.9,<8
PySide6>=6.7,<7

66
tests/test_gui.py Normal file
View File

@@ -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()

85
tests/test_services.py Normal file
View File

@@ -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()

54
tools/generate_icon.py Normal file
View File

@@ -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()