chore: 同步本地更改

- 新增 aio-mcp 项目框架
- 新增 .trae/skills/ OCR/SKU 设计工具
- 更新 auto-check 配置和 stress 包
- 删除 fnhelp 文档目录
- 更新 Docker 压力测试脚本
This commit is contained in:
张威33321
2026-08-27 20:38:13 +08:00
parent a172a9bb5e
commit a8fa159f0f
75 changed files with 2266 additions and 4068 deletions

View File

@@ -32,20 +32,34 @@ FROM debian:12-slim
ENV LANG=C.UTF-8 LC_ALL=C.UTF-8
# 清理缓存,减少镜像体积
# - python3/python3-psutil/memtester/stress/sudofnscript 稳定性门禁的依赖
# self_inspection.py 调 psutil 计算测试内存、sudo memtester 做数据校验、
# stress 做 CPU 压测;脚本原样收录在 vendor/fnscript/,源码不做修改)
RUN apt-get update && apt-get install -y --no-install-recommends \
stress-ng \
fio \
lm-sensors \
sysstat \
procps \
python3 \
python3-psutil \
memtester \
stress \
sudo \
&& rm -rf /var/lib/apt/lists/*
# 验证工具安装
RUN stress-ng --version && fio --version && sensors --version
RUN stress-ng --version && fio --version && sensors --version \
&& python3 -c "import psutil; print('psutil', psutil.__version__)" \
&& command -v memtester && command -v stress && command -v sudo
# 复制测试脚本到镜像中
COPY scripts/ /scripts/
# fnscript 稳定性测试脚本(社区脚本 vendor 收录,源码不改,见 vendor/fnscript/README.md
COPY vendor/fnscript/script/self_inspection.py /vendor/fnscript/self_inspection.py
COPY vendor/fnscript/LICENSE /vendor/fnscript/LICENSE
# 设置脚本执行权限
RUN chmod +x /scripts/*.sh

View File

@@ -1,141 +0,0 @@
# stress-test 镜像
压力测试镜像,为 auto-check 提供预装压测工具与测试脚本的容器环境。压测统一在容器内执行,无需在目标机在线安装任何依赖。
## 工作流程
```
┌─────────────────────────────────────────────────────────────┐
│ auto-check 工作流程 │
│ │
│ 1. 检测目标机器的 Docker 可用性 │
│ 2. 检查镜像是否存在(不存在则上传本地导出的 tar 并 docker load
│ 3. 启动容器执行测试(脚本已打包在镜像中) │
│ 4. 收集测试结果并解析 │
│ 5. 生成报告 │
└─────────────────────────────────────────────────────────────┘
```
## 优势
| 特性 | 说明 |
|------|------|
| 工具获取 | 预装在镜像中,无需在线安装 |
| 网络依赖 | 镜像已缓存后目标机无需联网 |
| 环境一致性 | 统一的容器环境,不受目标机发行版差异影响 |
## 构建与导出(维护时使用)
固定分发模式:不依赖 registry镜像以 tar 文件形式放在工程目录,
目标机无镜像时由 auto-check 自动上传并 `docker load`
### 推荐:使用构建脚本(自动时间戳 tag
`build.sh` 会以 `YYYYMMDD-HHMMSS` 格式自动生成时间戳 tag`stress-test:20260824-153045`
便于在 `docker images` 中一眼区分新旧版本。脚本会同时:
1. 用时间戳 tag 构建镜像(**不再保留 `latest`**
2. 导出两份 tar带时间戳的归档 + 同步覆盖 `stress-test.tar`auto-check 默认引用)
3. 自动更新 `auto-check.yaml``docker_image` 字段,确保配置与新 tag 同步
```bash
cd docker/stress-test
./build.sh # 自动取当前时间戳
./build.sh 20260824-153045 # 手动指定时间戳(重打/复现)
```
产物:
| 文件 | 用途 |
|------|------|
| 镜像 `stress-test:<时间戳>` | 唯一 tag`docker images` 可看到每次构建 |
| Tar `stress-test-<时间戳>.tar` | 带时间戳归档,保留历史版本 |
| Tar `stress-test.tar` | auto-check 默认引用,脚本自动覆盖为最新 |
| `auto-check.yaml` | `docker_image` 自动同步为新 tag |
> 旧时间戳镜像可手动 `docker rmi stress-test:<旧时间戳>` 清理。
### 手动构建(不推荐)
如需绕过脚本手动构建,必须自行同步 `auto-check.yaml`
```bash
cd docker/stress-test
docker build -t stress-test:20260824-153045 . # 必须带时间戳
docker save -o stress-test.tar stress-test:20260824-153045
# 然后编辑 ../../auto-check.yaml把 docker_image 改成新 tag
```
> 注意:手动构建如忘记带时间戳,重复构建后会丢失版本信息,调试时易混淆。
## 使用
`auto-check.yaml` 中配置(`docker_image``build.sh` 自动维护):
```yaml
stress:
docker_image: "stress-test:20260824-153045" # 每次构建自动更新为新时间戳
docker_tar: "docker/stress-test/stress-test.tar"
docker_args: "--network=host --privileged"
```
## 环境变量
容器运行时通过以下环境变量传入参数:
| 变量 | 说明 | 默认值 |
|------|------|--------|
| `AUTOCHECK_TYPES` | 测试类型(逗号分隔) | `cpu,memory` |
| `AUTOCHECK_DURATION` | 持续时间(秒) | `30` |
| `AUTOCHECK_THREADS` | 线程数 | `4` |
| `AUTOCHECK_MEM_SIZE_MB` | 内存压测大小 | `0` (自动) |
| `AUTOCHECK_DISK_SIZE_MB` | 磁盘压测大小 | `1024` |
| `AUTOCHECK_DISK_DIR` | 磁盘压测目录 | `/tmp` |
| `AUTOCHECK_TEMP_INTERVAL` | 温度采样间隔 | `10` |
| `AUTOCHECK_SAMPLE_INTERVAL` | 运行时采样间隔 | `2` |
## 容器内执行流程run.sh
1. 系统信息(`===SYSTEM_INFO===`
2. 温度监控后台采样(`temp_start.sh`
3. 各项压测先单项cpu/memory/disk/memnative、综合full最后
4. 温度监控收尾(`===MONITOR:temp===`
5. 内核日志(`===MONITOR:dmesg===``-T` 带时间戳)+ 结束标记
## 容器运行时挂载(由 auto-check 自动添加)
| 挂载 | 作用 |
|------|------|
| `-v /sys:/sys:ro` | 容器内是独立 sysfs挂载后才能读到宿主机硬件传感器温度采集依赖此项 |
| `-v /etc/localtime:/etc/localtime:ro` | 容器时区与宿主机一致dmesg 时间戳与采样时间才准确 |
> 注意:镜像内已设置 `LC_ALL=C.UTF-8`。若 locale 为纯 C`sensors` 会输出 "+51.0 C"(无 ° 符号),导致温度正则失配、采集不到数据。
## 镜像内容
```
/stress-test/
├── Dockerfile # 镜像定义
├── scripts/ # 测试脚本
│ ├── run.sh # 入口脚本
│ ├── cpu.sh # CPU 压测
│ ├── memory.sh # 内存压测
│ ├── disk.sh # 磁盘压测
│ ├── full.sh # 综合压测
│ ├── memnative.sh # 内存精压
│ ├── sysinfo.sh # 系统信息
│ ├── temp_start.sh # 温度监控启动
│ ├── temp_end.sh # 温度监控收尾
│ └── dmesg.sh # 内核日志
└── README.md # 本文档
```
## 依赖工具
| 工具 | 用途 |
|------|------|
| `stress-ng` | CPU/内存/磁盘压测 |
| `fio` | 磁盘 IO 测试 |
| `lm-sensors` | 温度监控 |
| `sysstat` | 系统性能统计 |
| `procps` | 进程监控 |

View File

@@ -0,0 +1,81 @@
# fnstable.sh - fnscript 稳定性门禁(社区脚本 self_inspection.pyvendor 原样收录不改源码)
#
# 思路先用社区脚本检测硬件真实稳定性stress CPU 压测 + memtester 内存数据校验),
# 通过后自有压测工具再跑一遍出报告;不通过则 run.sh 跳过自有压测。
#
# 适配注意(不改源码的代价,已逐一规避):
# 1. self_inspection.py CLI 模式无论成败 exit code 均为 0
# 只能按 stdout 关键词判定结果("未发现问题" = 通过)。
# 2. --memory 内含两处 y/n 交互确认input()),非交互环境会 EOFError
# 用 printf 'y\ny\n' 管道喂入(两处确认都覆盖)。
# 3. 脚本内部调 sudo memtester / stress依赖已在镜像中预装root 下 sudo 直接可用)。
# 4. memtester 测试可用内存的 97%,单遍耗时 30-120 分钟(由内存大小决定),
# 可用 AUTOCHECK_FN_MEMORY=0 关闭内存校验只做 CPU 门禁。
#
# 输出遵守 auto-check 结构化协议(===TEST:fnstable=== 块),
# 并设置 GATE_PASSED 供 run.sh 决定是否继续自有压测。
echo "===TEST:fnstable==="
FN_SCRIPT="/vendor/fnscript/self_inspection.py"
FN_DURATION="${AUTOCHECK_DURATION:-30}"
FN_MEMORY="${AUTOCHECK_FN_MEMORY:-1}"
FN_ALL_PASSED=1
START_TIME=$(date +%s%N)
# ---- 1. CPU 稳定性fnscript: stress --cpu 4 --timeout N----
set +e
FN_CPU_OUT=$(python3 "$FN_SCRIPT" --cpu "$FN_DURATION" 2>&1)
set -e
if printf '%s' "$FN_CPU_OUT" | grep -q "未发现问题"; then
FN_CPU_STATUS="pass"
else
FN_CPU_STATUS="fail"
FN_ALL_PASSED=0
fi
# ---- 2. 内存稳定性fnscript: sudo memtester <可用内存97%> 1数据校验----
FN_MEM_STATUS="skip"
FN_MEM_NOTE="未启用AUTOCHECK_FN_MEMORY=0"
if [ "$FN_MEMORY" = "1" ]; then
FN_MEM_NOTE=""
set +e
FN_MEM_OUT=$(printf 'y\ny\n' | python3 "$FN_SCRIPT" --memory 2>&1)
set -e
if printf '%s' "$FN_MEM_OUT" | grep -q "未发现问题"; then
FN_MEM_STATUS="pass"
else
FN_MEM_STATUS="fail"
FN_ALL_PASSED=0
fi
fi
END_TIME=$(date +%s%N)
DURATION_MS=$(( (END_TIME - START_TIME) / 1000000 ))
# ---- 3. 汇总结果(门禁语义:任一子项失败即整体失败)----
if [ "$FN_ALL_PASSED" -eq 1 ]; then
echo 'status:pass'
else
echo 'status:fail'
fi
echo "duration:${DURATION_MS}ms"
echo "output:[CPU] ${FN_CPU_STATUS} (stress ${FN_DURATION}s)"
printf '%s\n' "$FN_CPU_OUT" | while IFS= read -r line; do
[ -n "$line" ] && echo "output: $line"
done
if [ "$FN_MEM_STATUS" = "skip" ]; then
echo "output:[MEM] skip - ${FN_MEM_NOTE}"
else
echo "output:[MEM] ${FN_MEM_STATUS} (memtester 可用内存97% 单遍校验)"
printf '%s\n' "$FN_MEM_OUT" | while IFS= read -r line; do
[ -n "$line" ] && echo "output: $line"
done
fi
echo ''
# 供 run.sh 门禁判断fnstable 失败 → 跳过自有压测)
GATE_PASSED=$FN_ALL_PASSED

View File

@@ -28,7 +28,7 @@ source /scripts/temp_start.sh
# 未知类型输出 skip保证报告完整
for T in $(echo "$AUTOCHECK_TYPES" | tr ',' ' '); do
case "$T" in
cpu|memory|disk|memnative|full) ;;
cpu|memory|disk|memnative|full|fnstable) ;;
*)
echo "===TEST:$T==="
echo "status:skip"
@@ -38,16 +38,37 @@ for T in $(echo "$AUTOCHECK_TYPES" | tr ',' ' '); do
esac
done
# 3. 按固定顺序执行请求的测试(先单项、综合 full 最后
# 3. 稳定性门禁fnscript 社区脚本先行,不通过则跳过自有压测
# fnstable.sh 内部设置 GATE_PASSED1=通过 / 0=失败)
GATE_PASSED=1
REQ_TYPES=",${AUTOCHECK_TYPES// /},"
for TYPE in cpu memory disk memnative full; do
case "$REQ_TYPES" in
*",$TYPE,"*) source "/scripts/$TYPE.sh" ;;
esac
done
case "$REQ_TYPES" in
*,fnstable,*) source /scripts/fnstable.sh ;;
esac
# 4. 温度监控收尾
# 4. 按固定顺序执行请求的自有测试(先单项、综合 full 最后)
# 门禁未通过 → 硬件已有问题,自有压测全部 skip报告仍完整输出
if [ "$GATE_PASSED" -eq 1 ]; then
for TYPE in cpu memory disk memnative full; do
case "$REQ_TYPES" in
*",$TYPE,"*) source "/scripts/$TYPE.sh" ;;
esac
done
else
for TYPE in cpu memory disk memnative full; do
case "$REQ_TYPES" in
*",$TYPE,"*)
echo "===TEST:$TYPE==="
echo "status:skip"
echo "output:fnscript 稳定性门禁未通过,跳过自有压测"
echo ""
;;
esac
done
fi
# 5. 温度监控收尾
source /scripts/temp_end.sh
# 5. 内核日志 + 结束标记
# 6. 内核日志 + 结束标记
source /scripts/dmesg.sh

View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 又菜又爱玩的小朱猪
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

View File

@@ -0,0 +1,54 @@
# 2.0版本重磅来袭
### ✅脚本菜单改进
![image](https://github.com/user-attachments/assets/efed14a9-c2a1-475e-8c67-cdd42dc0c95c) \
![image](https://github.com/user-attachments/assets/66b43e51-0eb5-41b8-8770-6484e953afda)
### 🖱TUI交互式界面感觉就像在使用Windows
![output](https://github.com/user-attachments/assets/f0a63758-8174-408d-94a0-bbe637bd1bd2)
### 终端绑定功能进入root自动运行脚本
![output](https://github.com/user-attachments/assets/f330ba45-3a7d-4bbf-bd0a-84530d0e1663)
### ⬇git下载和更新
✅ 操作简单维护更新方便采用Gitee国内源确保全国地区均可访问
### 📄独立脚本文件
✅继续沿用1.0脚本分开管理,便于维护
### MIT最宽松的开源协议
# 如何使用
## TUI交互式运行
所有脚本目前均已实现图形化TUI交互式操作您可以直接使用一键使用命令后根据菜单提示选择对应功能默认使用的是TUI。由于脚本TUI采用的是 `textual`这个库无法通过apt进行安装和管理所以必须强行通过pip来进行安装曾考虑过进行打包后执行但部分脚本可能存在打包后运行问题
[点击这里查阅详细冲突风险以及解决方案](https://github.com/yxsj245/fnscript/blob/2.0/pip%E5%BA%93%E5%86%B2%E7%AA%81%E9%A3%8E%E9%99%A9.md)
### 使用运行命令
```bash
git clone https://gitee.com/xiao-zhu245/fnscript.git
cd fnscript/
python3 menu.py
```
### 建议使用 1panel面板终端或者CMD终端进行TUI操作其它软件的终端可能存在兼容问题。使用脚本建议进入root用户下运行。
## 传统命令行运行
若您担心pip库冲突风险或无法使用交互式终端情况下我将部分功能单一的脚本增加了传统通过命令行形式运行的功能在无需安装`textual`库就可以直接运行使用
### 如何使用
您可以通过此命令拉取任意脚本后直接使用
```bash
python3 此处替换脚本名称 -h #获取参数信息
```
# 脚本信息
| 脚本名称 | 作用 | 是否支持TUI | 是否支持命令行 |
| ------- | ------- | ------- | ------- |
| cdmount.py | 挂载CD/DVD |✅|❎|
| filestorage.py | 分析目录文件大小 |✅|❎|
| Powermanagement.py | 电源管理 |✅|✅|
| qcowtools.py | qcow2转换工具 |✅|✅|
| self_inspection.py | 硬件压测 |✅|✅|
| startIOMMU.py | 开启IOMMU直通 |✅|✅|
| swapinfo.py | 管理swap |✅|✅|
| vmtoolses.py | 安装虚拟机工具 |✅|✅|
| WOLstart.py | 开启WOL网络唤醒 |✅|✅|
| network_diagnostic_tool.py | 网络诊断与修改 |✅|✅|
| raid_repair_tui.py | 阵列状态和修复 |✅|❎|
| ffmpeg_converter_tui.py | 影音万能格式转换 |✅|❎|
# 其它合作脚本
[七月七夕纯命令行传统脚本](https://github.com/qiyueqixi/fnos)

View File

@@ -0,0 +1,757 @@
#!/usr/bin/env python
import os
import subprocess
import psutil
import time
import signal
import sys
import argparse
# 尝试导入textual库如果不存在则设置标志
HAS_TEXTUAL = True
try:
from textual.app import App, ComposeResult
from textual.widgets import Header, Footer, Static, Button, Label, Input
from textual.containers import Container, Horizontal, Vertical
from textual.screen import Screen, ModalScreen
import asyncio
except ImportError:
HAS_TEXTUAL = False
# 如果没有textual但尝试无参数运行我们需要提示用户
if len(sys.argv) == 1:
print("错误: 未安装textual库无法启动图形界面。")
print("您可以通过命令行参数使用此脚本的核心功能:")
print(" 内存测试: python self_inspection.py --memory")
print(" CPU测试: python self_inspection.py --cpu [测试时长(秒)]")
print(" 查看帮助: python self_inspection.py -h")
sys.exit(1)
# 定义颜色常量(保留以兼容原始代码)
RESET = "\033[0m"
BOLD = "\033[1m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
BLUE = "\033[34m"
MAGENTA = "\033[35m"
CYAN = "\033[36m"
WHITE = "\033[37m"
# 全局变量存储当前运行的进程
current_process = None
# 工具函数
def clear_screen():
"""清空终端屏幕"""
os.system('clear' if os.name == 'posix' else 'cls')
def get_test_memory():
"""计算可用于测试的内存 (剩余内存的 97%)"""
available_memory = psutil.virtual_memory().available // (1024 * 1024) # 获取可用内存 (MB)
return int(available_memory * 0.97) # 取 97% 进行测试
def check_memory_usage():
"""检查当前内存使用情况"""
return psutil.virtual_memory().percent
def run_command(command):
"""执行命令并返回结果"""
global current_process
result = {
'status_code': 0,
'stdout': '',
'stderr': ''
}
try:
# 保存进程引用以便后续可以终止
current_process = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1
)
# 等待进程完成
stdout, stderr = current_process.communicate()
# 获取返回码
result['status_code'] = current_process.returncode
result['stdout'] = stdout
result['stderr'] = stderr
# 清除当前进程引用
current_process = None
return result
except Exception as e:
if current_process:
try:
current_process.terminate()
except:
pass
current_process = None
result['status_code'] = 1
result['stderr'] = str(e)
return result
def stop_current_process():
"""停止当前正在运行的进程"""
global current_process
if current_process:
try:
current_process.terminate()
# 给进程一点时间来终止
time.sleep(0.5)
# 如果进程还在运行,强制结束
if current_process.poll() is None:
current_process.kill()
return True
except Exception:
return False
finally:
current_process = None
return False
# 命令行直接执行内存测试的函数
def cli_memory_test():
"""通过命令行直接执行内存测试"""
memory_usage = check_memory_usage()
if memory_usage > 80:
print(f"警告: 当前系统内存使用率已达 {memory_usage}%!")
print("建议关闭一些软件后再运行,继续运行可能检测效果不够准确")
print("是否继续? (y/n)")
response = input().strip().lower()
if response != 'y':
print("操作已取消")
return False
print("建议关闭所有docker容器以及停用所有应用程序以保证检测的准确性。")
print("是否继续? (y/n)")
response = input().strip().lower()
if response != 'y':
print("操作已取消")
return False
mem_size = f"{get_test_memory()}M"
print("正在运行内存测试,请勿进行其它操作,如果运行过程中出现死机重启则大概率为内存问题。")
print("整个测试过程预计 30-120 分钟 具体根据内存大小如果需要终止测试请按Ctrl+C。")
try:
# 设置信号处理器以便可以通过Ctrl+C终止测试
def signal_handler(sig, frame):
print("\n测试已被用户终止")
stop_current_process()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
# 执行测试
result = run_command(["sudo", "memtester", mem_size, "1"])
if result['status_code'] == 0:
print("内存测试完成!未发现问题。")
return True
else:
print(f"内存测试失败!错误信息:{result['stderr']}")
return False
except Exception as e:
print(f"测试过程中发生错误: {str(e)}")
return False
# 命令行直接执行CPU测试的函数
def cli_cpu_test(duration=60):
"""通过命令行直接执行CPU测试
Args:
duration (int): 测试持续时间默认为60秒
"""
try:
duration = int(duration)
if duration <= 0:
print("错误: 测试时长必须大于0")
return False
except ValueError:
print("错误: 测试时长必须是一个有效的整数")
return False
print(f"正在进行 CPU 压测,持续 {duration} 秒...如需提前结束测试请按Ctrl+C。")
try:
# 设置信号处理器以便可以通过Ctrl+C终止测试
def signal_handler(sig, frame):
print("\n测试已被用户终止")
stop_current_process()
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
# 执行测试
result = run_command(["stress", "--cpu", "4", "--timeout", str(duration)])
if result['status_code'] == 0:
print("CPU 压测完成!未发现问题。")
return True
else:
print(f"CPU 压测失败!错误信息:{result['stderr']}")
return False
except Exception as e:
print(f"测试过程中发生错误: {str(e)}")
return False
def parse_arguments():
"""解析命令行参数"""
parser = argparse.ArgumentParser(description='系统健康检测工具')
parser.add_argument('--memory', action='store_true',
help='执行内存测试')
parser.add_argument('--cpu', nargs='?', const=60, type=int,
help='执行CPU压测可选参数为测试时长(秒)默认为60秒')
return parser.parse_args()
# 只有在导入了textual库的情况下才定义这些类
if HAS_TEXTUAL:
# 确认对话框
class ConfirmDialog(ModalScreen):
"""确认对话框"""
def __init__(self, message: str, action_callback=None):
super().__init__()
self.message = message
self.action_callback = action_callback
def compose(self) -> ComposeResult:
with Container(id="dialog-container"):
yield Label(f"[yellow]{self.message}[/yellow]", id="dialog-title")
with Horizontal(id="dialog-buttons"):
yield Button("确认", id="confirm", variant="primary")
yield Button("取消", id="cancel", variant="error")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "confirm" and self.action_callback:
self.dismiss(True)
self.action_callback()
else:
self.dismiss(False)
# 成功提示对话框
class SuccessDialog(ModalScreen):
"""成功提示对话框"""
def __init__(self, message: str):
super().__init__()
self.message = message
def compose(self) -> ComposeResult:
with Container(id="dialog-container"):
yield Label(f"[green]{self.message}[/green]", id="dialog-title")
with Horizontal(id="dialog-buttons"):
yield Button("确定", id="ok", variant="primary")
def on_button_pressed(self, event: Button.Pressed) -> None:
self.dismiss(True)
# 错误提示对话框
class ErrorDialog(ModalScreen):
"""错误提示对话框"""
def __init__(self, message: str):
super().__init__()
self.message = message
def compose(self) -> ComposeResult:
with Container(id="dialog-container"):
yield Label(f"[red]{self.message}[/red]", id="dialog-title")
with Horizontal(id="dialog-buttons"):
yield Button("确定", id="ok", variant="primary")
def on_button_pressed(self, event: Button.Pressed) -> None:
self.dismiss(True)
# 输入对话框
class InputDialog(ModalScreen):
"""输入对话框"""
def __init__(self, message: str, default_value: str = "", action_callback=None):
super().__init__()
self.message = message
self.default_value = default_value
self.action_callback = action_callback
def compose(self) -> ComposeResult:
with Container(id="dialog-container"):
yield Label(f"[blue]{self.message}[/blue]", id="dialog-title")
yield Input(value=self.default_value, id="input-field")
with Horizontal(id="dialog-buttons"):
yield Button("确认", id="confirm", variant="primary")
yield Button("取消", id="cancel", variant="error")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "confirm" and self.action_callback:
input_value = self.query_one("#input-field", Input).value
self.dismiss(input_value)
self.action_callback(input_value)
else:
self.dismiss(None)
class SystemHealthApp(App):
CSS = """
Screen {
background: $surface;
}
#title {
dock: top;
text-align: center;
text-style: bold;
background: $accent;
color: $text;
padding: 1;
margin-bottom: 1;
}
#main-container {
width: 100%;
height: auto;
padding: 1;
}
#menu-container {
width: 100%;
height: auto;
margin: 1;
border: solid $primary;
padding: 1;
}
.menu-button {
margin: 1;
min-width: 30;
min-height: 3;
background: $primary;
color: $text;
border: tall $primary-lighten-2;
}
#memory-test {
background: $success-darken-1;
color: $text;
border: tall $success;
}
#cpu-test {
background: $warning-darken-1;
color: $text;
border: tall $warning;
}
#stop-test {
background: $error-darken-1;
color: $text;
border: tall $error;
margin: 1;
min-width: 30;
min-height: 3;
}
#status {
height: auto;
min-height: 3;
margin: 1;
padding: 1;
border: solid $accent;
background: $surface-darken-1;
}
/* 对话框样式 */
#dialog-container {
width: 60%;
height: auto;
padding: 2;
background: $surface;
border: thick $accent;
margin: 1 0;
}
/* 成功对话框样式 */
SuccessDialog #dialog-container {
border: thick $success;
}
/* 错误对话框样式 */
ErrorDialog #dialog-container {
border: thick $error;
}
/* 确认对话框样式 */
ConfirmDialog #dialog-container {
border: thick $warning;
}
/* 输入对话框样式 */
InputDialog #dialog-container {
border: thick $primary;
}
#dialog-title {
text-align: center;
width: 100%;
margin-bottom: 2;
}
#input-field {
margin-bottom: 1;
}
#dialog-buttons {
align: center middle;
width: 100%;
}
#dialog-buttons Button {
margin: 0 2;
min-width: 10;
}
/* 按钮样式增强 */
Button {
background: $primary;
border: tall $primary-lighten-2;
padding: 1 2;
}
Button:hover {
background: $primary-lighten-1;
}
Button#confirm {
background: $success-darken-1;
border: tall $success;
}
Button#confirm:hover {
background: $success;
}
Button#cancel, Button#stop-test {
background: $error-darken-1;
border: tall $error;
}
Button#cancel:hover, Button#stop-test:hover {
background: $error;
}
Button#ok {
background: $primary-darken-1;
border: tall $primary;
}
Button#ok:hover {
background: $primary;
}
"""
BINDINGS = [
("q", "quit", "退出"),
("escape", "quit", "退出"),
("ctrl+c", "stop_test", "停止测试"),
]
def __init__(self):
super().__init__()
self.is_testing = False
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
yield Footer()
with Container(id="main-container"):
yield Label("[b]系统健康检测工具[/b]", id="title")
with Container(id="menu-container"):
yield Button("内存检测 (使用 memtester)", id="memory-test", classes="menu-button")
yield Button("CPU 压测", id="cpu-test", classes="menu-button")
yield Button("停止当前测试", id="stop-test", disabled=True)
# 状态输出
yield Static("准备就绪,请选择检测项目。", id="status")
def on_button_pressed(self, event: Button.Pressed) -> None:
"""按钮点击事件处理"""
button_id = event.button.id
if button_id == "memory-test":
self.start_memory_test()
elif button_id == "cpu-test":
self.start_cpu_test()
elif button_id == "stop-test":
self.stop_test()
def action_stop_test(self) -> None:
"""键盘快捷键停止测试"""
self.stop_test()
def stop_test(self) -> None:
"""停止当前测试"""
if not self.is_testing:
return
status = self.query_one("#status")
status.update("[yellow]正在停止测试...[/yellow]")
# 停止当前进程
if stop_current_process():
status.update("[green]测试已停止。[/green]")
self.show_success_dialog("测试已成功停止。")
else:
status.update("[red]无法停止测试,可能测试已经完成。[/red]")
# 更新界面状态
self.is_testing = False
# 使用try/except捕获可能的NoMatches异常
try:
stop_button = self.query_one("#stop-test", Button)
if stop_button:
stop_button.disabled = True
memory_button = self.query_one("#memory-test", Button)
if memory_button:
memory_button.disabled = False
cpu_button = self.query_one("#cpu-test", Button)
if cpu_button:
cpu_button.disabled = False
except Exception as e:
# 记录错误但不中断程序
print(f"更新按钮状态时出错: {str(e)}")
def start_memory_test(self) -> None:
"""开始内存测试"""
status = self.query_one("#status")
memory_usage = check_memory_usage()
if memory_usage > 80:
warning_message = f"警告: 当前系统内存使用率已达 {memory_usage}%!\n建议关闭一些软件后再运行,继续运行可能检测效果不够准确"
status.update(f"[yellow]{warning_message}[/yellow]")
# 显示确认对话框
def confirm_callback():
self.show_memory_test_warning()
self.push_screen(ConfirmDialog(warning_message, confirm_callback))
else:
self.show_memory_test_warning()
def show_memory_test_warning(self):
"""显示内存测试警告"""
warning = "建议关闭所有docker容器以及停用所有应用程序以保证检测的准确性。"
status = self.query_one("#status")
status.update(f"[red]{warning}[/red]")
def confirm_callback():
self.run_memory_test()
self.push_screen(ConfirmDialog(warning, confirm_callback))
def run_memory_test(self) -> None:
"""执行内存测试"""
status = self.query_one("#status")
mem_size = f"{get_test_memory()}M"
message = "正在运行内存测试,请勿进行其它操作,如果运行过程中出现死机重启则大概率为内存问题。整个测试过程预计 30-120 分钟 具体根据内存大小如果需要终止测试请按Ctrl+C或点击停止按钮。"
status.update(f"[blue]{message}[/blue]")
# 更新界面状态
self.is_testing = True
# 使用try/except捕获可能的NoMatches异常
try:
stop_button = self.query_one("#stop-test", Button)
if stop_button:
stop_button.disabled = False
memory_button = self.query_one("#memory-test", Button)
if memory_button:
memory_button.disabled = True
cpu_button = self.query_one("#cpu-test", Button)
if cpu_button:
cpu_button.disabled = True
except Exception as e:
# 记录错误但不中断程序
print(f"更新按钮状态时出错: {str(e)}")
# 创建异步任务
asyncio.create_task(self.do_memory_test(mem_size))
async def do_memory_test(self, mem_size):
"""异步执行内存测试"""
status = self.query_one("#status")
try:
# 在异步中执行测试
result = await asyncio.to_thread(run_command, ["sudo", "memtester", mem_size, "1"])
if result['status_code'] == 0:
success_message = "内存测试完成!未发现问题。"
status.update(f"[green]{success_message}[/green]")
self.show_success_dialog(success_message)
else:
# 检查是否是被用户手动终止的
if current_process is None and not result['stderr']:
status.update("[yellow]测试已被用户终止。[/yellow]")
else:
error_message = f"内存测试失败!错误信息:{result['stderr']}"
status.update(f"[red]{error_message}[/red]")
self.show_error_dialog(error_message)
except Exception as e:
error_message = f"测试过程中发生错误: {str(e)}"
status.update(f"[red]{error_message}[/red]")
self.show_error_dialog(error_message)
finally:
# 恢复界面状态
self.is_testing = False
# 使用try/except捕获可能的NoMatches异常
try:
stop_button = self.query_one("#stop-test", Button)
if stop_button:
stop_button.disabled = True
memory_button = self.query_one("#memory-test", Button)
if memory_button:
memory_button.disabled = False
cpu_button = self.query_one("#cpu-test", Button)
if cpu_button:
cpu_button.disabled = False
except Exception as e:
# 记录错误但不中断程序
print(f"更新按钮状态时出错: {str(e)}")
def start_cpu_test(self) -> None:
"""开始CPU测试"""
def input_callback(duration):
if duration and duration.isdigit() and int(duration) > 0:
self.run_cpu_test(duration)
else:
self.show_error_dialog("请输入有效的时间(秒)!")
self.push_screen(InputDialog("请输入 CPU 压测时长 (秒):", "60", input_callback))
def run_cpu_test(self, duration):
"""执行CPU测试"""
status = self.query_one("#status")
message = f"正在进行 CPU 压测,持续 {duration} 秒...如需提前结束测试请按Ctrl+C或点击停止按钮。"
status.update(f"[blue]{message}[/blue]")
# 更新界面状态
self.is_testing = True
# 使用try/except捕获可能的NoMatches异常
try:
stop_button = self.query_one("#stop-test", Button)
if stop_button:
stop_button.disabled = False
memory_button = self.query_one("#memory-test", Button)
if memory_button:
memory_button.disabled = True
cpu_button = self.query_one("#cpu-test", Button)
if cpu_button:
cpu_button.disabled = True
except Exception as e:
# 记录错误但不中断程序
print(f"更新按钮状态时出错: {str(e)}")
# 创建异步任务
asyncio.create_task(self.do_cpu_test(duration))
async def do_cpu_test(self, duration):
"""异步执行CPU测试"""
status = self.query_one("#status")
try:
# 在异步中执行测试
result = await asyncio.to_thread(run_command, ["stress", "--cpu", "4", "--timeout", duration])
if result['status_code'] == 0:
success_message = "CPU 压测完成!未发现问题。"
status.update(f"[green]{success_message}[/green]")
self.show_success_dialog(success_message)
else:
# 检查是否是被用户手动终止的
if current_process is None and not result['stderr']:
status.update("[yellow]测试已被用户终止。[/yellow]")
else:
error_message = f"CPU 压测失败!错误信息:{result['stderr']}"
status.update(f"[red]{error_message}[/red]")
self.show_error_dialog(error_message)
except Exception as e:
error_message = f"测试过程中发生错误: {str(e)}"
status.update(f"[red]{error_message}[/red]")
self.show_error_dialog(error_message)
finally:
# 恢复界面状态
self.is_testing = False
# 使用try/except捕获可能的NoMatches异常
try:
stop_button = self.query_one("#stop-test", Button)
if stop_button:
stop_button.disabled = True
memory_button = self.query_one("#memory-test", Button)
if memory_button:
memory_button.disabled = False
cpu_button = self.query_one("#cpu-test", Button)
if cpu_button:
cpu_button.disabled = False
except Exception as e:
# 记录错误但不中断程序
print(f"更新按钮状态时出错: {str(e)}")
def show_success_dialog(self, message: str) -> None:
"""显示成功对话框"""
self.push_screen(SuccessDialog(message))
def show_error_dialog(self, message: str) -> None:
"""显示错误对话框"""
self.push_screen(ErrorDialog(message))
if __name__ == "__main__":
args = parse_arguments()
# 如果指定了命令行参数,则直接执行相应功能
if args.memory:
cli_memory_test()
sys.exit(0)
if args.cpu is not None:
cli_cpu_test(args.cpu)
sys.exit(0)
# 如果没有指定命令行参数则启动图形界面如果textual可用
if HAS_TEXTUAL:
app = SystemHealthApp()
app.run()
else:
# 这种情况不应该发生,因为在导入时已经处理了,但为了代码完整性保留
print("错误: 未安装textual库无法启动图形界面。")
print("您可以通过命令行参数使用此脚本的核心功能:")
print(" 内存测试: python self_inspection.py --memory")
print(" CPU测试: python self_inspection.py --cpu [测试时长(秒)]")
print(" 查看帮助: python self_inspection.py -h")
sys.exit(1)