mirror of
https://gitee.com/xmz7891/ykxmz.git
synced 2026-10-18 17:10:43 +00:00
xmz7891
b62ac865ea
被控端 - 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%） - 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败） - 每分钟回头找局域网，自动切回快 400 倍的路径 - 画面心跳帧：静止时省流量，客户端仍知道流是活的 - 锁屏时给出可操作的提示 手表端 - 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException - 连接错误翻译成人话（证书/时间/解析/超时分别提示） - 经中转连上后用服务器报的地址刷新已配对记录 - 新增局域网连接页面、服务器延迟检测 服务器端 - 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502） - 网络调优脚本 net-tune.sh（开机自动应用） - 管理页：批量删除、网络体检、下载区 - 隧道改 HTTP/2 + 双连接 文档 - 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
3609 lines
154 KiB
Python
3609 lines
154 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""电脑端功能注册表。
|
||
|
||
为什么用注册表而不是一个功能一条协议命令：
|
||
功能会越加越多，每加一个就改协议、改安卓端、改测试，成本太高。
|
||
这里做成"电脑端自己声明有哪些功能"，手表端只是拉列表 + 按名字调用。
|
||
以后加功能只改这一个文件。
|
||
|
||
功能分三类：
|
||
toggle 开关型（夜间模式、任务栏隐藏…）
|
||
action 执行型（清空回收站、锁屏+关屏…）
|
||
value 带数值（音量、亮度、鼠标速度…）
|
||
|
||
安全边界：
|
||
文件类操作一律**限定在用户自己的目录内**（家目录下的桌面/文档/下载/图片…）。
|
||
这是"远程控制"和"任意文件读写后门"的分界线，不会解除。
|
||
但允许的目录、允许的程序、允许的命令都可以**在电脑端设置里自己加**。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import ctypes
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
|
||
IS_WINDOWS = os.name == "nt"
|
||
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
|
||
# 文件操作允许的根目录（可在家目录内自由扩展）
|
||
DEFAULT_ROOTS = ("Desktop", "Documents", "Downloads", "Pictures", "Music", "Videos")
|
||
|
||
|
||
def _run(cmd: list[str], timeout: int = 20, shell: bool = False) -> tuple[int, str]:
|
||
try:
|
||
proc = subprocess.run(
|
||
cmd, capture_output=True, timeout=timeout, shell=shell,
|
||
creationflags=_NO_WINDOW,
|
||
encoding="mbcs" if IS_WINDOWS else "utf-8", errors="replace")
|
||
return proc.returncode, ((proc.stdout or "") + (proc.stderr or "")).strip()
|
||
except Exception as exc: # noqa: BLE001
|
||
return -1, str(exc)
|
||
|
||
|
||
class Features:
|
||
"""所有"电脑端能做的小事"都从这里走。"""
|
||
|
||
def __init__(self, actions):
|
||
self.actions = actions
|
||
self.store = actions.store
|
||
self._notes: list[dict] = []
|
||
self._todos: list[dict] = []
|
||
self._clip_fav: list[str] = []
|
||
self._timers: list = []
|
||
|
||
# ==================================================================
|
||
# 设置（可在界面里改的那些）
|
||
# ==================================================================
|
||
|
||
def extra_roots(self) -> list[Path]:
|
||
"""用户额外允许的目录。"""
|
||
raw = str(self.store.get("file_roots") or "").strip()
|
||
out: list[Path] = []
|
||
for piece in raw.split(";"):
|
||
piece = piece.strip()
|
||
if piece:
|
||
path = Path(piece)
|
||
if path.is_dir():
|
||
out.append(path)
|
||
return out
|
||
|
||
def allowed_roots(self) -> list[Path]:
|
||
home = Path.home()
|
||
roots = [home / name for name in DEFAULT_ROOTS if (home / name).is_dir()]
|
||
roots.append(home)
|
||
roots.extend(self.extra_roots())
|
||
return roots
|
||
|
||
def sensitive_dirs(self) -> list:
|
||
"""程序自己的数据目录 —— 里面有配对码和设备令牌，任何情况下都不给碰。"""
|
||
out = []
|
||
try:
|
||
out.append(Path(self.store.dir))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
from .config import app_dir
|
||
out.append(Path(app_dir()))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return out
|
||
|
||
def is_sensitive(self, path: Path) -> bool:
|
||
try:
|
||
resolved = path.resolve()
|
||
except OSError:
|
||
return False
|
||
|
||
# 接收目录要豁免！它就在程序目录里面（app_dir()/received），
|
||
# 如果不豁免，每个要拉回手表的文件都会被当成"敏感目录"挡掉，
|
||
# 表现为「接收文件」永远拿不到东西、报"文件不存在"。
|
||
try:
|
||
received = Path(self.store.received_dir()).resolve()
|
||
resolved.relative_to(received)
|
||
return False
|
||
except (ValueError, OSError):
|
||
pass
|
||
|
||
for folder in self.sensitive_dirs():
|
||
try:
|
||
folder = folder.resolve()
|
||
except OSError:
|
||
continue
|
||
if resolved == folder or folder in resolved.parents:
|
||
return True
|
||
return False
|
||
|
||
def assert_not_root(self, path: Path) -> None:
|
||
"""不允许对「允许目录」本身动手。
|
||
|
||
否则一条 rename_file(path="Downloads", name="x") 就能把整个
|
||
下载目录改名（delete_file / move_file 同理，能直接删掉）。
|
||
"""
|
||
try:
|
||
resolved = path.resolve()
|
||
except OSError:
|
||
return
|
||
for root in self.allowed_roots():
|
||
try:
|
||
if resolved == root.resolve():
|
||
raise ValueError("不能对允许目录本身做这个操作")
|
||
except OSError:
|
||
continue
|
||
|
||
def resolve_in_roots(self, raw: str) -> Path:
|
||
"""把用户给的路径解析成一个允许访问的绝对路径。"""
|
||
text = (raw or "").strip().strip('"')
|
||
if not text:
|
||
raise ValueError("路径为空")
|
||
candidate = Path(text)
|
||
if not candidate.is_absolute():
|
||
# 相对路径：先按家目录试（Downloads、Desktop 这种直接写得通），
|
||
# 不存在再退到"下载目录里的相对路径"
|
||
home = Path.home()
|
||
direct = home / text
|
||
candidate = direct if direct.exists() else home / "Downloads" / text
|
||
try:
|
||
resolved = candidate.resolve()
|
||
except OSError as exc:
|
||
raise ValueError(f"路径不合法: {exc}") from exc
|
||
|
||
if self.is_sensitive(resolved):
|
||
raise ValueError("这是程序自己的数据目录，不允许访问")
|
||
|
||
for root in self.allowed_roots():
|
||
try:
|
||
resolved.relative_to(root.resolve())
|
||
return resolved
|
||
except ValueError:
|
||
continue
|
||
raise ValueError(f"只允许访问自己的目录（{resolved} 不在允许范围内）")
|
||
|
||
def extra_apps(self) -> dict:
|
||
"""用户自定义的快捷程序。格式：显示名=程序路径;显示名2=路径2"""
|
||
raw = str(self.store.get("extra_apps") or "").strip()
|
||
out: dict = {}
|
||
for piece in raw.split(";"):
|
||
if "=" not in piece:
|
||
continue
|
||
label, _, target = piece.partition("=")
|
||
label = label.strip()
|
||
target = target.strip()
|
||
if label and target:
|
||
out[label] = target
|
||
return out
|
||
|
||
def extra_commands(self) -> dict:
|
||
"""用户自定义的命令别名。手表只能选这些，不能随便执行。"""
|
||
raw = str(self.store.get("extra_commands") or "").strip()
|
||
out: dict = {}
|
||
for piece in raw.split(";"):
|
||
if "=" not in piece:
|
||
continue
|
||
label, _, command = piece.partition("=")
|
||
label = label.strip()
|
||
command = command.strip()
|
||
if label and command:
|
||
out[label] = command
|
||
return out
|
||
|
||
# ==================================================================
|
||
# 功能声明
|
||
# ==================================================================
|
||
|
||
def defs(self) -> list[dict]:
|
||
items: list[dict] = []
|
||
|
||
def add(name, label, kind, group, **extra):
|
||
item = {"name": name, "label": label, "kind": kind, "group": group}
|
||
item.update(extra)
|
||
items.append(item)
|
||
|
||
# ---- 显示 ----
|
||
add("night_mode", "夜间模式", "toggle", "显示", on=self._night_mode())
|
||
add("taskbar_hide", "任务栏自动隐藏", "toggle", "显示", on=self._taskbar_hidden())
|
||
add("brightness", "屏幕亮度", "value", "显示",
|
||
min=10, max=100, value=self._brightness())
|
||
add("mouse_speed", "鼠标速度", "value", "显示",
|
||
min=1, max=20, value=self._mouse_speed())
|
||
add("monitor_on", "唤醒屏幕", "action", "显示")
|
||
|
||
# ---- 音量 ----
|
||
add("mute_all", "全部静音", "action", "音量")
|
||
|
||
# ---- 静默运行 ----
|
||
add("autostart", "开机自启（静默）", "toggle", "静默",
|
||
on=self._autostart_on())
|
||
add("autostart_status", "运行状态", "action", "静默")
|
||
add("quit_agent", "退出被控端", "action", "静默")
|
||
add("silent_help", "静默模式说明", "action", "静默")
|
||
|
||
# ---- 查看（续）----
|
||
add("screen_resolution", "显示器分辨率", "action", "查看")
|
||
add("list_printers", "打印机列表", "action", "查看")
|
||
add("list_drives", "驱动器详情", "action", "查看")
|
||
add("audio_devices", "音频设备", "action", "查看")
|
||
add("usb_devices", "USB 设备", "action", "查看")
|
||
add("installed_drivers", "已装驱动", "action", "查看")
|
||
add("windows_detail", "Windows 版本详情", "action", "查看")
|
||
add("recycle_size", "回收站大小", "action", "查看")
|
||
add("temp_size", "临时文件大小", "action", "查看")
|
||
add("certificates", "系统证书", "action", "查看")
|
||
|
||
# ---- 网络（续）----
|
||
add("open_ports", "本机监听端口", "action", "网络")
|
||
add("firewall_status", "防火墙状态", "action", "网络")
|
||
add("dns_servers", "DNS 服务器", "action", "网络")
|
||
add("hosts_file", "查看 hosts", "action", "网络")
|
||
add("wifi_networks", "附近 WiFi", "action", "网络")
|
||
add("bluetooth_devices", "蓝牙设备", "action", "网络")
|
||
|
||
# ---- 系统（续）----
|
||
add("open_services", "打开服务", "action", "系统")
|
||
add("open_task_scheduler", "打开任务计划", "action", "系统")
|
||
add("open_device_manager", "打开设备管理器", "action", "系统")
|
||
add("open_disk_cleanup", "打开磁盘清理", "action", "系统")
|
||
add("open_windows_update", "打开系统更新", "action", "系统")
|
||
add("clear_temp", "清理临时文件", "action", "系统")
|
||
add("env_vars", "环境变量", "action", "系统")
|
||
add("volume_up", "音量 +5", "action", "音量")
|
||
add("volume_down", "音量 -5", "action", "音量")
|
||
|
||
# ---- 系统 ----
|
||
add("lock_and_off", "锁屏 + 关屏", "action", "系统")
|
||
add("empty_recycle", "清空回收站", "action", "系统")
|
||
add("flush_dns", "刷新 DNS", "action", "系统")
|
||
add("restart_explorer", "重启资源管理器", "action", "系统")
|
||
add("clear_clip", "清空电脑剪贴板", "action", "系统")
|
||
|
||
# ---- 查看 ----
|
||
add("uptime", "开机时长", "action", "查看")
|
||
add("perf", "性能（CPU/内存）", "action", "查看")
|
||
add("disk", "磁盘占用", "action", "查看")
|
||
add("battery", "电量", "action", "查看")
|
||
add("network_speed", "网络速度", "action", "查看")
|
||
add("installed_apps", "已安装软件", "action", "查看")
|
||
add("top_services", "运行中的服务", "action", "查看")
|
||
add("recent_files", "最近打开的文件", "action", "查看")
|
||
add("event_errors", "系统错误日志", "action", "查看")
|
||
add("temperature", "温度 / 负载", "action", "查看")
|
||
|
||
# ---- 效率 ----
|
||
add("note_add", "记一条笔记", "input", "效率", hint="内容", key="text")
|
||
add("note_list", "看所有笔记", "action", "效率")
|
||
add("note_clear", "清空笔记", "action", "效率")
|
||
add("todo_add", "加一条待办", "input", "效率", hint="内容", key="text")
|
||
add("todo_list", "看所有待办", "action", "效率")
|
||
add("reminder", "定时提醒", "input", "效率",
|
||
hint="几分钟后", key="minutes", hint2="提醒内容", key2="text")
|
||
add("pomodoro", "番茄钟 25 分钟", "action", "效率")
|
||
add("clip_fav_add", "收藏当前剪贴板", "action", "效率")
|
||
add("clip_fav_list", "剪贴板收藏夹", "action", "效率")
|
||
|
||
# ---- 教学（老师的刚需）----
|
||
add("big_text", "全屏大字", "input", "教学",
|
||
hint="要显示的文字", key="text")
|
||
add("big_text_clear", "关掉全屏内容", "action", "教学")
|
||
add("close_visuals", "关掉画面", "action", "显示")
|
||
add("countdown_screen", "全屏倒计时", "input", "教学",
|
||
hint="几分钟", key="minutes")
|
||
add("roll_call", "随机点名", "action", "教学")
|
||
add("roll_call_set", "设置点名名单", "input", "教学",
|
||
hint="名字用逗号隔开", key="names")
|
||
add("whiteboard", "全屏白板", "action", "教学")
|
||
add("display_mode", "投影 / 显示设置", "action", "教学")
|
||
add("big_text_quiet", "全屏显示「安静」", "action", "教学")
|
||
|
||
# ---- 系统控制 ----
|
||
add("minimize_all", "显示桌面", "action", "系统")
|
||
add("task_view", "任务视图", "action", "系统")
|
||
add("close_windows", "关闭所有窗口", "action", "系统")
|
||
add("clock_seconds", "任务栏时钟显示秒", "toggle", "系统",
|
||
on=self._clock_seconds())
|
||
add("hidden_files", "显示隐藏文件", "toggle", "系统",
|
||
on=self._hidden_files())
|
||
add("sleep", "睡眠", "action", "系统")
|
||
add("hibernate", "休眠", "action", "系统")
|
||
add("lock_only", "只锁屏", "action", "系统")
|
||
add("sign_out", "注销当前用户", "action", "系统")
|
||
add("restart_graphics", "重启显卡驱动", "action", "系统")
|
||
|
||
# ---- 网络 ----
|
||
add("wifi_password", "查看 WiFi 密码", "action", "网络")
|
||
add("public_ip", "查公网 IP", "action", "网络")
|
||
add("ping_host", "Ping 一个地址", "input", "网络",
|
||
hint="如 192.168.2.1", key="host")
|
||
add("arp_devices", "局域网设备列表", "action", "网络")
|
||
add("net_adapters", "网卡列表", "action", "网络")
|
||
|
||
# ---- 文件（限定自己的目录）----
|
||
add("list_dir", "浏览文件夹", "input", "文件", hint="如 Downloads", key="path")
|
||
add("download_file", "下载文件到手表", "file", "文件",
|
||
hint="完整路径", key="path")
|
||
add("search_files", "搜索文件", "input", "文件",
|
||
hint="关键词", key="keyword")
|
||
add("make_dir", "新建文件夹", "input", "文件",
|
||
hint="父目录", key="path", hint2="文件夹名", key2="name")
|
||
add("rename_file", "重命名", "input", "文件",
|
||
hint="文件路径", key="path", hint2="新名字", key2="name")
|
||
add("delete_file", "删除到回收站", "input", "文件",
|
||
hint="文件路径", key="path")
|
||
add("open_file", "用默认程序打开", "input", "文件",
|
||
hint="文件路径", key="path")
|
||
add("copy_file", "复制文件", "input", "文件",
|
||
hint="源路径", key="path", hint2="目标路径", key2="to")
|
||
add("move_file", "移动文件", "input", "文件",
|
||
hint="源路径", key="path", hint2="目标路径", key2="to")
|
||
add("zip_dir", "压缩为 zip", "input", "文件",
|
||
hint="文件夹路径", key="path")
|
||
add("unzip", "解压 zip", "input", "文件",
|
||
hint="zip 文件路径", key="path")
|
||
add("file_info", "文件详情", "input", "文件",
|
||
hint="文件路径", key="path")
|
||
add("folder_size", "文件夹大小", "input", "文件",
|
||
hint="文件夹路径", key="path")
|
||
add("open_recycle", "打开回收站", "action", "文件")
|
||
add("disk_by_type", "按类型统计磁盘", "action", "文件")
|
||
|
||
# ---- 工具 ----
|
||
add("password_gen", "生成随机密码", "action", "工具")
|
||
add("stopwatch", "秒表（电脑上显示）", "action", "工具")
|
||
add("world_clock", "世界时钟", "action", "工具")
|
||
add("hardware_info", "硬件信息", "action", "工具")
|
||
add("startup_items", "开机启动项", "action", "工具")
|
||
add("scheduled_tasks", "计划任务", "action", "工具")
|
||
add("system_detail", "系统详细信息", "action", "工具")
|
||
|
||
# ---- 媒体 ----
|
||
add("stop_after", "定时停止播放", "input", "媒体",
|
||
hint="几分钟后停", key="minutes")
|
||
add("fade_volume", "音量淡出到静音", "action", "媒体")
|
||
add("gpu_info", "显卡信息", "action", "媒体")
|
||
|
||
# ---- 教室管控 ----
|
||
add("hide_icons", "隐藏桌面图标", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Explorer"
|
||
"\\Advanced", "HideIcons", 1))
|
||
add("restrict_taskmgr", "禁用任务管理器", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
||
"\\Policies\\System", "DisableTaskMgr", 1))
|
||
add("restrict_regedit", "禁用注册表编辑器", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
||
"\\Policies\\System", "DisableRegistryTools", 1))
|
||
add("restrict_cmd", "禁用命令提示符", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
||
"\\Policies\\System", "DisableCMD", 1))
|
||
add("restrict_wallpaper", "禁止换壁纸", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
||
"\\Policies\\System", "NoDispBackgroundPage", 1))
|
||
add("restrict_power", "藏掉关机按钮", "toggle", "教室", on=self._reg_on(
|
||
"HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
||
"\\Policies\\Explorer", "NoClose", 1))
|
||
add("lock_input", "锁键盘鼠标", "toggle", "教室", on=False)
|
||
add("game_block_common", "屏蔽常见游戏网站", "action", "教室")
|
||
add("site_block", "屏蔽一个网站", "input", "教室",
|
||
hint="如 game.com", key="site")
|
||
add("site_unblock", "解除屏蔽一个网站", "input", "教室",
|
||
hint="如 game.com", key="site")
|
||
add("site_list", "看屏蔽列表", "action", "教室")
|
||
add("site_clear", "清空屏蔽列表", "action", "教室")
|
||
add("kill_games", "结束游戏进程", "action", "教室")
|
||
add("kill_by_name", "按名字结束进程", "input", "教室",
|
||
hint="进程名，如 game.exe", key="name")
|
||
add("bell", "下课铃", "action", "教室")
|
||
add("class_attention", "全屏「请安静」", "action", "教室")
|
||
add("class_start", "上课（锁屏+提示）", "action", "教室")
|
||
add("class_timer", "课堂倒计时", "input", "教室",
|
||
hint="几分钟", key="minutes")
|
||
add("restore_all", "一键恢复全部限制", "action", "教室")
|
||
add("undo_last", "回退上一步", "action", "系统")
|
||
add("undo_all", "回退全部", "action", "系统")
|
||
add("undo_list", "看能回退什么", "action", "系统")
|
||
add("restrict_status", "看当前有哪些限制", "action", "教室")
|
||
|
||
# ---- 教学（续）----
|
||
add("roll_group", "随机分组", "input", "教学",
|
||
hint="几个人一组", key="size")
|
||
add("roll_call_no_repeat", "点名（不重复）", "action", "教学")
|
||
add("roll_call_reset", "重置点名记录", "action", "教学")
|
||
add("class_schedule", "显示课程表", "input", "教学",
|
||
hint="用逗号隔开，如 语文,数学,英语", key="text")
|
||
add("hand_count", "举手计数", "action", "教学")
|
||
add("hand_reset", "举手清零", "action", "教学")
|
||
add("answer_poll", "答题统计", "action", "教学")
|
||
add("magnifier", "屏幕放大镜", "action", "教学")
|
||
add("projector_switch", "切换投影模式", "action", "教学")
|
||
add("lesson_timer", "上课计时", "action", "教学")
|
||
|
||
# ---- 教室（续）----
|
||
add("lock_now", "立刻锁屏", "action", "教室")
|
||
add("game_watch_start", "开始盯游戏进程", "action", "教室")
|
||
add("game_watch_stop", "停止盯游戏进程", "action", "教室")
|
||
add("usb_check", "检查有没有插U盘", "action", "教室")
|
||
add("login_history", "最近登录记录", "action", "教室")
|
||
add("boot_history", "开关机记录", "action", "教室")
|
||
add("clean_desktop", "整理桌面（收进文件夹）", "action", "教室")
|
||
add("block_new_app", "禁止新程序启动", "toggle", "教室", on=False)
|
||
add("show_desktop_now", "立刻显示桌面", "action", "教室")
|
||
add("count_windows", "看目前开了几个窗口", "action", "教室")
|
||
add("class_end_cleanup", "下课清理（关游戏+恢复）", "action", "教室")
|
||
add("who_is_playing", "谁在玩游戏", "action", "教室")
|
||
|
||
# ---- 系统（续）----
|
||
add("open_taskmgr2", "打开任务管理器", "action", "系统")
|
||
add("open_resmon", "打开资源监视器", "action", "系统")
|
||
add("power_plan", "切换电源计划", "action", "系统")
|
||
add("window_top", "当前窗口置顶", "action", "系统")
|
||
add("screen_off_lock", "关屏并锁定", "action", "系统")
|
||
add("clear_all_clip", "清空剪贴板历史", "action", "系统")
|
||
add("restart_audio", "重启音频服务", "action", "系统")
|
||
add("system_sounds", "打开声音设置", "action", "系统")
|
||
|
||
# ---- 查看（续）----
|
||
add("cpu_model", "CPU 型号", "action", "查看")
|
||
add("memory_sticks", "内存条信息", "action", "查看")
|
||
add("disk_serial", "硬盘序列号", "action", "查看")
|
||
add("gpu_temp", "显卡温度", "action", "查看")
|
||
add("net_total", "网络累计流量", "action", "查看")
|
||
add("running_games", "正在运行的游戏", "action", "查看")
|
||
add("activation", "系统激活状态", "action", "查看")
|
||
add("disk_health", "硬盘健康", "action", "查看")
|
||
|
||
# ---- 效率（续）----
|
||
add("sticky_note", "便签", "input", "效率", hint="写点什么", key="text")
|
||
add("search_note", "搜索笔记", "input", "效率", hint="关键词", key="word")
|
||
add("translator", "打开翻译", "input", "效率", hint="要翻译的词", key="text")
|
||
add("calc_open", "打开计算器2", "action", "效率")
|
||
add("unit_convert", "单位换算", "input", "效率",
|
||
hint="如 100cm", key="text")
|
||
add("countdown_day", "倒数日", "input", "效率",
|
||
hint="几月几号，如 06-20", key="date")
|
||
|
||
# ---- 媒体（续）----
|
||
add("loop_play", "循环播放", "toggle", "媒体", on=False)
|
||
add("shuffle_play", "随机播放开关", "toggle", "媒体", on=False)
|
||
add("stop_all_sound", "停掉所有声音", "action", "媒体")
|
||
|
||
# ---- 网络（续）----
|
||
add("net_traffic", "实时网速", "action", "网络")
|
||
add("relay_saver", "跨网络省流量", "toggle", "网络",
|
||
on=bool(self.store.get("relay_saver")))
|
||
add("traceroute", "路由追踪", "input", "网络",
|
||
hint="如 8.8.8.8", key="host")
|
||
add("net_reset", "重置网络", "action", "网络")
|
||
|
||
# ---- 娱乐 ----
|
||
add("danmaku", "全屏弹幕", "input", "娱乐", hint="弹幕内容", key="text")
|
||
add("marquee", "滚动字幕", "input", "娱乐", hint="要滚动的字", key="text")
|
||
add("lottery", "抽奖", "action", "娱乐")
|
||
add("dice", "掷骰子", "action", "娱乐")
|
||
add("coin_flip", "抛硬币", "action", "娱乐")
|
||
add("countdown_321", "3-2-1 倒计时", "action", "娱乐")
|
||
add("cinema_mode", "影院模式（黑屏）", "action", "娱乐")
|
||
add("random_song", "随机放一首歌", "action", "娱乐")
|
||
add("play_video", "播放视频文件", "input", "娱乐",
|
||
hint="视频路径", key="path")
|
||
add("score_board", "显示计分板", "action", "娱乐")
|
||
add("score_change", "改分数", "input", "娱乐",
|
||
hint="A 或 B", key="team", hint2="加几分（负数减）", key2="delta")
|
||
add("cheer", "加油动画", "action", "娱乐")
|
||
|
||
# ---- 极客 ----
|
||
add("matrix_rain", "数字雨（黑客帝国）", "action", "极客")
|
||
add("starfield", "星空穿梭", "action", "极客")
|
||
add("dashboard", "全屏实时监控", "action", "极客")
|
||
add("qr_code", "生成二维码", "input", "极客",
|
||
hint="要变成二维码的内容", key="text")
|
||
add("hash_text", "算哈希（MD5/SHA）", "input", "极客",
|
||
hint="要计算的文字", key="text")
|
||
add("base64_encode", "Base64 编码", "input", "极客",
|
||
hint="原文", key="text")
|
||
add("base64_decode", "Base64 解码", "input", "极客",
|
||
hint="Base64 内容", key="text")
|
||
add("uuid_gen", "生成 UUID", "action", "极客")
|
||
add("timestamp_now", "当前时间戳", "action", "极客")
|
||
add("json_format", "JSON 格式化", "input", "极客",
|
||
hint="JSON 文本", key="text")
|
||
add("screen_color", "鼠标位置取色", "action", "极客")
|
||
add("speed_test", "网络测速", "action", "极客")
|
||
add("port_check", "检查常用端口", "input", "极客",
|
||
hint="目标 IP 或主机名", key="host")
|
||
|
||
return items
|
||
|
||
# ==================================================================
|
||
# 执行
|
||
# ==================================================================
|
||
|
||
@staticmethod
|
||
def _truthy(value) -> bool:
|
||
"""字符串 "false"/"0"/"off" 不能被当成真 —— bool("false") 是 True。"""
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, (int, float)):
|
||
return value != 0
|
||
if isinstance(value, str):
|
||
return value.strip().lower() in ("1", "true", "on", "yes", "是", "开")
|
||
return bool(value)
|
||
|
||
def run(self, name: str, params: dict) -> str:
|
||
handler = getattr(self, f"_do_{name}", None)
|
||
if handler is None:
|
||
raise ValueError(f"没有这个功能: {name}")
|
||
result = handler(params or {})
|
||
# 执行完顺手记一条"怎么退回去"。
|
||
# 判断逻辑集中在 _remember_undo 里，各个功能自己不用管。
|
||
try:
|
||
self._remember_undo(name, params or {}, result)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return result
|
||
|
||
# ==================================================================
|
||
# 回退（撤销）
|
||
# ==================================================================
|
||
|
||
# 这些**撤不回来**。记了只会给用户"还能退回去"的错觉，反而更危险。
|
||
IRREVERSIBLE = {
|
||
# 电源
|
||
"shutdown", "restart", "sleep", "hibernate", "power",
|
||
"lock_now", "screen_off_lock", "shutdown_timer", "reboot_now",
|
||
"lock_and_off", "class_end_cleanup",
|
||
# 破坏性
|
||
"delete_file", "empty_recycle", "clean_temp", "clean_desktop",
|
||
"kill_process", "kill_games", "kill_by_name", "quit_agent",
|
||
"net_reset", "restart_audio", "usb_check",
|
||
# 一次性只读
|
||
"netcheck", "selfcheck", "site_list", "restrict_status",
|
||
}
|
||
|
||
# 这些会在电脑上开一个全屏窗口 —— 回退就是把它关掉
|
||
VISUAL_FEATURES = {
|
||
"big_text", "big_text_quiet", "class_attention", "class_start",
|
||
"bell", "countdown_screen", "class_timer", "whiteboard",
|
||
"stopwatch", "danmaku", "marquee", "lottery", "dice", "coin_flip",
|
||
"countdown_321", "cinema_mode", "cheer", "matrix_rain",
|
||
"starfield", "dashboard", "roll_group", "roll_call_no_repeat",
|
||
"hand_count", "class_schedule", "countdown_day", "sticky_note",
|
||
}
|
||
|
||
def _spec_map(self) -> dict:
|
||
"""name -> 功能声明。用来查 kind / label。"""
|
||
cached = getattr(self, "_spec_cache", None)
|
||
if isinstance(cached, dict):
|
||
return cached
|
||
out: dict = {}
|
||
try:
|
||
for item in self.defs():
|
||
key = item.get("name")
|
||
if key:
|
||
out[key] = item
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self._spec_cache = out
|
||
return out
|
||
|
||
def _undo_stack(self) -> list:
|
||
stack = getattr(self, "_undo", None)
|
||
if stack is None:
|
||
stack = []
|
||
self._undo = stack
|
||
return stack
|
||
|
||
def _remember_undo(self, name: str, params: dict, result) -> None:
|
||
if not result or name in self.IRREVERSIBLE:
|
||
return
|
||
if name in ("undo_last", "undo_all", "undo_list"):
|
||
return
|
||
|
||
spec = self._spec_map().get(name)
|
||
if spec is None:
|
||
return
|
||
label = str(spec.get("label") or name)
|
||
|
||
entry = None
|
||
if spec.get("kind") == "toggle":
|
||
back = not self._truthy(params.get("on"))
|
||
entry = {
|
||
"name": name,
|
||
"params": {"on": back},
|
||
"label": label,
|
||
"what": f"把「{label}」{'打开' if back else '关掉'}回去",
|
||
}
|
||
elif name in self.VISUAL_FEATURES:
|
||
entry = {
|
||
"name": "close_visuals",
|
||
"params": {},
|
||
"label": label,
|
||
"what": f"关掉「{label}」显示的画面",
|
||
}
|
||
|
||
if entry is None:
|
||
# 其它功能（打开程序、改设置…）暂时不记，
|
||
# 免得给用户"什么都能退"的错误预期
|
||
return
|
||
|
||
entry["at"] = time.strftime("%H:%M:%S")
|
||
stack = self._undo_stack()
|
||
stack.append(entry)
|
||
# 只留最近 20 条，别让内存涨（这台机器还要长时间跑）
|
||
if len(stack) > 20:
|
||
del stack[:-20]
|
||
|
||
def _apply_undo(self, entry: dict) -> str:
|
||
"""执行一条回退。
|
||
|
||
注意：这里**直接调 _do_xxx，不走 run()** ——
|
||
走 run 的话又会记一条新的回退，点一下就来回横跳。
|
||
"""
|
||
target = str(entry.get("name") or "")
|
||
what = str(entry.get("what") or target)
|
||
if not target:
|
||
return "这条回退记录不完整"
|
||
handler = getattr(self, f"_do_{target}", None)
|
||
if handler is None:
|
||
return f"回退失败：找不到 {target}"
|
||
try:
|
||
detail = handler(dict(entry.get("params") or {}))
|
||
except Exception as exc: # noqa: BLE001
|
||
return f"回退失败：{what} —— {exc}"
|
||
return f"已回退：{what}" + (f"（{detail}）" if detail else "")
|
||
|
||
def _do_undo_last(self, params) -> str:
|
||
stack = self._undo_stack()
|
||
if not stack:
|
||
return "没有可以回退的操作"
|
||
entry = stack.pop()
|
||
return self._apply_undo(entry)
|
||
|
||
def _do_undo_all(self, params) -> str:
|
||
stack = self._undo_stack()
|
||
if not stack:
|
||
return "没有可以回退的操作"
|
||
count = len(stack)
|
||
lines = []
|
||
# 倒序：后做的先撤
|
||
for entry in reversed(stack):
|
||
lines.append(self._apply_undo(entry))
|
||
stack.clear()
|
||
return f"已回退 {count} 步：\n" + "\n".join(lines)
|
||
|
||
def _do_undo_list(self, params) -> str:
|
||
stack = self._undo_stack()
|
||
if not stack:
|
||
return "没有可以回退的操作"
|
||
lines = [f"{i + 1}. [{e.get('at', '')}] {e.get('what', '')}"
|
||
for i, e in enumerate(reversed(stack))]
|
||
return (f"可以回退 {len(stack)} 步（最近的在前）：\n"
|
||
+ "\n".join(lines))
|
||
|
||
# ---------------- 显示 ----------------
|
||
|
||
def _night_mode(self) -> bool:
|
||
code, out = _run(["reg", "query",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion\CloudStore\Store"
|
||
r"\DefaultAccount\Current\default$windows.data.bluelightreduction"
|
||
r".bluelightreductionstate\windows.data.bluelightreduction.bluelightreductionstate"])
|
||
return code == 0 and "02 01" in out
|
||
|
||
def _do_night_mode(self, params) -> str:
|
||
want_on = self._truthy(params.get("on"))
|
||
# 用 PowerShell 切注册表，Win10/11 通用
|
||
if want_on:
|
||
script = (
|
||
"$p='HKCU:\\Software\\Microsoft\\Windows\\CurrentVersion\\CloudStore\\Store"
|
||
"\\DefaultAccount\\Current\\default$windows.data.bluelightreduction"
|
||
".bluelightreductionstate\\windows.data.bluelightreduction.bluelightreductionstate';"
|
||
"if(-not(Test-Path $p)){New-Item -Path $p -Force|Out-Null;"
|
||
"New-ItemProperty -Path $p -Name Data -PropertyType Binary "
|
||
"-Value ([byte[]](0x02,0x01,0x00)) -Force|Out-Null};"
|
||
"$d=(Get-ItemProperty -Path $p -Name Data).Data;"
|
||
"if($d.Length -lt 4){Set-ItemProperty -Path $p -Name Data "
|
||
"-Value ([byte[]](0x10,0x00,0x00,0x00,0x43,0x42,0x01,0x00,0x0A,0x02,0x01,0x00))};"
|
||
"else{Set-ItemProperty -Path $p -Name Data -Value "
|
||
"([byte[]](0x10,0x00,0x00,0x00,0x43,0x42,0x01,0x00,0x0A,0x02,0x01,0x00))}")
|
||
else:
|
||
script = (
|
||
"$p='HKCU:\\Software\\Microsoft\\Windows\\CurrentVersion\\CloudStore\\Store"
|
||
"\\DefaultAccount\\Current\\default$windows.data.bluelightreduction"
|
||
".bluelightreductionstate\\windows.data.bluelightreduction.bluelightreductionstate';"
|
||
"if(Test-Path $p){Set-ItemProperty -Path $p -Name Data "
|
||
"-Value ([byte[]](0x10,0x00,0x00,0x00,0x43,0x42,0x01,0x00,0x0A,0x02,0x00,0x00))}")
|
||
_run(["powershell", "-NoProfile", "-Command", script])
|
||
self._broadcast_setting()
|
||
return "夜间模式已开" if want_on else "夜间模式已关"
|
||
|
||
def _taskbar_hidden(self) -> bool:
|
||
"""任务栏是不是自动隐藏。
|
||
|
||
reg query 把 REG_BINARY 输出成逗号分隔的十六进制：
|
||
Settings REG_BINARY 30,00,00,00,00,00,00,00,03,00,00,00,...
|
||
下标 8 那个字节的最低位就是「自动隐藏」开关。
|
||
|
||
以前是拿字符串 "03,00,00,00" 去整段匹配 —— 大小写和位置都不稳，
|
||
状态永远读成 False，于是手表每次点都发"隐藏"，
|
||
表现成"能隐藏但恢复不了"。这里老老实实按字节解析。
|
||
"""
|
||
code, out = _run(["reg", "query",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion"
|
||
r"\Explorer\StuckRects3",
|
||
"/v", "Settings"])
|
||
if code != 0 or not out:
|
||
return False
|
||
match = re.search(r"REG_BINARY\s+([0-9a-fA-F,\s]+)", out)
|
||
if not match:
|
||
return False
|
||
values = [v for v in re.split(r"[,\s]+", match.group(1).strip()) if v]
|
||
if len(values) <= 8:
|
||
return False
|
||
try:
|
||
return bool(int(values[8], 16) & 0x01)
|
||
except ValueError:
|
||
return False
|
||
|
||
def _do_taskbar_hide(self, params) -> str:
|
||
want = self._truthy(params.get("on"))
|
||
script = (
|
||
"$p='HKCU:\\Software\\Microsoft\\Windows\\CurrentVersion\\Explorer\\StuckRects3';"
|
||
"$s=(Get-ItemProperty -Path $p -Name Settings).Settings;"
|
||
f"$s[8]={3 if want else 2};"
|
||
"Set-ItemProperty -Path $p -Name Settings -Value $s -Type Binary;"
|
||
"Stop-Process -Name explorer -Force -ErrorAction SilentlyContinue")
|
||
_run(["powershell", "-NoProfile", "-Command", script])
|
||
return "任务栏已设为自动隐藏" if want else "任务栏已恢复常显"
|
||
|
||
def _brightness(self) -> int:
|
||
code, out = _run(["powershell", "-NoProfile", "-Command",
|
||
"(Get-CimInstance -Namespace root/WMI "
|
||
"-ClassName WmiMonitorBrightness -ErrorAction SilentlyContinue)"
|
||
".CurrentBrightness"])
|
||
digits = re.findall(r"\d+", out or "")
|
||
return int(digits[0]) if digits else 50
|
||
|
||
def _do_brightness(self, params) -> str:
|
||
try:
|
||
level = max(10, min(100, int(params.get("value") or 50)))
|
||
except (TypeError, ValueError):
|
||
raise ValueError("亮度要 10~100")
|
||
_run(["powershell", "-NoProfile", "-Command",
|
||
f"(Get-CimInstance -Namespace root/WMI -ClassName "
|
||
f"WmiMonitorBrightnessMethods).WmiSetBrightness(1,{level})"])
|
||
return f"亮度已设为 {level}%"
|
||
|
||
def _mouse_speed(self) -> int:
|
||
if not IS_WINDOWS:
|
||
return 10
|
||
try:
|
||
value = ctypes.c_int(0)
|
||
ctypes.windll.user32.SystemParametersInfoW(0x0070, 0, ctypes.byref(value), 0)
|
||
return int(value.value)
|
||
except Exception: # noqa: BLE001
|
||
return 10
|
||
|
||
def _do_mouse_speed(self, params) -> str:
|
||
try:
|
||
level = max(1, min(20, int(params.get("value") or 10)))
|
||
except (TypeError, ValueError):
|
||
raise ValueError("鼠标速度要 1~20")
|
||
ctypes.windll.user32.SystemParametersInfoW(0x0071, 0, level, 3)
|
||
return f"鼠标速度已设为 {level}"
|
||
|
||
def _do_monitor_on(self, params) -> str:
|
||
return self.actions.monitor_power(True)
|
||
|
||
def _broadcast_setting(self) -> None:
|
||
HWND_BROADCAST = 0xFFFF
|
||
WM_SETTINGCHANGE = 0x001A
|
||
ctypes.windll.user32.SendMessageW(HWND_BROADCAST, WM_SETTINGCHANGE, 0, 0)
|
||
|
||
# ---------------- 音量 ----------------
|
||
|
||
def _volume_key(self, vk: int, times: int = 1) -> None:
|
||
VK_VOLUME_MUTE = 0xAD
|
||
VK_VOLUME_DOWN = 0xAE
|
||
VK_VOLUME_UP = 0xAF
|
||
for _ in range(max(1, min(50, times))):
|
||
ctypes.windll.user32.keybd_event(vk, 0, 0, 0)
|
||
ctypes.windll.user32.keybd_event(vk, 0, 2, 0)
|
||
time.sleep(0.02)
|
||
|
||
def _do_volume_up(self, params) -> str:
|
||
self._volume_key(0xAF, 5)
|
||
return "音量 +5"
|
||
|
||
def _do_volume_down(self, params) -> str:
|
||
self._volume_key(0xAE, 5)
|
||
return "音量 -5"
|
||
|
||
def _do_mute_all(self, params) -> str:
|
||
self._volume_key(0xAD, 1)
|
||
return "已切换静音"
|
||
|
||
# ---------------- 系统 ----------------
|
||
|
||
def _do_lock_and_off(self, params) -> str:
|
||
ctypes.windll.user32.LockWorkStation()
|
||
time.sleep(0.4)
|
||
self.actions.monitor_power(False)
|
||
return "已锁屏并关屏"
|
||
|
||
def _do_empty_recycle(self, params) -> str:
|
||
code, out = _run(["powershell", "-NoProfile", "-Command",
|
||
"Clear-RecycleBin -Force -ErrorAction SilentlyContinue; 'ok'"])
|
||
return "回收站已清空" if "ok" in out else f"清空失败: {out[:60]}"
|
||
|
||
def _do_flush_dns(self, params) -> str:
|
||
_run(["ipconfig", "/flushdns"])
|
||
return "DNS 缓存已刷新"
|
||
|
||
def _do_restart_explorer(self, params) -> str:
|
||
_run(["taskkill", "/F", "/IM", "explorer.exe"])
|
||
time.sleep(1.0)
|
||
subprocess.Popen("explorer.exe", shell=True,
|
||
creationflags=_NO_WINDOW)
|
||
return "资源管理器已重启"
|
||
|
||
def _do_clear_clip(self, params) -> str:
|
||
self.actions.clip_set("")
|
||
return "电脑剪贴板已清空"
|
||
|
||
# ---------------- 查看 ----------------
|
||
|
||
def _do_uptime(self, params) -> str:
|
||
info = self.actions.perf()
|
||
return f"已开机 {info.get('uptime', '-')}"
|
||
|
||
def _do_perf(self, params) -> str:
|
||
info = self.actions.perf()
|
||
return (f"CPU {info.get('cpu_percent')}% "
|
||
f"内存 {info.get('mem_percent')}%"
|
||
f"（{info.get('mem_used_gb')}/{info.get('mem_total_gb')} GB）")
|
||
|
||
def _do_disk(self, params) -> str:
|
||
return " | ".join(f"{d['drive']} 剩 {d['free_gb']}GB"
|
||
for d in self.actions.disk()) or "读不到"
|
||
|
||
def _do_battery(self, params) -> str:
|
||
info = self.actions.battery()
|
||
if not info.get("present"):
|
||
return info.get("note") or "没有电池"
|
||
return f"{info['percent']}%" + ("（充电中）" if info.get("charging") else "")
|
||
|
||
def _do_network_speed(self, params) -> str:
|
||
"""真实速率：采两次求差，不是开机以来的累计流量。"""
|
||
script = ("(Get-NetAdapterStatistics | Measure-Object -Property "
|
||
"ReceivedBytes -Sum).Sum")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
digits = re.findall(r"\d+", out or "")
|
||
now_bytes = int(digits[0]) if digits else 0
|
||
now = time.time()
|
||
last_bytes = getattr(self, "_net_last_bytes", None)
|
||
last_time = getattr(self, "_net_last_time", None)
|
||
self._net_last_bytes = now_bytes
|
||
self._net_last_time = now
|
||
|
||
if last_bytes is None or last_time is None or now <= last_time:
|
||
time.sleep(1.0)
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
digits = re.findall(r"\d+", out or "")
|
||
again = int(digits[0]) if digits else now_bytes
|
||
self._net_last_bytes = again
|
||
self._net_last_time = time.time()
|
||
return f"当前约 {(again - now_bytes) / 1048576:.2f} MB/s（采样 1 秒）"
|
||
|
||
elapsed = now - last_time
|
||
delta = max(0, now_bytes - last_bytes)
|
||
mbps = delta / elapsed / 1048576
|
||
return (f"下载 {mbps:.2f} MB/s（{mbps * 8:.1f} Mbps）\n"
|
||
f"这段时间共收了 {delta / 1048576:.1f} MB")
|
||
|
||
def _do_installed_apps(self, params) -> str:
|
||
if not IS_WINDOWS:
|
||
return "仅支持 Windows"
|
||
import winreg
|
||
names: list[str] = []
|
||
for hive, path in (
|
||
(winreg.HKEY_LOCAL_MACHINE,
|
||
r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"),
|
||
(winreg.HKEY_LOCAL_MACHINE,
|
||
r"SOFTWARE\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall"),
|
||
(winreg.HKEY_CURRENT_USER,
|
||
r"SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall"),
|
||
):
|
||
try:
|
||
with winreg.OpenKey(hive, path) as key:
|
||
for i in range(winreg.QueryInfoKey(key)[0]):
|
||
try:
|
||
sub = winreg.EnumKey(key, i)
|
||
with winreg.OpenKey(key, sub) as item:
|
||
name = winreg.QueryValueEx(item, "DisplayName")[0]
|
||
if name and name not in names:
|
||
names.append(str(name))
|
||
except OSError:
|
||
continue
|
||
except OSError:
|
||
continue
|
||
names.sort()
|
||
return f"共 {len(names)} 个：\n" + "\n".join(names[:60])
|
||
|
||
def _do_top_services(self, params) -> str:
|
||
code, out = _run(["powershell", "-NoProfile", "-Command",
|
||
"Get-Service | Where-Object {$_.Status -eq 'Running'} | "
|
||
"Select-Object -First 40 -ExpandProperty Name"])
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"运行中 {len(lines)} 个（前 40）：\n" + "\n".join(lines)
|
||
|
||
def _do_recent_files(self, params) -> str:
|
||
recent = Path(os.environ.get("APPDATA", "")) / "Microsoft/Windows/Recent"
|
||
if not recent.is_dir():
|
||
return "读不到最近文件"
|
||
items = sorted(recent.glob("*.lnk"),
|
||
key=lambda p: p.stat().st_mtime, reverse=True)[:25]
|
||
names = [p.stem for p in items]
|
||
return f"最近 {len(names)} 个：\n" + "\n".join(names)
|
||
|
||
def _do_event_errors(self, params) -> str:
|
||
script = ("Get-EventLog -LogName System -EntryType Error -Newest 15 "
|
||
"-ErrorAction SilentlyContinue | "
|
||
"ForEach-Object { \"$($_.TimeGenerated.ToString('MM-dd HH:mm')) "
|
||
"$($_.Source)\" }")
|
||
code, out = _run(["powershell", "-NoProfile", "-Command", script], timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return (f"最近 {len(lines)} 条系统错误：\n" + "\n".join(lines)
|
||
if lines else "没有系统错误记录")
|
||
|
||
def _do_temperature(self, params) -> str:
|
||
script = ("$t=Get-CimInstance -Namespace root/WMI -ClassName "
|
||
"MSAcpi_ThermalZoneTemperature -ErrorAction SilentlyContinue | "
|
||
"Select-Object -First 1;"
|
||
"if($t){[math]::Round(($t.CurrentTemperature/10)-273.15,1)}else{'n/a'}")
|
||
code, out = _run(["powershell", "-NoProfile", "-Command", script], timeout=25)
|
||
cpu_temp = out.strip().splitlines()[-1] if out.strip() else "n/a"
|
||
info = self.actions.perf()
|
||
return (f"CPU {info.get('cpu_percent')}% 内存 {info.get('mem_percent')}% "
|
||
f"温度 {cpu_temp}°C")
|
||
|
||
# ---------------- 效率 ----------------
|
||
|
||
def _do_note_add(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("笔记内容为空")
|
||
# 必须先 load。__init__ 里 self._notes 是空的，
|
||
# 直接 insert 再 persist 会把历史笔记全抹掉（重启后第一次记笔记就中招）。
|
||
self._notes = self._load("notes", dict)
|
||
self._notes.insert(0, {"text": text[:2000],
|
||
"at": time.strftime("%m-%d %H:%M")})
|
||
del self._notes[200:]
|
||
self._persist("notes", self._notes)
|
||
return f"已记下（共 {len(self._notes)} 条）"
|
||
|
||
def _do_note_list(self, params) -> str:
|
||
self._notes = self._load("notes")
|
||
if not self._notes:
|
||
return "还没有笔记"
|
||
return f"共 {len(self._notes)} 条：\n" + "\n".join(
|
||
f"[{n['at']}] {n['text'][:80]}" for n in self._notes[:30])
|
||
|
||
def _do_note_clear(self, params) -> str:
|
||
self._notes = []
|
||
self._persist("notes", [])
|
||
return "笔记已清空"
|
||
|
||
def _do_todo_add(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("待办内容为空")
|
||
todos = self._load("todos", dict)
|
||
todos.append({"text": text[:500], "done": False})
|
||
self._persist("todos", todos)
|
||
return f"已添加（共 {len(todos)} 条待办）"
|
||
|
||
def _do_todo_list(self, params) -> str:
|
||
todos = self._load("todos")
|
||
if not todos:
|
||
return "待办是空的"
|
||
done = sum(1 for t in todos if t.get("done"))
|
||
lines = [f"{'✓' if t.get('done') else '○'} {t['text'][:60]}" for t in todos[:40]]
|
||
return f"共 {len(todos)} 条（完成 {done}）：\n" + "\n".join(lines)
|
||
|
||
def _do_reminder(self, params) -> str:
|
||
try:
|
||
minutes = max(1, min(600, int(params.get("minutes") or 5)))
|
||
except (TypeError, ValueError):
|
||
minutes = 5
|
||
text = str(params.get("text") or "时间到了").strip()
|
||
|
||
import threading
|
||
|
||
def fire():
|
||
self.actions.notify("定时提醒", text)
|
||
self.actions.say(text)
|
||
|
||
timer = threading.Timer(minutes * 60, fire)
|
||
timer.daemon = True
|
||
timer.start()
|
||
self._timers.append(timer)
|
||
return f"{minutes} 分钟后提醒你：{text}"
|
||
|
||
def _do_pomodoro(self, params) -> str:
|
||
import threading
|
||
|
||
def fire():
|
||
self.actions.say("番茄钟结束，休息一下")
|
||
self.actions.notify("番茄钟", "25 分钟到了，休息 5 分钟")
|
||
|
||
timer = threading.Timer(25 * 60, fire)
|
||
timer.daemon = True
|
||
timer.start()
|
||
self._timers.append(timer)
|
||
self._do_lock_and_off({})
|
||
return "番茄钟开始：25 分钟，已锁屏关屏"
|
||
|
||
def _do_clip_fav_add(self, params) -> str:
|
||
text = str(self.store.get("last_clip") or "").strip()
|
||
if not text:
|
||
raise ValueError("电脑剪贴板是空的")
|
||
favs = self._load("clip_fav", str)
|
||
favs.insert(0, text[:2000])
|
||
self._persist("clip_fav", favs[:50])
|
||
return f"已收藏（共 {len(favs[:50])} 条）"
|
||
|
||
def _do_clip_fav_list(self, params) -> str:
|
||
favs = self._load("clip_fav")
|
||
if not favs:
|
||
return "收藏夹是空的"
|
||
return f"共 {len(favs)} 条：\n" + "\n".join(
|
||
f"{i + 1}. {t[:60]}" for i, t in enumerate(favs[:30]))
|
||
|
||
# ---------------- 文件 ----------------
|
||
|
||
def _do_list_dir(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or "Downloads"))
|
||
if not target.is_dir():
|
||
raise ValueError(f"不是文件夹: {target}")
|
||
try:
|
||
entries = sorted(target.iterdir(),
|
||
key=lambda p: (p.is_file(), p.name.lower()))
|
||
except OSError as exc:
|
||
raise ValueError(f"读不了这个文件夹: {exc}") from exc
|
||
lines = []
|
||
for entry in entries[:60]:
|
||
if entry.is_dir():
|
||
lines.append(f"[目录] {entry.name}")
|
||
else:
|
||
try:
|
||
size = entry.stat().st_size
|
||
except OSError:
|
||
size = 0
|
||
lines.append(f"{size // 1024} KB {entry.name}")
|
||
return f"{target}\n共 {len(entries)} 项：\n" + "\n".join(lines)
|
||
|
||
def _do_download_file(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.is_file():
|
||
raise ValueError(f"不是文件: {target}")
|
||
size = target.stat().st_size
|
||
if size > 30 * 1024 * 1024:
|
||
raise ValueError("文件超过 30MB，太大")
|
||
readable = (f"{size / 1048576:.1f} MB" if size > 1048576
|
||
else f"{max(1, size // 1024)} KB")
|
||
return f"{target.name}（{readable}），正在传到手表…"
|
||
|
||
def _do_search_files(self, params) -> str:
|
||
keyword = str(params.get("keyword") or "").strip().lower()
|
||
if not keyword:
|
||
raise ValueError("关键词为空")
|
||
found: list[str] = []
|
||
for root in self.allowed_roots():
|
||
try:
|
||
for path in root.rglob(f"*{keyword}*"):
|
||
if path.is_file():
|
||
found.append(str(path))
|
||
if len(found) >= 40:
|
||
break
|
||
except OSError:
|
||
continue
|
||
if len(found) >= 40:
|
||
break
|
||
return (f"找到 {len(found)} 个：\n" + "\n".join(found)
|
||
if found else "没找到")
|
||
|
||
def _do_make_dir(self, params) -> str:
|
||
parent = self.resolve_in_roots(str(params.get("path") or "Downloads"))
|
||
name = re.sub(r'[<>:"/\\|?*]', "_", str(params.get("name") or "").strip())
|
||
if not name:
|
||
raise ValueError("文件夹名为空")
|
||
if not parent.is_dir():
|
||
raise ValueError("父目录不是文件夹")
|
||
target = parent / name
|
||
if target.exists():
|
||
raise ValueError("同名文件或文件夹已存在")
|
||
try:
|
||
target.mkdir(parents=True, exist_ok=False)
|
||
except OSError as exc:
|
||
raise ValueError(f"新建失败: {exc}") from exc
|
||
return f"已新建 {target}"
|
||
|
||
def _do_rename_file(self, params) -> str:
|
||
source = self.resolve_in_roots(str(params.get("path") or ""))
|
||
name = re.sub(r'[<>:"/\\|?*]', "_", str(params.get("name") or "").strip())
|
||
if not name:
|
||
raise ValueError("新名字为空")
|
||
if not source.exists():
|
||
raise ValueError("文件不存在")
|
||
self.assert_not_root(source)
|
||
target = source.parent / name
|
||
if target.exists():
|
||
raise ValueError("同名文件已存在")
|
||
try:
|
||
source.rename(target)
|
||
except OSError as exc:
|
||
raise ValueError(f"重命名失败: {exc}") from exc
|
||
return f"已重命名为 {target.name}"
|
||
|
||
def _recycle(self, target: Path) -> None:
|
||
"""删除到回收站。
|
||
|
||
之前是把路径用单引号拼进 PowerShell 脚本，文件名里带一个单引号
|
||
就能闭合字符串、注入任意命令（新建文件夹的清洗正则不删单引号，
|
||
所以远端两步就能造出这种文件名）。现在直接调 Windows 自己的
|
||
SHFileOperationW，不经过任何 shell，没有注入面。
|
||
"""
|
||
class SHFILEOPSTRUCTW(ctypes.Structure):
|
||
_fields_ = [("hwnd", ctypes.c_void_p),
|
||
("wFunc", ctypes.c_uint),
|
||
("pFrom", ctypes.c_wchar_p),
|
||
("pTo", ctypes.c_wchar_p),
|
||
("fFlags", ctypes.c_ushort),
|
||
("fAnyOperationsAborted", ctypes.c_int),
|
||
("hNameMappings", ctypes.c_void_p),
|
||
("lpszProgressTitle", ctypes.c_wchar_p)]
|
||
|
||
FO_DELETE = 3
|
||
FOF_ALLOWUNDO = 0x0040
|
||
FOF_NOCONFIRMATION = 0x0010
|
||
FOF_SILENT = 0x0004
|
||
FOF_NOERRORUI = 0x0400
|
||
|
||
# pFrom 要双 \0 结尾
|
||
raw = str(target) + "\0"
|
||
buffer = ctypes.create_unicode_buffer(len(raw) + 1)
|
||
buffer[:len(raw)] = raw
|
||
|
||
op = SHFILEOPSTRUCTW()
|
||
op.wFunc = FO_DELETE
|
||
op.pFrom = ctypes.cast(buffer, ctypes.c_wchar_p)
|
||
op.fFlags = (FOF_ALLOWUNDO | FOF_NOCONFIRMATION
|
||
| FOF_SILENT | FOF_NOERRORUI)
|
||
result = ctypes.windll.shell32.SHFileOperationW(ctypes.byref(op))
|
||
if result != 0:
|
||
raise ValueError(f"删除失败（系统错误码 {result}）")
|
||
|
||
def _do_delete_file(self, params) -> str:
|
||
"""删除到回收站，不做永久删除。"""
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.exists():
|
||
raise ValueError("文件不存在")
|
||
self.assert_not_root(target)
|
||
try:
|
||
if IS_WINDOWS:
|
||
self._recycle(target)
|
||
else:
|
||
raise ValueError("仅支持 Windows")
|
||
except ValueError:
|
||
raise
|
||
except OSError as exc:
|
||
raise ValueError(f"删除失败: {exc}") from exc
|
||
return f"已删除到回收站：{target.name}"
|
||
|
||
def _do_open_file(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.exists():
|
||
raise ValueError("文件不存在")
|
||
os.startfile(str(target)) # noqa: S606
|
||
return f"已用默认程序打开 {target.name}"
|
||
|
||
# ==================================================================
|
||
# 教学
|
||
# ==================================================================
|
||
|
||
def _close_fullscreen_windows(self) -> int:
|
||
"""关掉所有由本程序开的全屏窗口。
|
||
|
||
注意：**必须回到 Tk 线程去 destroy**。Tk 的对象只能在创建它的
|
||
线程里操作，直接在服务端线程里调 win.destroy() 会静默失败，
|
||
表现成"全屏大字关不掉"。
|
||
"""
|
||
closers = list(getattr(self, "_fullscreen_closers", []))
|
||
self._fullscreen_closers = []
|
||
if not closers:
|
||
return 0
|
||
|
||
def run_all():
|
||
for closer in closers:
|
||
try:
|
||
closer()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
ui = getattr(self.actions, "ui", None)
|
||
if ui is not None:
|
||
try:
|
||
ui.ui_call(run_all)
|
||
return len(closers)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
# 界面拿不到（或超时）才退回直接调，至少试一下
|
||
closed = 0
|
||
for closer in closers:
|
||
try:
|
||
closer()
|
||
closed += 1
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return closed
|
||
|
||
def _fullscreen_window(self, builder, seconds: float = 0.0,
|
||
title: str = ""):
|
||
"""在电脑上开一个全屏窗口。builder(win, close) 负责放内容。
|
||
|
||
先关掉上一个：不然连着点两次「全屏大字」，窗口会一层层叠上去，
|
||
Esc 只能关最上面那个，下面的再也关不掉。
|
||
"""
|
||
self._close_fullscreen_windows()
|
||
|
||
def build():
|
||
import tkinter as tk
|
||
root = getattr(self.actions.ui, "root", None)
|
||
if root is None:
|
||
raise RuntimeError("界面不可用")
|
||
win = tk.Toplevel(root)
|
||
win.attributes("-fullscreen", True)
|
||
win.attributes("-topmost", True)
|
||
|
||
def close(_event=None):
|
||
try:
|
||
win.destroy()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
self._fullscreen_closers.remove(close)
|
||
except (ValueError, AttributeError):
|
||
pass
|
||
|
||
builder(win, close)
|
||
# 给窗口一个标题：不然枚举窗口、任务栏、Alt+Tab 都看不到它，
|
||
# 用户也不知道电脑上那个大黑屏是什么。
|
||
try:
|
||
win.title(title or "局域网遥控")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
win.bind("<Escape>", close)
|
||
win.bind("<Button-1>", close)
|
||
win.protocol("WM_DELETE_WINDOW", close)
|
||
if seconds > 0:
|
||
win.after(int(seconds * 1000), close)
|
||
self._fullscreen_closers.append(close)
|
||
|
||
try:
|
||
self.actions.ui.ui_call(build)
|
||
except (RuntimeError, TimeoutError, AttributeError) as exc:
|
||
# 界面关了 / 忙超时 —— 转成用户看得懂的话，别把栈信息甩出去
|
||
raise ValueError(f"电脑上的界面不可用（{exc}）") from exc
|
||
|
||
def _do_big_text(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("文字为空")
|
||
size = 96 if len(text) <= 8 else (64 if len(text) <= 20 else 40)
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
tk.Label(win, text=text, bg="#0B1020", fg="#FFFFFF",
|
||
font=("Microsoft YaHei UI", size, "bold"),
|
||
wraplength=win.winfo_screenwidth() - 80,
|
||
justify="center").pack(expand=True, fill="both")
|
||
tk.Label(win, text="按 Esc 或点一下关闭", bg="#0B1020", fg="#55607A",
|
||
font=("Microsoft YaHei UI", 12)).place(
|
||
relx=0.5, rely=0.97, anchor="center")
|
||
|
||
self._fullscreen_window(builder, title=text[:40])
|
||
return f"已全屏显示「{text[:20]}」"
|
||
|
||
def _do_big_text_quiet(self, params) -> str:
|
||
return self._do_big_text({"text": "安静"})
|
||
|
||
def _do_big_text_clear(self, params) -> str:
|
||
closed = self._close_fullscreen_windows()
|
||
return f"已关掉 {closed} 个全屏窗口" if closed else "当前没有全屏窗口"
|
||
|
||
def _do_close_visuals(self, params) -> str:
|
||
"""把本程序在电脑上开出来的画面全部收掉。
|
||
|
||
覆盖：全屏大字 / 倒计时 / 请安静 / 下课铃 / 投图 / 通知 / 屏幕共享提示。
|
||
手表能打开的东西，必须也能关掉。
|
||
"""
|
||
done = []
|
||
|
||
closed = self._close_fullscreen_windows()
|
||
if closed:
|
||
done.append(f"全屏窗口 {closed} 个")
|
||
|
||
for label, method in (("投图", "close_image"),
|
||
("通知", "close_notifications")):
|
||
func = getattr(self.actions.ui, method, None)
|
||
if func is None:
|
||
continue
|
||
try:
|
||
if func():
|
||
done.append(label)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
# 屏幕镜像提示条也一起收掉（镜像本身由手表那个开关控制）
|
||
try:
|
||
hide = getattr(self.actions.ui, "_hide_notice", None)
|
||
if hide is not None and getattr(self.actions.ui, "_notice", None):
|
||
hide()
|
||
done.append("共享提示")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
return "已关掉：" + "、".join(done) if done else "电脑上现在没有本程序开的画面"
|
||
|
||
def _do_countdown_screen(self, params) -> str:
|
||
try:
|
||
minutes = max(1, min(180, int(params.get("minutes") or 5)))
|
||
except (TypeError, ValueError):
|
||
minutes = 5
|
||
total = minutes * 60
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Consolas", 150, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
tk.Label(win, text=f"倒计时 {minutes} 分钟 · Esc 结束",
|
||
bg="#0B1020", fg="#55607A",
|
||
font=("Microsoft YaHei UI", 14)).place(
|
||
relx=0.5, rely=0.95, anchor="center")
|
||
state = {"left": total}
|
||
|
||
def tick():
|
||
if state["left"] < 0:
|
||
return
|
||
mins, secs = divmod(max(0, state["left"]), 60)
|
||
label.configure(text=f"{mins:02d}:{secs:02d}",
|
||
fg="#FF6B6B" if state["left"] <= 10 else "#FFFFFF")
|
||
if state["left"] == 0:
|
||
try:
|
||
win.bell()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
state["left"] -= 1
|
||
win.after(1000, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"已开始 {minutes} 分钟倒计时"
|
||
|
||
def _do_whiteboard(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#FFFFFF")
|
||
canvas = tk.Canvas(win, bg="#FFFFFF", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
state = {"x": None, "y": None, "color": "#1F2328"}
|
||
|
||
def press(event):
|
||
state["x"], state["y"] = event.x, event.y
|
||
|
||
def drag(event):
|
||
if state["x"] is None:
|
||
return
|
||
canvas.create_line(state["x"], state["y"], event.x, event.y,
|
||
fill=state["color"], width=4, capstyle="round")
|
||
state["x"], state["y"] = event.x, event.y
|
||
|
||
def release(_event):
|
||
state["x"] = state["y"] = None
|
||
|
||
canvas.bind("<Button-1>", press)
|
||
canvas.bind("<B1-Motion>", drag)
|
||
canvas.bind("<ButtonRelease-1>", release)
|
||
|
||
bar = tk.Frame(win, bg="#F5F7FA")
|
||
bar.place(relx=1.0, rely=0.0, anchor="ne")
|
||
for text, color in (("黑", "#1F2328"), ("红", "#CF222E"),
|
||
("蓝", "#185CE8"), ("绿", "#1A7F37")):
|
||
tk.Button(bar, text=text, bg="#FFFFFF", fg=color, relief="flat",
|
||
font=("Microsoft YaHei UI", 11, "bold"), width=3,
|
||
command=lambda c=color: state.update(color=c)).pack(
|
||
side="left", padx=2, pady=4)
|
||
tk.Button(bar, text="清空", bg="#EAEEF2", relief="flat",
|
||
font=("Microsoft YaHei UI", 10),
|
||
command=lambda: canvas.delete("all")).pack(
|
||
side="left", padx=6, pady=4)
|
||
|
||
self._fullscreen_window(builder)
|
||
return "白板已打开（鼠标画，Esc 关闭）"
|
||
|
||
def _do_display_mode(self, params) -> str:
|
||
subprocess.Popen(["cmd", "/c", "start", "ms-settings:display"],
|
||
creationflags=_NO_WINDOW)
|
||
return "已打开显示设置"
|
||
|
||
def _do_roll_call_set(self, params) -> str:
|
||
raw = str(params.get("names") or "").strip()
|
||
if not raw:
|
||
raise ValueError("名单为空")
|
||
for sep in ("，", "、", " ", ";", "；"):
|
||
raw = raw.replace(sep, ",")
|
||
parts = [p.strip() for p in raw.split(",") if p.strip()]
|
||
if not parts:
|
||
raise ValueError("没解析出名字")
|
||
self.store.set("roll_call_names", parts)
|
||
return f"名单已存，共 {len(parts)} 人"
|
||
|
||
def _do_roll_call(self, params) -> str:
|
||
import random
|
||
names = self.store.get("roll_call_names") or []
|
||
if not names:
|
||
raise ValueError("还没有名单，先点「设置点名名单」")
|
||
picked = random.choice(names)
|
||
self._do_big_text({"text": picked})
|
||
return f"点到：{picked}（共 {len(names)} 人）"
|
||
|
||
# ==================================================================
|
||
# 系统控制
|
||
# ==================================================================
|
||
|
||
def _key_combo(self, *vks) -> None:
|
||
for vk in vks:
|
||
ctypes.windll.user32.keybd_event(vk, 0, 0, 0)
|
||
for vk in reversed(vks):
|
||
ctypes.windll.user32.keybd_event(vk, 0, 2, 0)
|
||
|
||
def _do_minimize_all(self, params) -> str:
|
||
self._key_combo(0x5B, 0x44) # Win + D
|
||
return "已显示桌面"
|
||
|
||
def _do_task_view(self, params) -> str:
|
||
self._key_combo(0x5B, 0x09) # Win + Tab
|
||
return "已打开任务视图"
|
||
|
||
def _do_close_windows(self, params) -> str:
|
||
WM_CLOSE = 0x0010
|
||
user32 = ctypes.windll.user32
|
||
EnumProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p,
|
||
ctypes.c_void_p)
|
||
own_pid = os.getpid()
|
||
state = {"n": 0}
|
||
|
||
def callback(hwnd, _lparam):
|
||
if not user32.IsWindowVisible(hwnd):
|
||
return True
|
||
if user32.GetWindowTextLengthW(hwnd) == 0:
|
||
return True
|
||
pid = ctypes.c_ulong()
|
||
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
||
if pid.value == own_pid:
|
||
return True
|
||
cls = ctypes.create_unicode_buffer(256)
|
||
user32.GetClassNameW(hwnd, cls, 256)
|
||
if cls.value in ("Progman", "Shell_TrayWnd", "WorkerW"):
|
||
return True
|
||
user32.PostMessageW(hwnd, WM_CLOSE, 0, 0)
|
||
state["n"] += 1
|
||
return True
|
||
|
||
user32.EnumWindows(EnumProc(callback), 0)
|
||
return f"已向 {state['n']} 个窗口发出关闭"
|
||
|
||
def _clock_seconds(self) -> bool:
|
||
code, out = _run(["reg", "query",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion"
|
||
r"\Explorer\Advanced", "/v", "ShowSecondsInSystemClock"])
|
||
return code == 0 and "0x1" in out
|
||
|
||
def _do_clock_seconds(self, params) -> str:
|
||
want = 1 if self._truthy(params.get("on")) else 0
|
||
_run(["reg", "add",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion\Explorer\Advanced",
|
||
"/v", "ShowSecondsInSystemClock", "/t", "REG_DWORD",
|
||
"/d", str(want), "/f"])
|
||
self._restart_explorer_quiet()
|
||
return "任务栏时钟已显示秒" if want else "任务栏时钟已恢复"
|
||
|
||
def _hidden_files(self) -> bool:
|
||
code, out = _run(["reg", "query",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion"
|
||
r"\Explorer\Advanced", "/v", "Hidden"])
|
||
if code != 0 or "REG_DWORD" not in out:
|
||
return False
|
||
tail = out.split("REG_DWORD")[-1]
|
||
return "0x1" in tail
|
||
|
||
def _do_hidden_files(self, params) -> str:
|
||
want = 2 if not self._truthy(params.get("on")) else 1
|
||
_run(["reg", "add",
|
||
r"HKCU\Software\Microsoft\Windows\CurrentVersion\Explorer\Advanced",
|
||
"/v", "Hidden", "/t", "REG_DWORD", "/d", str(want), "/f"])
|
||
self._restart_explorer_quiet()
|
||
return "已显示隐藏文件" if want == 1 else "已恢复不显示隐藏文件"
|
||
|
||
def _restart_explorer_quiet(self) -> None:
|
||
_run(["taskkill", "/F", "/IM", "explorer.exe"])
|
||
time.sleep(0.8)
|
||
subprocess.Popen("explorer.exe", shell=True, creationflags=_NO_WINDOW)
|
||
|
||
def _do_sleep(self, params) -> str:
|
||
_run(["rundll32.exe", "powrprof.dll,SetSuspendState", "0,1,0"])
|
||
return "已进入睡眠（台式机可能不支持）"
|
||
|
||
def _do_hibernate(self, params) -> str:
|
||
code, out = _run(["shutdown", "/h"])
|
||
return "已进入休眠" if code == 0 else "这台机器没启用休眠"
|
||
|
||
def _do_lock_only(self, params) -> str:
|
||
ctypes.windll.user32.LockWorkStation()
|
||
return "已锁屏"
|
||
|
||
def _do_sign_out(self, params) -> str:
|
||
_run(["shutdown", "/l"])
|
||
return "已注销"
|
||
|
||
def _do_restart_graphics(self, params) -> str:
|
||
self._key_combo(0x5B, 0x11, 0x10, 0x42) # Win+Ctrl+Shift+B
|
||
time.sleep(0.5)
|
||
return "已重启显卡驱动"
|
||
|
||
# ==================================================================
|
||
# 网络
|
||
# ==================================================================
|
||
|
||
def _do_wifi_password(self, params) -> str:
|
||
code, out = _run(["netsh", "wlan", "show", "profiles"])
|
||
if code != 0:
|
||
return "读不到 WiFi 配置"
|
||
names = []
|
||
for raw in out.splitlines():
|
||
text = raw.strip()
|
||
if text.startswith("All User Profile") or "所有用户配置文件" in text:
|
||
names.append(text.split(":", 1)[-1].strip())
|
||
if not names:
|
||
return "没有保存过 WiFi"
|
||
lines = []
|
||
for name in names[:6]:
|
||
_code, detail = _run(["netsh", "wlan", "show", "profile",
|
||
f"name={name}", "key=clear"])
|
||
pwd = ""
|
||
for raw in detail.splitlines():
|
||
text = raw.strip()
|
||
if text.startswith("Key Content") or "关键内容" in text:
|
||
pwd = text.split(":", 1)[-1].strip()
|
||
lines.append(f"{name}：{pwd or '(读不到)'}")
|
||
return "\n".join(lines)
|
||
|
||
def _do_public_ip(self, params) -> str:
|
||
import urllib.request
|
||
for url in ("https://api.ipify.org", "https://ifconfig.me/ip"):
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=6) as resp:
|
||
text = resp.read().decode("utf-8", errors="replace").strip()
|
||
if text:
|
||
return f"公网 IP：{text}"
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
return "查不到（可能没联网）"
|
||
|
||
def _do_ping_host(self, params) -> str:
|
||
host = str(params.get("host") or "").strip()
|
||
if not host:
|
||
raise ValueError("地址为空")
|
||
if not re.match(r"^[A-Za-z0-9_.:\-]+$", host):
|
||
raise ValueError("地址含有不合法字符")
|
||
_code, out = _run(["ping", "-n", "3", host], timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines[-8:]) or "没有输出"
|
||
|
||
def _do_arp_devices(self, params) -> str:
|
||
_code, out = _run(["arp", "-a"])
|
||
lines = [l.strip() for l in out.splitlines() if l.strip()]
|
||
lines = [l for l in lines if "---" not in l and "Interface" not in l
|
||
and "接口" not in l]
|
||
return f"局域网设备 {len(lines)} 条：\n" + "\n".join(lines[:40])
|
||
|
||
def _do_net_adapters(self, params) -> str:
|
||
script = ("Get-NetAdapter | ForEach-Object { "
|
||
"\"$($_.Name) $($_.Status) $($_.LinkSpeed)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script])
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到网卡"
|
||
|
||
# ==================================================================
|
||
# 文件（继续限定在自己的目录内）
|
||
# ==================================================================
|
||
|
||
def _do_copy_file(self, params) -> str:
|
||
import shutil as _shutil
|
||
source = self.resolve_in_roots(str(params.get("path") or ""))
|
||
target = self.resolve_in_roots(str(params.get("to") or ""))
|
||
if not source.exists():
|
||
raise ValueError("源文件不存在")
|
||
self.assert_not_root(source)
|
||
self.assert_not_root(target)
|
||
if source == target:
|
||
raise ValueError("源和目标是同一个位置")
|
||
if target.exists():
|
||
raise ValueError("目标已存在，不覆盖（先改名或删掉）")
|
||
try:
|
||
if source.is_dir():
|
||
_shutil.copytree(source, target)
|
||
else:
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
_shutil.copy2(source, target)
|
||
except OSError as exc:
|
||
raise ValueError(f"复制失败: {exc}") from exc
|
||
return f"已复制到 {target}"
|
||
|
||
def _do_move_file(self, params) -> str:
|
||
import shutil as _shutil
|
||
source = self.resolve_in_roots(str(params.get("path") or ""))
|
||
target = self.resolve_in_roots(str(params.get("to") or ""))
|
||
if not source.exists():
|
||
raise ValueError("源文件不存在")
|
||
self.assert_not_root(source)
|
||
self.assert_not_root(target)
|
||
if source == target:
|
||
raise ValueError("源和目标是同一个位置")
|
||
if target.exists():
|
||
raise ValueError("目标已存在，不覆盖（先改名或删掉）")
|
||
try:
|
||
target.parent.mkdir(parents=True, exist_ok=True)
|
||
_shutil.move(str(source), str(target))
|
||
except OSError as exc:
|
||
raise ValueError(f"移动失败: {exc}") from exc
|
||
return f"已移动到 {target}"
|
||
|
||
def _do_zip_dir(self, params) -> str:
|
||
import zipfile
|
||
source = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not source.is_dir():
|
||
raise ValueError("不是文件夹")
|
||
out = source.parent / f"{source.name}.zip"
|
||
if out.exists():
|
||
raise ValueError("同名的 zip 已存在，先改名或删掉")
|
||
try:
|
||
with zipfile.ZipFile(out, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for item in source.rglob("*"):
|
||
if item.is_file():
|
||
zf.write(item, item.relative_to(source.parent))
|
||
size_kb = out.stat().st_size // 1024
|
||
except OSError as exc:
|
||
raise ValueError(f"压缩失败: {exc}") from exc
|
||
return f"已压缩为 {out.name}（{size_kb} KB）"
|
||
|
||
def _do_unzip(self, params) -> str:
|
||
import zipfile
|
||
source = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not source.is_file() or source.suffix.lower() != ".zip":
|
||
raise ValueError("不是 zip 文件")
|
||
target = source.parent / source.stem
|
||
target.mkdir(parents=True, exist_ok=True)
|
||
with zipfile.ZipFile(source) as zf:
|
||
zf.extractall(target)
|
||
return f"已解压到 {target.name}"
|
||
|
||
def _do_file_info(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.exists():
|
||
raise ValueError("文件不存在")
|
||
stat = target.stat()
|
||
size = stat.st_size
|
||
readable = (f"{size / 1048576:.2f} MB" if size > 1048576
|
||
else f"{size // 1024} KB")
|
||
kind = "文件夹" if target.is_dir() else "文件"
|
||
return (f"{kind}：{target.name}\n大小：{readable}\n"
|
||
f"修改：{time.strftime('%Y-%m-%d %H:%M', time.localtime(stat.st_mtime))}\n"
|
||
f"位置：{target.parent}")
|
||
|
||
def _do_folder_size(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.is_dir():
|
||
raise ValueError("不是文件夹")
|
||
total = 0
|
||
count = 0
|
||
deadline = time.time() + 15 # 大目录别把请求线程占死
|
||
try:
|
||
for item in target.rglob("*"):
|
||
if time.time() > deadline:
|
||
break
|
||
try:
|
||
if item.is_file():
|
||
total += item.stat().st_size
|
||
count += 1
|
||
except OSError:
|
||
continue
|
||
except OSError as exc:
|
||
raise ValueError(f"算不了: {exc}") from exc
|
||
return f"{target.name}：{count} 个文件，共 {total / 1048576:.1f} MB"
|
||
|
||
def _do_open_recycle(self, params) -> str:
|
||
subprocess.Popen(["explorer.exe", "shell:RecycleBinFolder"],
|
||
creationflags=_NO_WINDOW)
|
||
return "已打开回收站"
|
||
|
||
def _do_disk_by_type(self, params) -> str:
|
||
buckets = {"图片": 0, "视频": 0, "音频": 0, "文档": 0, "压缩包": 0, "其他": 0}
|
||
exts = {
|
||
"图片": {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".webp"},
|
||
"视频": {".mp4", ".avi", ".mkv", ".mov", ".wmv", ".flv"},
|
||
"音频": {".mp3", ".wav", ".flac", ".m4a", ".aac"},
|
||
"文档": {".doc", ".docx", ".pdf", ".ppt", ".pptx", ".xls",
|
||
".xlsx", ".txt", ".md"},
|
||
"压缩包": {".zip", ".rar", ".7z", ".tar", ".gz"},
|
||
}
|
||
for root in self.allowed_roots():
|
||
try:
|
||
for item in root.rglob("*"):
|
||
if not item.is_file():
|
||
continue
|
||
suffix = item.suffix.lower()
|
||
for name, group in exts.items():
|
||
if suffix in group:
|
||
buckets[name] += item.stat().st_size
|
||
break
|
||
else:
|
||
buckets["其他"] += item.stat().st_size
|
||
except OSError:
|
||
continue
|
||
parts = [f"{k} {v / 1048576:.0f}MB" for k, v in buckets.items() if v]
|
||
return " | ".join(parts) or "统计不到"
|
||
|
||
# ==================================================================
|
||
# 工具
|
||
# ==================================================================
|
||
|
||
def _do_password_gen(self, params) -> str:
|
||
import secrets
|
||
import string
|
||
alphabet = string.ascii_letters + string.digits + "!@#$%^&*-_"
|
||
pwd = "".join(secrets.choice(alphabet) for _ in range(16))
|
||
self.actions.clip_set(pwd)
|
||
return f"已生成并复制到剪贴板：\n{pwd}"
|
||
|
||
def _do_stopwatch(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="00:00.0", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Consolas", 130, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"start": time.time()}
|
||
|
||
def tick():
|
||
elapsed = time.time() - state["start"]
|
||
mins, secs = divmod(int(elapsed), 60)
|
||
label.configure(
|
||
text=f"{mins:02d}:{secs:02d}.{int(elapsed * 10) % 10}")
|
||
win.after(100, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "秒表已打开（Esc 关闭）"
|
||
|
||
def _do_world_clock(self, params) -> str:
|
||
"""写死偏移会忽略夏令时（现在纽约/伦敦/悉尼都差 1 小时），改用 zoneinfo。"""
|
||
zones = [("北京", "Asia/Shanghai"), ("伦敦", "Europe/London"),
|
||
("纽约", "America/New_York"), ("洛杉矶", "America/Los_Angeles"),
|
||
("东京", "Asia/Tokyo"), ("悉尼", "Australia/Sydney")]
|
||
try:
|
||
from zoneinfo import ZoneInfo
|
||
import datetime as _dt
|
||
now = _dt.datetime.now(_dt.timezone.utc)
|
||
lines = []
|
||
for name, zone in zones:
|
||
local = now.astimezone(ZoneInfo(zone))
|
||
lines.append(f"{name} {local.strftime('%H:%M')}"
|
||
f"{'（夏令时）' if local.dst() else ''}")
|
||
return "\n".join(lines)
|
||
except Exception: # noqa: BLE001
|
||
# 没有时区数据库时退回 UTC 偏移（可能差 1 小时）
|
||
now = time.time()
|
||
fallback = [("北京", 8), ("伦敦", 0), ("纽约", -5), ("洛杉矶", -8),
|
||
("东京", 9), ("悉尼", 10)]
|
||
return "\n".join(
|
||
f"{n} {time.strftime('%H:%M', time.gmtime(now + o * 3600))}"
|
||
for n, o in fallback)
|
||
|
||
def _do_hardware_info(self, params) -> str:
|
||
script = (
|
||
"$c=Get-CimInstance Win32_Processor|Select-Object -First 1;"
|
||
"$m=(Get-CimInstance Win32_PhysicalMemory|Measure-Object -Property "
|
||
"Capacity -Sum).Sum/1GB;"
|
||
"$g=Get-CimInstance Win32_VideoController|Select-Object -First 1;"
|
||
"\"CPU: $($c.Name)\";"
|
||
"\"核心: $($c.NumberOfCores) 核 $($c.NumberOfLogicalProcessors) 线程\";"
|
||
"\"内存: $([math]::Round($m,1)) GB\";"
|
||
"\"显卡: $($g.Name)\"")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到硬件信息"
|
||
|
||
def _do_startup_items(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_StartupCommand | "
|
||
"ForEach-Object { \"$($_.Name) -> $($_.Command)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
if not lines:
|
||
return "没有启动项"
|
||
return f"启动项 {len(lines)} 个：\n" + "\n".join(l[:100] for l in lines[:30])
|
||
|
||
def _do_scheduled_tasks(self, params) -> str:
|
||
script = ("Get-ScheduledTask | Where-Object {$_.State -ne 'Disabled'} | "
|
||
"Select-Object -First 30 -ExpandProperty TaskName")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"计划任务 {len(lines)} 个：\n" + "\n".join(lines)
|
||
|
||
def _do_system_detail(self, params) -> str:
|
||
import platform
|
||
_code, out = _run(["wmic", "os", "get",
|
||
"Caption,Version,BuildNumber,OSArchitecture",
|
||
"/format:list"])
|
||
lines = [l.strip() for l in out.splitlines() if l.strip()]
|
||
return (f"{platform.system()} {platform.release()} {platform.machine()}\n"
|
||
+ "\n".join(lines))
|
||
|
||
# ==================================================================
|
||
# 媒体
|
||
# ==================================================================
|
||
|
||
def _do_stop_after(self, params) -> str:
|
||
try:
|
||
minutes = max(1, min(180, int(params.get("minutes") or 10)))
|
||
except (TypeError, ValueError):
|
||
minutes = 10
|
||
|
||
import threading
|
||
|
||
def fire():
|
||
self.actions.stop_audio()
|
||
|
||
timer = threading.Timer(minutes * 60, fire)
|
||
timer.daemon = True
|
||
timer.start()
|
||
self._timers.append(timer)
|
||
return f"{minutes} 分钟后停止播放"
|
||
|
||
def _do_fade_volume(self, params) -> str:
|
||
for _ in range(20):
|
||
self._volume_key(0xAE, 1)
|
||
time.sleep(0.05)
|
||
return "音量已淡出到最低"
|
||
|
||
def _do_gpu_info(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_VideoController | ForEach-Object { "
|
||
"\"$($_.Name) 驱动 $($_.DriverVersion)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到显卡信息"
|
||
|
||
# ==================================================================
|
||
# 娱乐
|
||
# ==================================================================
|
||
|
||
def _do_danmaku(self, params) -> str:
|
||
text = str(params.get("text") or "").strip() or "大家好"
|
||
speed = 6
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#000000")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#000000", highlightthickness=0,
|
||
width=width, height=height)
|
||
canvas.pack(fill="both", expand=True)
|
||
canvas.configure(highlightthickness=0)
|
||
items = []
|
||
colors = ["#FFFFFF", "#FFD33D", "#5AC8FA", "#FF6B6B", "#8EE36A"]
|
||
for i in range(14):
|
||
y = int(height * (i + 1) / 15)
|
||
item = canvas.create_text(width + 200, y, text=text,
|
||
fill=colors[i % len(colors)],
|
||
font=("Microsoft YaHei UI", 26, "bold"),
|
||
anchor="w")
|
||
items.append([item, speed + (i % 4)])
|
||
|
||
def tick():
|
||
alive = False
|
||
for entry in items:
|
||
canvas.move(entry[0], -entry[1], 0)
|
||
x, _y = canvas.coords(entry[0])
|
||
if x > -600:
|
||
alive = True
|
||
if alive:
|
||
win.after(30, tick)
|
||
else:
|
||
close()
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"弹幕已开始：{text[:20]}"
|
||
|
||
def _do_marquee(self, params) -> str:
|
||
text = str(params.get("text") or "").strip() or "欢迎"
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#0B1020", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
item = canvas.create_text(width, height // 2, text=text,
|
||
fill="#FFD33D",
|
||
font=("Microsoft YaHei UI", 60, "bold"),
|
||
anchor="w")
|
||
|
||
def tick():
|
||
canvas.move(item, -4, 0)
|
||
x, _y = canvas.coords(item)
|
||
if x < -len(text) * 70:
|
||
close()
|
||
return
|
||
win.after(25, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"字幕滚动中：{text[:20]}"
|
||
|
||
def _do_lottery(self, params) -> str:
|
||
import random
|
||
names = self.store.get("roll_call_names") or []
|
||
pool = names or [str(i) for i in range(1, 51)]
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFD33D",
|
||
font=("Microsoft YaHei UI", 120, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 0, "winner": random.choice(pool)}
|
||
|
||
def tick():
|
||
state["n"] += 1
|
||
if state["n"] <= 22:
|
||
label.configure(text=random.choice(pool), fg="#FFD33D")
|
||
win.after(60 + state["n"] * 12, tick)
|
||
else:
|
||
label.configure(text=state["winner"], fg="#5AC8FA")
|
||
try:
|
||
win.bell()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "抽奖中…"
|
||
|
||
def _do_dice(self, params) -> str:
|
||
import random
|
||
faces = ["⚀", "⚁", "⚂", "⚃", "⚄", "⚅"]
|
||
result = random.randint(1, 6)
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Segoe UI Symbol", 200, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 0, "final": faces[result - 1]}
|
||
|
||
def tick():
|
||
state["n"] += 1
|
||
if state["n"] <= 14:
|
||
label.configure(text=random.choice(faces))
|
||
win.after(70, tick)
|
||
else:
|
||
label.configure(text=state["final"], fg="#5AC8FA")
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"掷出 {result} 点"
|
||
|
||
def _do_coin_flip(self, params) -> str:
|
||
import random
|
||
result = random.choice(["正", "反"])
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Microsoft YaHei UI", 200, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 0}
|
||
|
||
def tick():
|
||
state["n"] += 1
|
||
if state["n"] <= 16:
|
||
label.configure(text=random.choice(["正", "反"]))
|
||
win.after(80, tick)
|
||
else:
|
||
label.configure(text=result, fg="#FFD33D")
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return f"硬币：{result}"
|
||
|
||
def _do_countdown_321(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
label = tk.Label(win, text="", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Consolas", 260, "bold"))
|
||
label.pack(expand=True, fill="both")
|
||
state = {"n": 3}
|
||
|
||
def tick():
|
||
if state["n"] <= 0:
|
||
label.configure(text="开始！", fg="#8EE36A",
|
||
font=("Microsoft YaHei UI", 120, "bold"))
|
||
win.after(900, close)
|
||
return
|
||
label.configure(text=str(state["n"]),
|
||
fg="#FF6B6B" if state["n"] == 1 else "#FFFFFF")
|
||
try:
|
||
win.bell()
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
state["n"] -= 1
|
||
win.after(1000, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "倒计时开始"
|
||
|
||
def _do_cinema_mode(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#000000")
|
||
tk.Label(win, text="影院模式", bg="#000000", fg="#1A1A1A",
|
||
font=("Microsoft YaHei UI", 40, "bold")).pack(
|
||
expand=True, fill="both")
|
||
|
||
self._fullscreen_window(builder)
|
||
return "影院模式（Esc 退出）"
|
||
|
||
def _do_random_song(self, params) -> str:
|
||
import random
|
||
files = self.actions.list_local_audio()
|
||
if not files:
|
||
raise ValueError("音乐目录里没有音频文件")
|
||
picked = random.choice(files)
|
||
self.actions.play_audio(picked["path"])
|
||
return f"正在播放：{picked['name']}"
|
||
|
||
def _do_play_video(self, params) -> str:
|
||
target = self.resolve_in_roots(str(params.get("path") or ""))
|
||
if not target.is_file():
|
||
raise ValueError("视频文件不存在")
|
||
# 之前走 cmd /c start，路径里的 & | ^ 会被 cmd 当成命令分隔符，
|
||
# 一个文件名就能变成任意命令执行。os.startfile 不经过 shell。
|
||
try:
|
||
os.startfile(str(target)) # noqa: S606
|
||
except OSError as exc:
|
||
raise ValueError(f"打不开: {exc}") from exc
|
||
return f"已用播放器打开 {target.name}"
|
||
|
||
def _scores(self) -> dict:
|
||
value = self.store.get("score_board")
|
||
if not isinstance(value, dict):
|
||
value = {"A": 0, "B": 0}
|
||
value.setdefault("A", 0)
|
||
value.setdefault("B", 0)
|
||
return value
|
||
|
||
def _show_score_board(self) -> None:
|
||
scores = self._scores()
|
||
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
row = tk.Frame(win, bg="#0B1020")
|
||
row.pack(expand=True, fill="both")
|
||
for team, color in (("A", "#5AC8FA"), ("B", "#FFD33D")):
|
||
box = tk.Frame(row, bg="#0B1020")
|
||
box.pack(side="left", expand=True, fill="both")
|
||
tk.Label(box, text=f"{team} 队", bg="#0B1020", fg=color,
|
||
font=("Microsoft YaHei UI", 40, "bold")).pack(pady=(60, 0))
|
||
tk.Label(box, text=str(scores.get(team, 0)), bg="#0B1020",
|
||
fg="#FFFFFF", font=("Consolas", 200, "bold")).pack(
|
||
expand=True)
|
||
|
||
self._fullscreen_window(builder)
|
||
|
||
def _do_score_board(self, params) -> str:
|
||
self._show_score_board()
|
||
scores = self._scores()
|
||
return f"A 队 {scores['A']} : {scores['B']} B 队"
|
||
|
||
def _do_score_change(self, params) -> str:
|
||
team = str(params.get("team") or "A").strip().upper()
|
||
if team not in ("A", "B"):
|
||
raise ValueError("队伍只能填 A 或 B")
|
||
try:
|
||
delta = int(params.get("delta") or 0)
|
||
except (TypeError, ValueError):
|
||
raise ValueError("分数要填数字")
|
||
scores = self._scores()
|
||
scores[team] = scores[team] + delta
|
||
self.store.set("score_board", scores)
|
||
self._show_score_board()
|
||
return f"A 队 {scores['A']} : {scores['B']} B 队"
|
||
|
||
def _do_cheer(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
import random
|
||
win.configure(bg="#0B1020")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#0B1020", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
canvas.create_text(width // 2, height // 2, text="加 油 !",
|
||
fill="#FFD33D",
|
||
font=("Microsoft YaHei UI", 120, "bold"))
|
||
colors = ["#FF6B6B", "#FFD33D", "#5AC8FA", "#8EE36A", "#FFFFFF"]
|
||
flakes = []
|
||
for _ in range(90):
|
||
x = random.randint(0, width)
|
||
y = random.randint(-height, 0)
|
||
item = canvas.create_rectangle(x, y, x + 10, y + 16,
|
||
fill=random.choice(colors),
|
||
outline="")
|
||
flakes.append(item)
|
||
|
||
def tick():
|
||
for item in flakes:
|
||
canvas.move(item, 0, 7)
|
||
_x, y = canvas.coords(item)[:2]
|
||
if y > height:
|
||
canvas.coords(item, random.randint(0, width), -20,
|
||
random.randint(0, width) + 10, 0)
|
||
win.after(40, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "加油！"
|
||
|
||
# ==================================================================
|
||
# 极客
|
||
# ==================================================================
|
||
|
||
def _do_matrix_rain(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
import random
|
||
win.configure(bg="#000000")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#000000", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
font_size = 18
|
||
columns = max(1, width // font_size)
|
||
drops = [random.randint(-40, 0) for _ in range(columns)]
|
||
chars = "01アイウエオカキクケコサシスセソタチツテト"
|
||
|
||
def tick():
|
||
canvas.delete("fade")
|
||
for i, y in enumerate(drops):
|
||
char = random.choice(chars)
|
||
canvas.create_text(i * font_size + font_size // 2,
|
||
y * font_size, text=char,
|
||
fill="#00FF66",
|
||
font=("Consolas", font_size, "bold"),
|
||
tags="fade")
|
||
drops[i] = y + 1 if y * font_size < height else random.randint(-20, 0)
|
||
win.after(60, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "数字雨已开始（Esc 退出）"
|
||
|
||
def _do_starfield(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
import random
|
||
win.configure(bg="#000010")
|
||
width = win.winfo_screenwidth()
|
||
height = win.winfo_screenheight()
|
||
canvas = tk.Canvas(win, bg="#000010", highlightthickness=0)
|
||
canvas.pack(fill="both", expand=True)
|
||
stars = []
|
||
for _ in range(240):
|
||
x = random.uniform(-width, width)
|
||
y = random.uniform(-height, height)
|
||
z = random.uniform(1, width)
|
||
item = canvas.create_oval(0, 0, 0, 0, fill="#FFFFFF", outline="")
|
||
stars.append([x, y, z, item])
|
||
|
||
def tick():
|
||
for star in stars:
|
||
star[2] -= 16
|
||
if star[2] <= 1:
|
||
star[0] = random.uniform(-width, width)
|
||
star[1] = random.uniform(-height, height)
|
||
star[2] = width
|
||
k = 220 / star[2]
|
||
px = star[0] * k + width / 2
|
||
py = star[1] * k + height / 2
|
||
size = max(1, int((1 - star[2] / width) * 4))
|
||
canvas.coords(star[3], px, py, px + size, py + size)
|
||
win.after(33, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "星空穿梭中（Esc 退出）"
|
||
|
||
def _do_dashboard(self, params) -> str:
|
||
def builder(win, close):
|
||
import tkinter as tk
|
||
win.configure(bg="#0B1020")
|
||
frame = tk.Frame(win, bg="#0B1020")
|
||
frame.pack(expand=True, fill="both", padx=60, pady=60)
|
||
titles = {}
|
||
for key, title, color in (("cpu", "CPU", "#5AC8FA"),
|
||
("mem", "内存", "#FFD33D"),
|
||
("net", "网络", "#8EE36A")):
|
||
box = tk.Frame(frame, bg="#0B1020")
|
||
box.pack(fill="x", pady=16)
|
||
tk.Label(box, text=title, bg="#0B1020", fg=color,
|
||
font=("Microsoft YaHei UI", 28, "bold"),
|
||
width=6, anchor="w").pack(side="left")
|
||
value = tk.Label(box, text="--", bg="#0B1020", fg="#FFFFFF",
|
||
font=("Consolas", 60, "bold"), anchor="w")
|
||
value.pack(side="left", fill="x", expand=True)
|
||
titles[key] = value
|
||
|
||
state = {"last": None, "last_bytes": None}
|
||
|
||
def tick():
|
||
info = self.actions.perf()
|
||
titles["cpu"].configure(text=f"{info.get('cpu_percent', 0)} %")
|
||
titles["mem"].configure(
|
||
text=f"{info.get('mem_percent', 0)} % "
|
||
f"({info.get('mem_used_gb')}/{info.get('mem_total_gb')} GB)")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command",
|
||
"(Get-NetAdapterStatistics | Measure-Object "
|
||
"-Property ReceivedBytes -Sum).Sum"])
|
||
digits = re.findall(r"\d+", out or "")
|
||
now_bytes = int(digits[0]) if digits else 0
|
||
if state["last_bytes"] is not None and state["last"] is not None:
|
||
delta = now_bytes - state["last_bytes"]
|
||
elapsed = max(0.1, time.time() - state["last"])
|
||
mbps = delta / elapsed / 1048576
|
||
titles["net"].configure(text=f"{mbps:.2f} MB/s")
|
||
state["last_bytes"] = now_bytes
|
||
state["last"] = time.time()
|
||
win.after(1500, tick)
|
||
|
||
tick()
|
||
|
||
self._fullscreen_window(builder)
|
||
return "实时监控已打开（Esc 退出）"
|
||
|
||
def _do_qr_code(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
import tempfile
|
||
import urllib.parse
|
||
import urllib.request
|
||
url = ("https://api.qrserver.com/v1/create-qr-code/?size=520x520&data="
|
||
+ urllib.parse.quote(text))
|
||
target = Path(tempfile.gettempdir()) / "_lanremote_qr.png"
|
||
try:
|
||
with urllib.request.urlopen(url, timeout=10) as resp:
|
||
target.write_bytes(resp.read())
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"生成二维码失败（要联网）: {exc}") from exc
|
||
self.actions.show_image(str(target), f"二维码 - {text[:20]}")
|
||
return f"二维码已显示（内容 {len(text)} 字）"
|
||
|
||
def _do_hash_text(self, params) -> str:
|
||
import hashlib
|
||
text = str(params.get("text") or "")
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
data = text.encode("utf-8")
|
||
return ("MD5: " + hashlib.md5(data).hexdigest()
|
||
+ "\nSHA1: " + hashlib.sha1(data).hexdigest()
|
||
+ "\nSHA256: " + hashlib.sha256(data).hexdigest())
|
||
|
||
def _do_base64_encode(self, params) -> str:
|
||
import base64
|
||
text = str(params.get("text") or "")
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
result = base64.b64encode(text.encode("utf-8")).decode("ascii")
|
||
self.actions.clip_set(result)
|
||
return result
|
||
|
||
def _do_base64_decode(self, params) -> str:
|
||
import base64
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
try:
|
||
result = base64.b64decode(text).decode("utf-8", errors="replace")
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"不是合法的 Base64: {exc}") from exc
|
||
self.actions.clip_set(result)
|
||
return result[:500]
|
||
|
||
def _do_uuid_gen(self, params) -> str:
|
||
import uuid
|
||
value = str(uuid.uuid4())
|
||
self.actions.clip_set(value)
|
||
return f"已生成并复制：\n{value}"
|
||
|
||
def _do_timestamp_now(self, params) -> str:
|
||
now = time.time()
|
||
return (f"秒：{int(now)}\n毫秒：{int(now * 1000)}\n"
|
||
f"本地：{time.strftime('%Y-%m-%d %H:%M:%S')}\n"
|
||
f"UTC：{time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(now))}")
|
||
|
||
def _do_json_format(self, params) -> str:
|
||
import json
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
try:
|
||
obj = json.loads(text)
|
||
except ValueError as exc:
|
||
raise ValueError(f"不是合法 JSON: {exc}") from exc
|
||
pretty = json.dumps(obj, ensure_ascii=False, indent=2)
|
||
self.actions.clip_set(pretty)
|
||
return pretty[:800]
|
||
|
||
def _do_screen_color(self, params) -> str:
|
||
if not IS_WINDOWS:
|
||
raise ValueError("仅支持 Windows")
|
||
|
||
class Point(ctypes.Structure):
|
||
_fields_ = [("x", ctypes.c_long), ("y", ctypes.c_long)]
|
||
|
||
point = Point()
|
||
ctypes.windll.user32.GetCursorPos(ctypes.byref(point))
|
||
hdc = ctypes.windll.user32.GetDC(0)
|
||
color = ctypes.windll.gdi32.GetPixel(hdc, point.x, point.y)
|
||
ctypes.windll.user32.ReleaseDC(0, hdc)
|
||
if color < 0:
|
||
raise ValueError("取色失败")
|
||
r = color & 0xFF
|
||
g = (color >> 8) & 0xFF
|
||
b = (color >> 16) & 0xFF
|
||
hex_color = f"#{r:02X}{g:02X}{b:02X}"
|
||
self.actions.clip_set(hex_color)
|
||
return (f"鼠标位置 ({point.x}, {point.y})\n"
|
||
f"颜色 {hex_color}\nRGB({r}, {g}, {b})")
|
||
|
||
def _do_speed_test(self, params) -> str:
|
||
import urllib.request
|
||
url = "https://speed.cloudflare.com/__down?bytes=5000000"
|
||
try:
|
||
start = time.time()
|
||
with urllib.request.urlopen(url, timeout=20) as resp:
|
||
total = 0
|
||
while True:
|
||
chunk = resp.read(65536)
|
||
if not chunk:
|
||
break
|
||
total += len(chunk)
|
||
if time.time() - start > 12:
|
||
break
|
||
elapsed = max(0.01, time.time() - start)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"测速失败（要联网）: {exc}") from exc
|
||
mbps = total * 8 / elapsed / 1_000_000
|
||
return (f"下载 {total / 1048576:.1f} MB，用时 {elapsed:.1f} 秒\n"
|
||
f"约 {mbps:.1f} Mbps（{total / elapsed / 1048576:.2f} MB/s）")
|
||
|
||
def _do_port_check(self, params) -> str:
|
||
import socket
|
||
host = str(params.get("host") or "").strip()
|
||
if not host:
|
||
raise ValueError("地址为空")
|
||
if not re.match(r"^[A-Za-z0-9_.:\-]+$", host):
|
||
raise ValueError("地址含有不合法字符")
|
||
common = [21, 22, 23, 25, 53, 80, 110, 135, 139, 443, 445, 1433,
|
||
3306, 3389, 5432, 5900, 6379, 8080, 8765]
|
||
open_ports = []
|
||
for port in common:
|
||
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
sock.settimeout(0.45)
|
||
try:
|
||
if sock.connect_ex((host, port)) == 0:
|
||
open_ports.append(port)
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
sock.close()
|
||
if not open_ports:
|
||
return f"{host}：常用端口都没开"
|
||
return f"{host} 开放端口：{', '.join(str(p) for p in open_ports)}"
|
||
|
||
# ==================================================================
|
||
# 静默运行
|
||
# ==================================================================
|
||
|
||
def _autostart_on(self) -> bool:
|
||
try:
|
||
from .autostart import is_enabled
|
||
return is_enabled()
|
||
except Exception: # noqa: BLE001
|
||
return False
|
||
|
||
def _do_autostart(self, params) -> str:
|
||
from .autostart import toggle
|
||
return toggle(self._truthy(params.get("on")))
|
||
|
||
def _do_autostart_status(self, params) -> str:
|
||
from .autostart import status as auto_status
|
||
info = auto_status()
|
||
lines = [
|
||
f"开机自启：{'已开启' if info['enabled'] else '未开启'}",
|
||
f"已配设备：{len(self.store.device_list())} 台",
|
||
f"允许新配对："
|
||
f"{'是' if self.store.get('allow_new_pairing', True) else '否'}",
|
||
f"监听端口：{self.store.get('port', 8765)}",
|
||
]
|
||
if info["enabled"]:
|
||
lines.append(f"自启命令：{info['command']}")
|
||
return "\n".join(lines)
|
||
|
||
def _do_quit_agent(self, params) -> str:
|
||
"""退出被控端进程。"""
|
||
import os
|
||
import threading
|
||
|
||
def later():
|
||
time.sleep(0.6) # 先让这条回复发出去
|
||
os._exit(0)
|
||
|
||
threading.Thread(target=later, daemon=True, name="quit").start()
|
||
return "被控端正在退出"
|
||
|
||
def _do_silent_help(self, params) -> str:
|
||
return ("静默模式：没有任何窗口，也没有终端，只在后台监听。\n"
|
||
"已经配过的手表会自动重连，不用再输配对码。\n"
|
||
"加新设备要先退出，用带窗口的方式启动，"
|
||
"或运行 ykxmz.exe --add-device 10")
|
||
|
||
# ==================================================================
|
||
# 查看（续）
|
||
# ==================================================================
|
||
|
||
def _do_screen_resolution(self, params) -> str:
|
||
if not IS_WINDOWS:
|
||
return "仅支持 Windows"
|
||
user32 = ctypes.windll.user32
|
||
user32.SetProcessDPIAware()
|
||
width = user32.GetSystemMetrics(0)
|
||
height = user32.GetSystemMetrics(1)
|
||
monitors = user32.GetSystemMetrics(80)
|
||
return f"主显示器 {width} x {height}\n显示器数量：{monitors}"
|
||
|
||
def _do_list_printers(self, params) -> str:
|
||
script = ("Get-Printer | ForEach-Object { "
|
||
"\"$($_.Name) [$($_.Status)]\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"打印机 {len(lines)} 台：\n" + "\n".join(lines) if lines else "没有打印机"
|
||
|
||
def _do_list_drives(self, params) -> str:
|
||
script = ("Get-Volume | Where-Object {$_.DriveLetter} | ForEach-Object { "
|
||
"\"$($_.DriveLetter): $($_.FileSystemLabel) "
|
||
"$([math]::Round($_.SizeRemaining/1GB,1))/"
|
||
"$([math]::Round($_.Size/1GB,1)) GB\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到驱动器"
|
||
|
||
def _do_audio_devices(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_SoundDevice | ForEach-Object { "
|
||
"\"$($_.Name) [$($_.Status)]\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到音频设备"
|
||
|
||
def _do_usb_devices(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_USBControllerDevice | "
|
||
"ForEach-Object { [wmi]$_.Dependent } | "
|
||
"Select-Object -ExpandProperty Name -Unique")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"USB 设备 {len(lines)} 个：\n" + "\n".join(lines[:30])
|
||
|
||
def _do_installed_drivers(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_PnPSignedDriver | "
|
||
"Where-Object {$_.DeviceName} | "
|
||
"Select-Object -ExpandProperty DeviceName -Unique")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=35)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"驱动 {len(lines)} 个（前 30）：\n" + "\n".join(lines[:30])
|
||
|
||
def _do_windows_detail(self, params) -> str:
|
||
script = ("$o=Get-CimInstance Win32_OperatingSystem;"
|
||
"\"版本: $($o.Caption)\";"
|
||
"\"构建: $($o.BuildNumber)\";"
|
||
"\"架构: $($o.OSArchitecture)\";"
|
||
"\"安装日期: $($o.InstallDate)\";"
|
||
"\"最后启动: $($o.LastBootUpTime)\"")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到"
|
||
|
||
def _do_recycle_size(self, params) -> str:
|
||
script = (
|
||
"$s=(New-Object -ComObject Shell.Application)"
|
||
".NameSpace(10).Items() | "
|
||
"Measure-Object -Property Size -Sum;"
|
||
"\"$($s.Count) 项，共 "
|
||
"$([math]::Round($s.Sum/1MB,1)) MB\"")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
text = out.strip().splitlines()
|
||
return text[-1] if text else "读不到回收站"
|
||
|
||
def _do_temp_size(self, params) -> str:
|
||
import os as _os
|
||
total = 0
|
||
count = 0
|
||
for root in (_os.environ.get("TEMP", ""), r"C:\Windows\Temp"):
|
||
if not root or not Path(root).is_dir():
|
||
continue
|
||
try:
|
||
for item in Path(root).rglob("*"):
|
||
try:
|
||
if item.is_file():
|
||
total += item.stat().st_size
|
||
count += 1
|
||
except OSError:
|
||
continue
|
||
except OSError:
|
||
continue
|
||
return f"临时文件 {count} 个，共 {total / 1048576:.1f} MB"
|
||
|
||
def _do_certificates(self, params) -> str:
|
||
script = ("Get-ChildItem Cert:\LocalMachine\Root | "
|
||
"Select-Object -First 25 -ExpandProperty Subject")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"受信任根证书（前 25 / 共若干）：\n" + "\n".join(lines)
|
||
|
||
# ==================================================================
|
||
# 网络（续）
|
||
# ==================================================================
|
||
|
||
def _do_open_ports(self, params) -> str:
|
||
script = ("Get-NetTCPConnection -State Listen | "
|
||
"Where-Object {$_.LocalAddress -eq '0.0.0.0'} | "
|
||
"Select-Object -ExpandProperty LocalPort -Unique | "
|
||
"Sort-Object")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
ports = [p.strip() for p in out.splitlines() if p.strip().isdigit()]
|
||
return f"对外监听端口 {len(ports)} 个：\n" + ", ".join(ports)
|
||
|
||
def _do_firewall_status(self, params) -> str:
|
||
script = ("Get-NetFirewallProfile | ForEach-Object { "
|
||
"\"$($_.Name): $($_.Enabled)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
if not lines:
|
||
# NetSecurity 模块坏掉时退化到 netsh
|
||
_code, out = _run(["netsh", "advfirewall", "show", "allprofiles",
|
||
"state"], timeout=25)
|
||
lines = [l.strip() for l in out.splitlines() if "ON" in l or "OFF" in l
|
||
or "开" in l or "关" in l]
|
||
return "\n".join(lines) or "读不到防火墙状态"
|
||
|
||
def _do_dns_servers(self, params) -> str:
|
||
script = ("Get-DnsClientServerAddress | "
|
||
"Where-Object {$_.ServerAddresses} | ForEach-Object { "
|
||
"\"$($_.InterfaceAlias): "
|
||
"$($_.ServerAddresses -join ', ')\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到 DNS"
|
||
|
||
def _do_hosts_file(self, params) -> str:
|
||
path = Path(r"C:\Windows\System32\drivers\etc\hosts")
|
||
if not path.is_file():
|
||
return "读不到 hosts"
|
||
try:
|
||
lines = [l.rstrip() for l in path.read_text(
|
||
encoding="utf-8", errors="replace").splitlines()
|
||
if l.strip() and not l.strip().startswith("#")]
|
||
except OSError as exc:
|
||
raise ValueError(f"读不到: {exc}") from exc
|
||
return (f"hosts 里 {len(lines)} 条有效记录：\n" + "\n".join(lines[:40])
|
||
if lines else "hosts 里没有自定义记录")
|
||
|
||
def _do_wifi_networks(self, params) -> str:
|
||
_code, out = _run(["netsh", "wlan", "show", "networks"], timeout=25)
|
||
lines = []
|
||
for raw in out.splitlines():
|
||
text = raw.strip()
|
||
if text.startswith("SSID") and "BSSID" not in text:
|
||
lines.append(text)
|
||
return (f"附近 {len(lines)} 个 WiFi：\n" + "\n".join(lines[:25])
|
||
if lines else "附近没扫到 WiFi")
|
||
|
||
def _do_bluetooth_devices(self, params) -> str:
|
||
script = ("Get-PnpDevice -Class Bluetooth -Status OK | "
|
||
"Select-Object -ExpandProperty FriendlyName")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return f"蓝牙设备 {len(lines)} 个：\n" + "\n".join(lines[:25]) if lines else "没有蓝牙设备"
|
||
|
||
# ==================================================================
|
||
# 系统（续）
|
||
# ==================================================================
|
||
|
||
def _open_msc(self, name: str, label: str) -> str:
|
||
try:
|
||
subprocess.Popen([name], shell=True, creationflags=_NO_WINDOW)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"打不开{label}: {exc}") from exc
|
||
return f"已打开{label}"
|
||
|
||
def _do_open_services(self, params) -> str:
|
||
return self._open_msc("services.msc", "服务")
|
||
|
||
def _do_open_task_scheduler(self, params) -> str:
|
||
return self._open_msc("taskschd.msc", "任务计划")
|
||
|
||
def _do_open_device_manager(self, params) -> str:
|
||
return self._open_msc("devmgmt.msc", "设备管理器")
|
||
|
||
def _do_open_disk_cleanup(self, params) -> str:
|
||
return self._open_msc("cleanmgr.exe", "磁盘清理")
|
||
|
||
def _do_open_windows_update(self, params) -> str:
|
||
subprocess.Popen(["cmd", "/c", "start", "ms-settings:windowsupdate"],
|
||
creationflags=_NO_WINDOW)
|
||
return "已打开系统更新"
|
||
|
||
def _do_clear_temp(self, params) -> str:
|
||
import os as _os
|
||
removed = 0
|
||
freed = 0
|
||
for root in (_os.environ.get("TEMP", ""), r"C:\Windows\Temp"):
|
||
if not root or not Path(root).is_dir():
|
||
continue
|
||
try:
|
||
for item in Path(root).iterdir():
|
||
try:
|
||
if item.is_file():
|
||
size = item.stat().st_size
|
||
item.unlink()
|
||
removed += 1
|
||
freed += size
|
||
elif item.is_dir():
|
||
import shutil as _shutil
|
||
_shutil.rmtree(item, ignore_errors=True)
|
||
removed += 1
|
||
except OSError:
|
||
continue # 正在被占用的跳过
|
||
except OSError:
|
||
continue
|
||
return f"清理了 {removed} 项，释放约 {freed / 1048576:.1f} MB"
|
||
|
||
def _do_env_vars(self, params) -> str:
|
||
names = ["PATH", "TEMP", "USERPROFILE", "COMPUTERNAME", "USERNAME",
|
||
"OS", "PROCESSOR_IDENTIFIER", "SystemRoot"]
|
||
lines = []
|
||
for name in names:
|
||
value = os.environ.get(name, "")
|
||
if name == "PATH":
|
||
value = f"（{len(value.split(os.pathsep))} 个路径）"
|
||
lines.append(f"{name} = {value[:80]}")
|
||
return "\n".join(lines)
|
||
|
||
# ==================================================================
|
||
# 教室管控
|
||
# ==================================================================
|
||
|
||
# 这些策略写在 HKCU 下，改完广播一次让 explorer 立刻生效
|
||
RESTRICT_KEYS = {
|
||
"hide_icons": ("HKCU\\Software\\Microsoft\\Windows\\CurrentVersion"
|
||
"\\Explorer\\Advanced", "HideIcons"),
|
||
"restrict_taskmgr": ("HKCU\\Software\\Microsoft\\Windows"
|
||
"\\CurrentVersion\\Policies\\System",
|
||
"DisableTaskMgr"),
|
||
"restrict_regedit": ("HKCU\\Software\\Microsoft\\Windows"
|
||
"\\CurrentVersion\\Policies\\System",
|
||
"DisableRegistryTools"),
|
||
"restrict_cmd": ("HKCU\\Software\\Microsoft\\Windows"
|
||
"\\CurrentVersion\\Policies\\System",
|
||
"DisableCMD"),
|
||
"restrict_wallpaper": ("HKCU\\Software\\Microsoft\\Windows"
|
||
"\\CurrentVersion\\Policies\\System",
|
||
"NoDispBackgroundPage"),
|
||
"restrict_power": ("HKCU\\Software\\Microsoft\\Windows"
|
||
"\\CurrentVersion\\Policies\\Explorer",
|
||
"NoClose"),
|
||
}
|
||
|
||
RESTRICT_LABELS = {
|
||
"hide_icons": "隐藏桌面图标",
|
||
"restrict_taskmgr": "禁用任务管理器",
|
||
"restrict_regedit": "禁用注册表编辑器",
|
||
"restrict_cmd": "禁用命令提示符",
|
||
"restrict_wallpaper": "禁止换壁纸",
|
||
"restrict_power": "藏掉关机按钮",
|
||
"lock_input": "锁键盘鼠标",
|
||
"hosts": "网站屏蔽",
|
||
}
|
||
|
||
def _reg_on(self, path: str, name: str, on_value) -> bool:
|
||
code, out = _run(["reg", "query", path, "/v", name])
|
||
return code == 0 and str(on_value) in out.replace("0x", "")
|
||
|
||
def _reg_set(self, path: str, name: str, value) -> bool:
|
||
code, _out = _run(["reg", "add", path, "/v", name, "/t", "REG_DWORD",
|
||
"/d", str(value), "/f"])
|
||
return code == 0
|
||
|
||
def _do_hide_icons(self, params) -> str:
|
||
return self._toggle_restrict("hide_icons", params)
|
||
|
||
def _do_restrict_taskmgr(self, params) -> str:
|
||
return self._toggle_restrict("restrict_taskmgr", params)
|
||
|
||
def _do_restrict_regedit(self, params) -> str:
|
||
return self._toggle_restrict("restrict_regedit", params)
|
||
|
||
def _do_restrict_cmd(self, params) -> str:
|
||
return self._toggle_restrict("restrict_cmd", params)
|
||
|
||
def _do_restrict_wallpaper(self, params) -> str:
|
||
return self._toggle_restrict("restrict_wallpaper", params)
|
||
|
||
def _do_restrict_power(self, params) -> str:
|
||
return self._toggle_restrict("restrict_power", params)
|
||
|
||
def _toggle_restrict(self, name: str, params) -> str:
|
||
path, value_name = self.RESTRICT_KEYS[name]
|
||
label = self.RESTRICT_LABELS[name]
|
||
on = self._truthy(params.get("on"))
|
||
if not self._reg_set(path, value_name, 1 if on else 0):
|
||
raise ValueError(f"{label}设置失败（可能被策略锁住）")
|
||
self._refresh_explorer()
|
||
return f"{label}：{'已开启' if on else '已关闭'}"
|
||
|
||
def _refresh_explorer(self) -> None:
|
||
try:
|
||
self.actions._mci # noqa: B018 仅确认对象存在
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
HWND_BROADCAST = 0xFFFF
|
||
WM_SETTINGCHANGE = 0x001A
|
||
try:
|
||
ctypes.windll.user32.SendMessageW(HWND_BROADCAST, WM_SETTINGCHANGE,
|
||
0, 0)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def _do_lock_input(self, params) -> str:
|
||
"""锁键盘鼠标。注意：这会连老师自己也动不了，必须靠手表解开。"""
|
||
on = self._truthy(params.get("on"))
|
||
try:
|
||
ok = ctypes.windll.user32.BlockInput(bool(on))
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"锁输入失败（需要管理员）：{exc}") from exc
|
||
if not ok:
|
||
raise ValueError("锁输入失败，需要以管理员身份运行被控端")
|
||
return "键盘鼠标已锁定（用手表解锁）" if on else "键盘鼠标已解锁"
|
||
|
||
# ---------------- 网站屏蔽（hosts）----------------
|
||
|
||
HOSTS = Path(r"C:\Windows\System32\drivers\etc\hosts")
|
||
MARK = "# ==== LanRemote 屏蔽 ===="
|
||
|
||
COMMON_GAMES = [
|
||
"4399.com", "www.4399.com", "7k7k.com", "www.7k7k.com",
|
||
"xiaoyouxi.4399.com", "qq.com", "www.qq.com", "game.qq.com",
|
||
"steamcommunity.com", "store.steampowered.com", "bilibili.com",
|
||
"www.bilibili.com", "douyin.com", "www.douyin.com",
|
||
"huya.com", "www.huya.com", "douyu.com", "www.douyu.com",
|
||
]
|
||
|
||
def _hosts_read(self) -> list:
|
||
try:
|
||
text = self.HOSTS.read_text(encoding="utf-8", errors="replace")
|
||
except OSError as exc:
|
||
raise ValueError(f"读不了 hosts（需要管理员）：{exc}") from exc
|
||
return text.splitlines()
|
||
|
||
def _hosts_blocked(self) -> list:
|
||
sites = []
|
||
try:
|
||
for line in self._hosts_read():
|
||
stripped = line.strip()
|
||
if not stripped.startswith("0.0.0.0"):
|
||
continue
|
||
parts = stripped.split()
|
||
if len(parts) >= 2:
|
||
sites.append(parts[-1])
|
||
except ValueError:
|
||
return []
|
||
return sites
|
||
|
||
def _hosts_write(self, sites: list) -> None:
|
||
lines = [l for l in self._hosts_read()
|
||
if not l.strip().startswith("0.0.0.0")
|
||
and self.MARK not in l]
|
||
if sites:
|
||
lines.append("")
|
||
lines.append(self.MARK)
|
||
for site in sorted(set(sites)):
|
||
lines.append(f"0.0.0.0 {site}")
|
||
lines.append(f"0.0.0.0 www.{site}" if not site.startswith("www.")
|
||
else f"0.0.0.0 {site[4:]}")
|
||
try:
|
||
self.HOSTS.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
except OSError as exc:
|
||
raise ValueError(
|
||
f"写不了 hosts（需要管理员）。以管理员身份运行被控端再试：{exc}"
|
||
) from exc
|
||
_run(["ipconfig", "/flushdns"])
|
||
|
||
def _do_site_block(self, params) -> str:
|
||
site = str(params.get("site") or "").strip().lower()
|
||
site = site.replace("http://", "").replace("https://", "").strip("/")
|
||
if not site or "." not in site:
|
||
raise ValueError("网站格式不对，例如 game.com")
|
||
sites = self._hosts_blocked()
|
||
if site not in sites:
|
||
sites.append(site)
|
||
self._hosts_write(sites)
|
||
return f"已屏蔽 {site}（共 {len(set(sites))} 个）"
|
||
|
||
def _do_site_unblock(self, params) -> str:
|
||
site = str(params.get("site") or "").strip().lower()
|
||
site = site.replace("http://", "").replace("https://", "").strip("/")
|
||
if not site:
|
||
raise ValueError("网站为空")
|
||
sites = [s for s in self._hosts_blocked() if s != site]
|
||
self._hosts_write(sites)
|
||
return f"已解除 {site}（剩 {len(set(sites))} 个）"
|
||
|
||
def _do_site_list(self, params) -> str:
|
||
sites = sorted(set(self._hosts_blocked()))
|
||
if not sites:
|
||
return "当前没有屏蔽任何网站"
|
||
return f"已屏蔽 {len(sites)} 个：\n" + "\n".join(sites[:60])
|
||
|
||
def _do_site_clear(self, params) -> str:
|
||
self._hosts_write([])
|
||
return "已清空全部网站屏蔽"
|
||
|
||
def _do_game_block_common(self, params) -> str:
|
||
sites = set(self._hosts_blocked())
|
||
sites.update(self.COMMON_GAMES)
|
||
self._hosts_write(list(sites))
|
||
return f"已屏蔽 {len(sites)} 个常见游戏/视频网站"
|
||
|
||
# ---------------- 进程 ----------------
|
||
|
||
GAME_WORDS = ["game", "steam", "wegame", "minecraft", "league", "lol",
|
||
"dota", "csgo", "cf", "dnf", "battle.net", "epicgames",
|
||
"原神", "yuanshen", "genshin", "4399", "7k7k", "flash"]
|
||
|
||
def _do_kill_games(self, params) -> str:
|
||
killed = []
|
||
for proc in self.actions.list_processes(limit=200):
|
||
name = (proc.get("name") or "").lower()
|
||
if any(word in name for word in self.GAME_WORDS):
|
||
try:
|
||
self.actions.kill_process(int(proc["pid"]))
|
||
killed.append(proc["name"])
|
||
except Exception: # noqa: BLE001
|
||
continue
|
||
if not killed:
|
||
return "没有发现正在运行的游戏进程"
|
||
return f"已结束 {len(killed)} 个：{', '.join(killed[:10])}"
|
||
|
||
def _do_kill_by_name(self, params) -> str:
|
||
name = str(params.get("name") or "").strip()
|
||
if not name:
|
||
raise ValueError("进程名为空")
|
||
return self.actions.kill_process(0, name)
|
||
|
||
# ---------------- 课堂 ----------------
|
||
|
||
def _do_bell(self, params) -> str:
|
||
try:
|
||
import winsound
|
||
for freq in (880, 660, 990):
|
||
winsound.Beep(freq, 260)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
self._do_big_text({"text": "下课"})
|
||
return "下课铃"
|
||
|
||
def _do_class_attention(self, params) -> str:
|
||
self._do_big_text({"text": "请 安 静"})
|
||
return "已全屏提示"
|
||
|
||
def _do_class_start(self, params) -> str:
|
||
self.actions.media_control("mute")
|
||
self._do_big_text({"text": "上课"})
|
||
return "已开始上课（全屏提示 + 静音）"
|
||
|
||
def _do_class_timer(self, params) -> str:
|
||
return self._do_countdown_screen(params)
|
||
|
||
# ---------------- 恢复 / 状态 ----------------
|
||
|
||
def _do_restore_all(self, params) -> str:
|
||
restored = []
|
||
for name in self.RESTRICT_KEYS:
|
||
path, value_name = self.RESTRICT_KEYS[name]
|
||
if self._reg_set(path, value_name, 0):
|
||
restored.append(self.RESTRICT_LABELS[name])
|
||
try:
|
||
ctypes.windll.user32.BlockInput(False)
|
||
restored.append("锁键盘鼠标")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
if self._hosts_blocked():
|
||
self._hosts_write([])
|
||
restored.append("网站屏蔽")
|
||
except ValueError:
|
||
pass
|
||
self._refresh_explorer()
|
||
if not restored:
|
||
return "没有需要恢复的限制"
|
||
return "已恢复：\n" + "\n".join(restored)
|
||
|
||
def _do_restrict_status(self, params) -> str:
|
||
active = []
|
||
for name, (path, value_name) in self.RESTRICT_KEYS.items():
|
||
if self._reg_on(path, value_name, 1):
|
||
active.append(self.RESTRICT_LABELS[name])
|
||
try:
|
||
if ctypes.windll.user32.GetAsyncKeyState(0) != 0:
|
||
pass
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
sites = self._hosts_blocked()
|
||
if sites:
|
||
active.append(f"网站屏蔽 {len(set(sites))} 个")
|
||
if not active:
|
||
return "当前没有任何限制"
|
||
return f"生效中的限制 {len(active)} 项：\n" + "\n".join(active)
|
||
|
||
# ==================================================================
|
||
# 教学（续）
|
||
# ==================================================================
|
||
|
||
def _roll_names(self) -> list:
|
||
names = self.store.get("roll_call_names") or []
|
||
return [str(n) for n in names if str(n).strip()]
|
||
|
||
def _do_roll_group(self, params) -> str:
|
||
import random
|
||
names = self._roll_names()
|
||
if not names:
|
||
raise ValueError("还没有名单，先点「设置点名名单」")
|
||
try:
|
||
size = max(2, min(20, int(params.get("size") or 4)))
|
||
except (TypeError, ValueError):
|
||
size = 4
|
||
pool = names[:]
|
||
random.shuffle(pool)
|
||
groups = [pool[i:i + size] for i in range(0, len(pool), size)]
|
||
lines = [f"第 {i + 1} 组：" + "、".join(g)
|
||
for i, g in enumerate(groups)]
|
||
self._do_big_text({"text": f"分 {len(groups)} 组"})
|
||
return "\n".join(lines)
|
||
|
||
def _do_roll_call_no_repeat(self, params) -> str:
|
||
import random
|
||
names = self._roll_names()
|
||
if not names:
|
||
raise ValueError("还没有名单，先点「设置点名名单」")
|
||
used = self.store.get("roll_call_used") or []
|
||
left = [n for n in names if n not in used]
|
||
if not left:
|
||
return "全班都点过了（点「重置点名记录」重来）"
|
||
picked = random.choice(left)
|
||
used.append(picked)
|
||
self.store.set("roll_call_used", used)
|
||
self._do_big_text({"text": picked})
|
||
return f"点到：{picked}（还剩 {len(left) - 1} 人没点过）"
|
||
|
||
def _do_roll_call_reset(self, params) -> str:
|
||
self.store.set("roll_call_used", [])
|
||
return "点名记录已重置"
|
||
|
||
def _do_class_schedule(self, params) -> str:
|
||
raw = str(params.get("text") or "").strip()
|
||
if not raw:
|
||
self.store.set("class_schedule", "")
|
||
return "课程表已清空"
|
||
for sep in ("，", "、", ";", "；"):
|
||
raw = raw.replace(sep, ",")
|
||
items = [x.strip() for x in raw.split(",") if x.strip()]
|
||
self.store.set("class_schedule", items)
|
||
self._do_big_text({"text": "\n".join(
|
||
f"{i + 1}. {x}" for i, x in enumerate(items))})
|
||
return f"课程表已保存 {len(items)} 节"
|
||
|
||
def _do_hand_count(self, params) -> str:
|
||
count = int(self.store.get("hand_count") or 0) + 1
|
||
self.store.set("hand_count", count)
|
||
self._do_big_text({"text": str(count)})
|
||
return f"举手计数：{count}"
|
||
|
||
def _do_hand_reset(self, params) -> str:
|
||
self.store.set("hand_count", 0)
|
||
return "举手计数已清零"
|
||
|
||
def _do_answer_poll(self, params) -> str:
|
||
counts = self.store.get("answer_poll") or {}
|
||
if not isinstance(counts, dict):
|
||
counts = {}
|
||
lines = [f"{k}：{v} 人" for k, v in sorted(counts.items())] or ["还没记录"]
|
||
lines.append("")
|
||
lines.append("（在电脑上点画面对应的选项还可以继续加）")
|
||
return "\n".join(lines)
|
||
|
||
def _do_magnifier(self, params) -> str:
|
||
subprocess.Popen("magnify.exe", shell=True, creationflags=_NO_WINDOW)
|
||
return "已打开屏幕放大镜"
|
||
|
||
def _do_projector_switch(self, params) -> str:
|
||
# Win + P
|
||
self._key_combo(0x5B, 0x50)
|
||
return "已弹出投影模式选择（Win+P）"
|
||
|
||
def _do_lesson_timer(self, params) -> str:
|
||
started = self.store.get("lesson_started")
|
||
if started:
|
||
import time as _t
|
||
mins = int((_t.time() - float(started)) / 60)
|
||
self.store.set("lesson_started", 0)
|
||
return f"本节课上了 {mins} 分钟"
|
||
import time as _t
|
||
self.store.set("lesson_started", _t.time())
|
||
return "开始计时（再点一次看上了多久）"
|
||
|
||
# ==================================================================
|
||
# 教室（续）
|
||
# ==================================================================
|
||
|
||
def _do_lock_now(self, params) -> str:
|
||
ctypes.windll.user32.LockWorkStation()
|
||
return "已锁屏"
|
||
|
||
def _game_watch_loop(self) -> None:
|
||
import time as _t
|
||
while self.store.get("game_watch"):
|
||
try:
|
||
for proc in self.actions.list_processes(limit=200):
|
||
name = (proc.get("name") or "").lower()
|
||
if any(w in name for w in self.GAME_WORDS):
|
||
self._do_big_text({"text": f"别玩游戏\n{proc['name']}"})
|
||
break
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
_t.sleep(15)
|
||
|
||
def _do_game_watch_start(self, params) -> str:
|
||
if self.store.get("game_watch"):
|
||
return "已经在盯着了"
|
||
self.store.set("game_watch", True)
|
||
import threading
|
||
threading.Thread(target=self._game_watch_loop, daemon=True,
|
||
name="game-watch").start()
|
||
return "开始盯游戏进程（每 15 秒查一次，发现就全屏提醒）"
|
||
|
||
def _do_game_watch_stop(self, params) -> str:
|
||
self.store.set("game_watch", False)
|
||
return "已停止盯游戏进程"
|
||
|
||
def _do_usb_check(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_DiskDrive | Where-Object "
|
||
"{$_.InterfaceType -eq 'USB'} | ForEach-Object { "
|
||
"\"$($_.Model) $([math]::Round($_.Size/1GB,1))GB\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return ("发现 U 盘：" + " | ".join(lines)) if lines else "没有插 U 盘"
|
||
|
||
def _do_login_history(self, params) -> str:
|
||
script = ("Get-WinEvent -LogName Security -MaxEvents 200 "
|
||
"-FilterXPath '*[System[EventID=4624]]' "
|
||
"-ErrorAction SilentlyContinue | Select-Object -First 10 | "
|
||
"ForEach-Object { $_.TimeCreated.ToString('MM-dd HH:mm') }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
if not lines:
|
||
return "读不到登录记录（需要管理员权限）"
|
||
return "最近登录：\n" + "\n".join(lines)
|
||
|
||
def _do_boot_history(self, params) -> str:
|
||
script = ("Get-WinEvent -LogName System -MaxEvents 500 "
|
||
"-ErrorAction SilentlyContinue | "
|
||
"Where-Object {$_.Id -eq 6005 -or $_.Id -eq 6006} | "
|
||
"Select-Object -First 8 | "
|
||
"ForEach-Object { $_.TimeCreated.ToString('MM-dd HH:mm') }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
if not lines:
|
||
return f"读不到开关机记录；本次已开机 {self.actions._uptime()}"
|
||
return "最近开关机：\n" + "\n".join(lines)
|
||
|
||
def _do_clean_desktop(self, params) -> str:
|
||
desktop = Path.home() / "Desktop"
|
||
if not desktop.is_dir():
|
||
raise ValueError("找不到桌面目录")
|
||
box = desktop / "_桌面整理"
|
||
box.mkdir(exist_ok=True)
|
||
moved = 0
|
||
for item in desktop.iterdir():
|
||
if item == box or item.is_dir():
|
||
continue
|
||
try:
|
||
item.rename(box / item.name)
|
||
moved += 1
|
||
except OSError:
|
||
continue
|
||
return f"把 {moved} 个文件收进了「_桌面整理」文件夹"
|
||
|
||
def _do_block_new_app(self, params) -> str:
|
||
on = self._truthy(params.get("on"))
|
||
# 用软件限制策略（HKCU 下不需要管理员）
|
||
path = r"Software\Microsoft\Windows\CurrentVersion\Policies\Explorer"
|
||
ok = self._reg_set(f"HKCU\\{path}",
|
||
"DisallowRun", 1 if on else 0)
|
||
if not ok:
|
||
raise ValueError("设置失败，可能被组策略锁住")
|
||
return f"禁止新程序启动：{'已开启' if on else '已关闭'}"
|
||
|
||
def _do_show_desktop_now(self, params) -> str:
|
||
self._key_combo(0x5B, 0x44)
|
||
return "已显示桌面"
|
||
|
||
def _do_count_windows(self, params) -> str:
|
||
user32 = ctypes.windll.user32
|
||
EnumProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p,
|
||
ctypes.c_void_p)
|
||
own = os.getpid()
|
||
titles = []
|
||
|
||
def callback(hwnd, _lparam):
|
||
if not user32.IsWindowVisible(hwnd):
|
||
return True
|
||
length = user32.GetWindowTextLengthW(hwnd)
|
||
if length == 0:
|
||
return True
|
||
pid = ctypes.c_ulong()
|
||
user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
||
if pid.value == own:
|
||
return True
|
||
buf = ctypes.create_unicode_buffer(length + 1)
|
||
user32.GetWindowTextW(hwnd, buf, length + 1)
|
||
titles.append(buf.value)
|
||
return True
|
||
|
||
user32.EnumWindows(EnumProc(callback), 0)
|
||
if not titles:
|
||
return "没有打开的窗口"
|
||
return f"开着 {len(titles)} 个窗口：\n" + "\n".join(titles[:15])
|
||
|
||
def _do_class_end_cleanup(self, params) -> str:
|
||
parts = []
|
||
try:
|
||
parts.append(self._do_kill_games({}))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
self.store.set("game_watch", False)
|
||
parts.append("已停止盯进程")
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
try:
|
||
parts.append(self._do_restore_all({}))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return "下课清理完成：\n" + "\n".join(parts)
|
||
|
||
def _do_who_is_playing(self, params) -> str:
|
||
found = []
|
||
for proc in self.actions.list_processes(limit=200):
|
||
name = (proc.get("name") or "").lower()
|
||
if any(w in name for w in self.GAME_WORDS):
|
||
found.append(f"{proc['name']} (PID {proc['pid']})")
|
||
if not found:
|
||
return "没有发现游戏进程"
|
||
return f"发现 {len(found)} 个：\n" + "\n".join(found[:15])
|
||
|
||
# ==================================================================
|
||
# 系统（续）
|
||
# ==================================================================
|
||
|
||
def _do_open_taskmgr2(self, params) -> str:
|
||
return self._open_msc("taskmgr.exe", "任务管理器")
|
||
|
||
def _do_open_resmon(self, params) -> str:
|
||
return self._open_msc("resmon.exe", "资源监视器")
|
||
|
||
def _do_power_plan(self, params) -> str:
|
||
_code, out = _run(["powercfg", "/list"])
|
||
plans = re.findall(r"([0-9a-fA-F-]{36})\s*\(([^)]+)\)", out or "")
|
||
if not plans:
|
||
return "读不到电源计划"
|
||
lines = [f"{name}" for _guid, name in plans]
|
||
return "可用的电源计划：\n" + "\n".join(lines)
|
||
|
||
def _do_window_top(self, params) -> str:
|
||
"""把当前前台窗口切换置顶。"""
|
||
user32 = ctypes.windll.user32
|
||
hwnd = user32.GetForegroundWindow()
|
||
if not hwnd:
|
||
raise ValueError("没有前台窗口")
|
||
HWND_TOPMOST = -1
|
||
HWND_NOTOPMOST = -2
|
||
SWP_NOMOVE = 0x0002
|
||
SWP_NOSIZE = 0x0001
|
||
# 先试着取消置顶，成功就说明本来是置顶的
|
||
user32.SetWindowPos(hwnd, HWND_NOTOPMOST, 0, 0, 0, 0,
|
||
SWP_NOMOVE | SWP_NOSIZE)
|
||
user32.SetWindowPos(hwnd, HWND_TOPMOST, 0, 0, 0, 0,
|
||
SWP_NOMOVE | SWP_NOSIZE)
|
||
return "已把当前窗口置顶"
|
||
|
||
def _do_screen_off_lock(self, params) -> str:
|
||
ctypes.windll.user32.LockWorkStation()
|
||
import time as _t
|
||
_t.sleep(0.4)
|
||
self.actions.monitor_power(False)
|
||
return "已关屏并锁定"
|
||
|
||
def _do_clear_all_clip(self, params) -> str:
|
||
self._clip_history = []
|
||
self.store.set("feat_clip_fav", [])
|
||
return "剪贴板历史和收藏夹已清空"
|
||
|
||
def _do_restart_audio(self, params) -> str:
|
||
_run(["net", "stop", "Audiosrv", "/y"])
|
||
_run(["net", "start", "Audiosrv"])
|
||
return "音频服务已重启（可能需要管理员权限）"
|
||
|
||
def _do_system_sounds(self, params) -> str:
|
||
subprocess.Popen(["cmd", "/c", "start", "ms-settings:sound"],
|
||
creationflags=_NO_WINDOW)
|
||
return "已打开声音设置"
|
||
|
||
# ==================================================================
|
||
# 查看（续）
|
||
# ==================================================================
|
||
|
||
def _do_cpu_model(self, params) -> str:
|
||
script = ("$c=Get-CimInstance Win32_Processor|Select-Object -First 1;"
|
||
"\"$($c.Name)\";\"$($c.NumberOfCores) 核 "
|
||
"$($c.NumberOfLogicalProcessors) 线程\";"
|
||
"\"主频 $($c.MaxClockSpeed) MHz\"")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到 CPU 信息"
|
||
|
||
def _do_memory_sticks(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_PhysicalMemory | ForEach-Object { "
|
||
"\"$([math]::Round($_.Capacity/1GB,0))GB "
|
||
"$($_.Speed)MHz $($_.Manufacturer)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到内存条信息"
|
||
|
||
def _do_disk_serial(self, params) -> str:
|
||
script = ("Get-CimInstance Win32_DiskDrive | ForEach-Object { "
|
||
"\"$($_.Model) SN=$($_.SerialNumber)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到硬盘序列号"
|
||
|
||
def _do_gpu_temp(self, params) -> str:
|
||
script = ("$t=Get-CimInstance -Namespace root/WMI -ClassName "
|
||
"MSAcpi_ThermalZoneTemperature -ErrorAction SilentlyContinue | "
|
||
"Select-Object -First 1;"
|
||
"if($t){[math]::Round(($t.CurrentTemperature/10)-273.15,1)}"
|
||
"else{'n/a'}")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
temp = (out or "").strip().splitlines()
|
||
gpu = self._do_gpu_info({})
|
||
return f"温度：{temp[-1] if temp else 'n/a'}°C\n{gpu}"
|
||
|
||
def _do_net_total(self, params) -> str:
|
||
script = ("$s=Get-NetAdapterStatistics | Measure-Object -Property "
|
||
"ReceivedBytes,SentBytes -Sum;"
|
||
"\"收 $([math]::Round($s[0].Sum/1GB,2)) GB 发 "
|
||
"$([math]::Round($s[1].Sum/1GB,2)) GB\"")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
return (out or "").strip().splitlines()[-1] if out.strip() else "读不到"
|
||
|
||
def _do_running_games(self, params) -> str:
|
||
return self._do_who_is_playing(params)
|
||
|
||
def _do_activation(self, params) -> str:
|
||
script = ("$a = Get-CimInstance SoftwareLicensingProduct "
|
||
"-ErrorAction SilentlyContinue | "
|
||
"Where-Object {$_.PartialProductKey} | "
|
||
"Select-Object -First 1; "
|
||
"if ($a) { $a.Name; 'LicenseStatus=' + $a.LicenseStatus } "
|
||
"else { 'none' }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=30)
|
||
lines = [l.strip() for l in out.splitlines() if l.strip()]
|
||
if not lines or lines[-1] == "none":
|
||
return "读不到激活信息"
|
||
text = "\n".join(lines)
|
||
if "LicenseStatus=1" in text:
|
||
text += "\n（LicenseStatus=1 表示已激活）"
|
||
return text
|
||
|
||
def _do_disk_health(self, params) -> str:
|
||
script = ("Get-PhysicalDisk | ForEach-Object { "
|
||
"\"$($_.FriendlyName) 健康=$($_.HealthStatus) "
|
||
"状态=$($_.OperationalStatus)\" }")
|
||
_code, out = _run(["powershell", "-NoProfile", "-Command", script],
|
||
timeout=25)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines) or "读不到硬盘健康信息"
|
||
|
||
# ==================================================================
|
||
# 效率（续）
|
||
# ==================================================================
|
||
|
||
def _do_sticky_note(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
notes = self._load("sticky", dict)
|
||
notes.insert(0, {"text": text[:500],
|
||
"at": time.strftime("%m-%d %H:%M")})
|
||
self._persist("sticky", notes[:100])
|
||
self._do_big_text({"text": text[:40]})
|
||
return f"便签已贴（共 {len(notes[:100])} 条）"
|
||
|
||
def _do_search_note(self, params) -> str:
|
||
word = str(params.get("word") or "").strip().lower()
|
||
if not word:
|
||
raise ValueError("关键词为空")
|
||
hits = []
|
||
for key, label in (("notes", "笔记"), ("sticky", "便签"),
|
||
("todos", "待办")):
|
||
for item in self._load(key, dict):
|
||
if word in str(item.get("text", "")).lower():
|
||
hits.append(f"[{label}] {item.get('text', '')[:60]}")
|
||
if not hits:
|
||
return f"没有含「{word}」的记录"
|
||
return f"找到 {len(hits)} 条：\n" + "\n".join(hits[:30])
|
||
|
||
def _do_translator(self, params) -> str:
|
||
text = str(params.get("text") or "").strip()
|
||
if not text:
|
||
raise ValueError("内容为空")
|
||
from urllib.parse import quote
|
||
url = f"https://translate.google.com/?text={quote(text)}"
|
||
subprocess.Popen(["cmd", "/c", "start", url], creationflags=_NO_WINDOW)
|
||
return f"已在浏览器打开「{text[:20]}」的翻译"
|
||
|
||
def _do_calc_open(self, params) -> str:
|
||
subprocess.Popen("calc.exe", shell=True, creationflags=_NO_WINDOW)
|
||
return "已打开计算器"
|
||
|
||
def _do_unit_convert(self, params) -> str:
|
||
text = str(params.get("text") or "").strip().lower().replace(" ", "")
|
||
match = re.match(r"^(-?\d+(?:\.\d+)?)([a-z厘米寸尺斤公斤克磅度]*)$", text)
|
||
if not match:
|
||
raise ValueError("格式如 100cm / 5kg / 30度")
|
||
value = float(match.group(1))
|
||
unit = match.group(2)
|
||
table = {
|
||
"cm": ("英寸", value / 2.54), "厘米": ("英寸", value / 2.54),
|
||
"m": ("英尺", value * 3.2808), "米": ("英尺", value * 3.2808),
|
||
"km": ("英里", value * 0.6214), "公里": ("英里", value * 0.6214),
|
||
"kg": ("磅", value * 2.2046), "公斤": ("磅", value * 2.2046),
|
||
"克": ("盎司", value * 0.0353),
|
||
"斤": ("公斤", value * 0.5),
|
||
"度": ("华氏度", value * 9 / 5 + 32),
|
||
"c": ("华氏度", value * 9 / 5 + 32),
|
||
}
|
||
item = table.get(unit)
|
||
if item is None:
|
||
raise ValueError(f"不认识的单位：{unit}")
|
||
return f"{value}{unit} = {item[1]:.2f}{item[0]}"
|
||
|
||
def _do_countdown_day(self, params) -> str:
|
||
date_text = str(params.get("date") or "").strip()
|
||
match = re.match(r"^(\d{1,2})[-/月](\d{1,2})", date_text)
|
||
if not match:
|
||
raise ValueError("格式如 06-20")
|
||
import datetime as _dt
|
||
now = _dt.date.today()
|
||
try:
|
||
target = _dt.date(now.year, int(match.group(1)), int(match.group(2)))
|
||
except ValueError as exc:
|
||
raise ValueError(f"日期不对：{exc}") from exc
|
||
if target < now:
|
||
target = _dt.date(now.year + 1, target.month, target.day)
|
||
days = (target - now).days
|
||
self._do_big_text({"text": f"还有 {days} 天"})
|
||
return f"距离 {target} 还有 {days} 天"
|
||
|
||
# ==================================================================
|
||
# 媒体（续）
|
||
# ==================================================================
|
||
|
||
def _do_loop_play(self, params) -> str:
|
||
on = self._truthy(params.get("on"))
|
||
self.store.set("loop_play", bool(on))
|
||
# 真正的循环由 Actions 的看门线程负责，这里只转达开关。
|
||
# 之前直接用了 self._lock / self._audio_alias / self._mci ——
|
||
# 那些是 Actions 的属性，Features 上没有，一点就崩。
|
||
setter = getattr(self.actions, "set_loop", None)
|
||
if setter is not None:
|
||
try:
|
||
setter(bool(on))
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
return f"循环播放：{'已开启' if on else '已关闭'}"
|
||
|
||
def _do_shuffle_play(self, params) -> str:
|
||
on = self._truthy(params.get("on"))
|
||
self.store.set("shuffle_play", bool(on))
|
||
return f"随机播放：{'已开启' if on else '已关闭'}"
|
||
|
||
def _do_stop_all_sound(self, params) -> str:
|
||
self.actions.stop_audio()
|
||
self._volume_key(0xAD, 1)
|
||
return "已停掉被控端播放的音频，并切换系统静音"
|
||
|
||
# ==================================================================
|
||
# 网络（续）
|
||
# ==================================================================
|
||
|
||
def _do_relay_saver(self, params) -> str:
|
||
"""跨网络省流量：把屏幕流降一档。
|
||
|
||
实测：经隧道上行只有 ~37 KB/s，而 480p/55 的一帧约 350KB，
|
||
传一帧要 9 秒 —— 屏幕镜像等于不能用。
|
||
降到 360p/40 之后一帧约 60KB，快 5 倍以上。
|
||
|
||
关掉时会**还原成开之前的设置**，所以先把原值存下来。
|
||
"""
|
||
on = self._truthy(params.get("on"))
|
||
if on:
|
||
saved = self.store.get("relay_saver_saved")
|
||
if not isinstance(saved, dict) or not saved:
|
||
# 第一次开：记下原值，方便关的时候还原
|
||
saved = {
|
||
"width": self.store.get("screen_max_width"),
|
||
"quality": self.store.get("screen_quality"),
|
||
}
|
||
self.store.set("relay_saver_saved", saved)
|
||
self.store.set("relay_saver", True)
|
||
self.store.set("screen_max_width", 360)
|
||
self.store.set("screen_quality", 40)
|
||
note = (f"已开启省流量：屏幕 360p / 画质 40"
|
||
f"（原来 {(saved or {}).get('width') or 480}p / "
|
||
f"{(saved or {}).get('quality') or 55}）")
|
||
else:
|
||
saved = self.store.get("relay_saver_saved")
|
||
self.store.set("relay_saver", False)
|
||
if isinstance(saved, dict) and saved:
|
||
if saved.get("width"):
|
||
self.store.set("screen_max_width", saved["width"])
|
||
if saved.get("quality"):
|
||
self.store.set("screen_quality", saved["quality"])
|
||
note = (f"已关闭省流量，还原为 {saved.get('width')}p / "
|
||
f"画质 {saved.get('quality')}")
|
||
self.store.set("relay_saver_saved", {})
|
||
else:
|
||
note = "已关闭省流量（没有记录到原设置，保持当前值）"
|
||
|
||
# 采集器那边会自己检测参数变化并重建（见 server._ensure_capturer），
|
||
# 这里不用额外做什么。
|
||
return note
|
||
|
||
def _do_net_traffic(self, params) -> str:
|
||
return self._do_network_speed(params)
|
||
|
||
def _do_traceroute(self, params) -> str:
|
||
host = str(params.get("host") or "").strip()
|
||
if not host:
|
||
raise ValueError("地址为空")
|
||
if not re.match(r"^[A-Za-z0-9_.:\-]+$", host):
|
||
raise ValueError("地址含有不合法字符")
|
||
_code, out = _run(["tracert", "-d", "-h", "12", host], timeout=60)
|
||
lines = [l for l in out.splitlines() if l.strip()]
|
||
return "\n".join(lines[-14:]) or "没有输出"
|
||
|
||
def _do_net_reset(self, params) -> str:
|
||
_run(["ipconfig", "/flushdns"])
|
||
_run(["netsh", "winsock", "reset"])
|
||
_run(["netsh", "int", "ip", "reset"])
|
||
return "网络已重置（部分项目需要重启才生效）"
|
||
|
||
# ---------------- 持久化 ----------------
|
||
|
||
def _persist(self, key: str, value) -> None:
|
||
try:
|
||
self.store.set(f"feat_{key}", value)
|
||
except Exception: # noqa: BLE001
|
||
pass
|
||
|
||
def _load(self, key: str, item_type=None):
|
||
"""读持久化列表。
|
||
|
||
设置文件可能被手改坏、或是旧版本残留，所以元素类型也要校验 ——
|
||
否则 notes/todos/clip_fav 的列表渲染会抛 TypeError 崩掉。
|
||
"""
|
||
try:
|
||
value = self.store.get(f"feat_{key}")
|
||
except Exception: # noqa: BLE001
|
||
return []
|
||
if not isinstance(value, list):
|
||
return []
|
||
if item_type is None:
|
||
return value
|
||
return [item for item in value if isinstance(item, item_type)]
|