ykxmz/windows/lanremote/actions.py

# -*- coding: utf-8 -*-
"""被控端功能实现：投图 / 放音 / 媒体控制 / 剪贴板 / 电源 / 自定义命令 / TTS / 通知。

所有会碰 UI 的操作都通过 ui_call 调度回 tkinter 主线程执行。
"""
from __future__ import annotations

import ctypes
import os
import subprocess
import sys
import threading
import time
from pathlib import Path

from .protocol import MEDIA_ACTIONS, POWER_ACTIONS

IS_WINDOWS = sys.platform.startswith("win")

# ---- Win32 常量 ----
KEYEVENTF_KEYUP = 0x0002
VK_MEDIA = MEDIA_ACTIONS


def _human_size(num: float) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if abs(num) < 1024.0:
            return f"{num:.1f}{unit}"
        num /= 1024.0
    return f"{num:.1f}TB"


class ActionError(RuntimeError):
    pass


def work_area() -> tuple:
    """屏幕工作区 (left, top, width, height) —— 已经扣掉任务栏。

    直接读 SPI_GETWORKAREA，比自己猜任务栏多高靠谱。
    """
    try:
        import ctypes.wintypes
        rect = ctypes.wintypes.RECT()
        ok = ctypes.windll.user32.SystemParametersInfoW(
            0x0030, 0, ctypes.byref(rect), 0)             # SPI_GETWORKAREA
        if ok and rect.right > rect.left and rect.bottom > rect.top:
            return (rect.left, rect.top,
                    rect.right - rect.left, rect.bottom - rect.top)
    except Exception:                                     # noqa: BLE001
        pass
    try:
        import tkinter as tk
        root = tk._default_root                                 # noqa: SLF001
        if root is not None:
            return (0, 0, root.winfo_screenwidth(),
                    root.winfo_screenheight())
    except Exception:                                     # noqa: BLE001
        pass
    return (0, 0, 1920, 1080)


def _outer_size(win) -> tuple:
    """窗口的真实外框尺寸（含标题栏和边框）。

    客户区尺寸不等于窗口占的地方 —— 不按外框算，看起来就不居中。
    """
    try:
        import ctypes
        import ctypes.wintypes
        hwnd = int(win.frame(), 0)                        # '0x00012345'
        rect = ctypes.wintypes.RECT()
        if ctypes.windll.user32.GetWindowRect(hwnd, ctypes.byref(rect)):
            return (rect.right - rect.left, rect.bottom - rect.top,
                    rect.left, rect.top)
    except Exception:                                     # noqa: BLE001
        pass
    return (win.winfo_width(), win.winfo_height(), 0, 0)


def _place_centered(win, client_w: int, client_h: int) -> None:
    """在工作区正中间摆好这个窗口（按外框算）。"""
    client_w = max(1, int(client_w))
    client_h = max(1, int(client_h))
    # 先放到 0,0，好量出真实外框
    win.geometry(f"{client_w}x{client_h}+0+0")
    win.update_idletasks()

    outer_w, outer_h, outer_x, outer_y = _outer_size(win)
    # 外框相对客户区多出来的部分（左边框 / 标题栏+上边框）
    chrome_w = max(0, outer_w - client_w)
    chrome_h = max(0, outer_h - client_h)

    area_x, area_y, area_w, area_h = work_area()
    x = area_x + max(0, (area_w - outer_w) // 2) + chrome_w // 2
    y = area_y + max(0, (area_h - outer_h) // 2) + chrome_h - chrome_h // 2

    win.geometry(f"{client_w}x{client_h}+{x}+{y}")
    win.update_idletasks()
    win.lift()
    win.focus_force()


class Actions:
    """把协议命令翻译成 Windows 上的实际操作。"""

    def __init__(self, ui, store):
        self.ui = ui                    # 需要提供 .root / .log() / .notify()
        self.store = store
        self._audio_alias: str | None = None
        self._audio_path: str = ""          # 上一首的路径，循环播放要用
        self._loop: bool = False             # 单曲循环开关
        self._loop_thread = None             # 循环看门线程
        self._image_window = None
        self._lock = threading.RLock()
        # 剪贴板历史（最近 20 条）
        self._clip_history: list[dict] = []
        # CPU 占用要两次采样才算得出来
        self._last_cpu_total = 0
        self._last_cpu_idle = 0
        # 电脑正在播的音频是否被暂停（MCI 不是系统媒体会话，得自己记）
        self._audio_paused = False
        self._audio_name = ""

    # ------------------------------------------------------------------
    # 投图：在电脑上全屏显示
    # ------------------------------------------------------------------

    def show_image(self, path: str, title: str = "") -> str:
        """显示图片。

        title 为窗口标题（可留空，用设置里的模板；模板支持 {name}）。
        方式由设置 image_display 决定：
          window      —— 屏幕正中间的普通 Windows 窗口（默认）
          fullscreen  —— 全屏
          custom      —— 按 image_custom_width/height 指定大小
        宽或高填 0 表示该方向自动。
        """
        p = Path(path)
        if not p.is_file():
            raise ActionError(f"图片不存在: {path}")

        from PIL import Image, ImageTk

        try:
            img = Image.open(p)
            img.load()
        except Exception as exc:
            raise ActionError(f"无法读取图片: {exc}") from exc

        mode = str(self.store.get("image_display", "window") or "window")
        if mode not in ("window", "fullscreen", "custom"):
            mode = "window"

        screen_w, screen_h = self.ui.screen_size()
        img = img.convert("RGB")

        if mode == "fullscreen":
            target = (screen_w, screen_h)
        elif mode == "custom":
            cw = int(self.store.get("image_custom_width", 0) or 0)
            ch = int(self.store.get("image_custom_height", 0) or 0)
            if cw <= 0 and ch <= 0:
                # 两个都没填就退回默认窗口大小
                cw, ch = int(screen_w * 0.78), int(screen_h * 0.78)
                self.store.set("image_custom_width", cw)
                self.store.set("image_custom_height", ch)
            elif cw <= 0:
                cw = screen_w
            elif ch <= 0:
                ch = screen_h
            target = (max(80, cw), max(80, ch))
        else:
            target = (int(screen_w * 0.78), int(screen_h * 0.78))

        img.thumbnail(target, Image.LANCZOS)
        fullscreen = (mode == "fullscreen")

        # 窗口标题：单次指定 > 设置里的模板 > 默认
        window_title = (title or "").strip() or str(
            self.store.get("image_window_title") or "").strip() or "图片 - {name}"
        try:
            window_title = window_title.replace("{name}", p.name)
        except Exception:
            window_title = f"图片 - {p.name}"
        backdrop = "#000000" if fullscreen else "#ffffff"

        def build():
            import tkinter as tk
            if self._image_window is not None:
                try:
                    self._image_window.destroy()
                except Exception:
                    pass

            win = tk.Toplevel(self.ui.root)
            win.title(window_title)
            win.configure(bg=backdrop)
            win.resizable(False, False)

            def close(_event=None):
                with self._lock:
                    self._image_window = None
                try:
                    win.destroy()
                except Exception:
                    pass

            photo = ImageTk.PhotoImage(img)
            label = tk.Label(win, image=photo, bg=backdrop, bd=0,
                             highlightthickness=0)
            label.image = photo                      # 防止被 GC
            label.pack(fill="both", expand=True)

            if fullscreen:
                # 注意：Windows 上先 overrideredirect(True) 再设 -fullscreen 会抛
                # "override-redirect flag is set"。全屏本身就会去掉边框，不需要它。
                win.attributes("-fullscreen", True)
                win.attributes("-topmost", True)
                tk.Label(win, text="按 Esc 或点击关闭", bg=backdrop, fg="#888888",
                         font=("Microsoft YaHei UI", 10)).place(
                    relx=0.5, rely=0.97, anchor="center")
            else:
                foot = tk.Frame(win, bg=backdrop)
                foot.pack(fill="x")
                tk.Button(foot, text="关闭", command=close,
                          bg="#eaeef2", fg="#1f2328", relief="flat",
                          font=("Microsoft YaHei UI", 9), padx=14, pady=2,
                          cursor="hand2").pack(side="right", padx=10, pady=8)
                # 摆到屏幕正中间 —— 普通窗口，有标题栏和关闭按钮
                # 必须先 update_idletasks，否则 winfo_reqwidth 还是 Tk
                # 默认的 200x200，自定义尺寸就白填了。
                win.update_idletasks()
                _place_centered(win, win.winfo_reqwidth(),
                                win.winfo_reqheight())

            win.bind("<Escape>", close)
            win.bind("<Button-1>", close)
            win.protocol("WM_DELETE_WINDOW", close)
            self._image_window = win

            timeout = int(self.store.get("image_timeout", 0) or 0)
            if timeout > 0:
                win.after(timeout * 1000, close)

        self.ui.ui_call(build)
        label_text = {"window": "屏幕中间", "fullscreen": "全屏",
                      "custom": f"{img.width}x{img.height}"}[mode]
        return f"已显示 {p.name}（{label_text}）"

    def close_image(self) -> str:
        def close():
            if self._image_window is not None:
                try:
                    self._image_window.destroy()
                except Exception:
                    pass
                self._image_window = None
        self.ui.ui_call(close)
        return "已关闭显示窗口"

    # ------------------------------------------------------------------
    # 放音：MCI（系统自带，无需外部库）
    # ------------------------------------------------------------------

    def play_audio(self, path: str) -> str:
        p = Path(path)
        if not p.is_file():
            raise ActionError(f"音频不存在: {path}")
        if not IS_WINDOWS:
            raise ActionError("音频播放目前仅支持 Windows")

        with self._lock:
            self._mci_close_locked()
            alias = f"lanremote{int(time.time() * 1000) % 1000000}"
            opened = self._mci(f'open "{p}" type mpegvideo alias {alias}')
            if not opened:
                opened = self._mci(f'open "{p}" alias {alias}')
            if not opened:
                raise ActionError(f"系统无法播放该音频格式: {p.suffix}")
            self._audio_alias = alias
            self._audio_paused = False
            self._audio_name = p.name
            self._audio_path = str(p)
            self._mci(f"play {alias}")
        return f"正在播放 {p.name}"

    def set_loop(self, on: bool) -> None:
        """开关单曲循环。开了就起一个看门线程，放完自动重放。"""
        self._loop = bool(on)
        if on and self._loop_thread is None:
            thread = threading.Thread(target=self._loop_watch, daemon=True,
                                      name="audio-loop")
            self._loop_thread = thread
            thread.start()

    def _loop_watch(self) -> None:
        while self._loop:
            time.sleep(0.6)
            if not self._loop:
                break
            with self._lock:
                alias = self._audio_alias
                path = self._audio_path
            if not alias or not path:
                continue
            if self._mci_mode(alias) != "stopped":
                continue
            # 放完了，再放一遍
            try:
                self.play_audio(path)
            except Exception:                             # noqa: BLE001
                pass
        self._loop_thread = None

    def stop_audio(self) -> str:
        with self._lock:
            self._mci_close_locked()
            self._audio_paused = False
            self._audio_name = ""
        return "已停止播放"

    def _mci_mode(self, alias: str) -> str:
        """问 MCI 现在是什么状态：playing / paused / stopped。"""
        buffer = ctypes.create_unicode_buffer(64)
        try:
            ctypes.windll.winmm.mciSendStringW(
                ctypes.c_wchar_p(f"status {alias} mode"), buffer, 64, None)
            return (buffer.value or "").strip().lower()
        except Exception:                                 # noqa: BLE001
            return ""

    def audio_pause(self) -> str:
        with self._lock:
            alias = self._audio_alias
            if not alias:
                raise ActionError("电脑当前没有播放手表推过去的音频")
            if self._audio_paused:
                return "已经暂停了"
            if not self._mci(f"pause {alias}"):
                raise ActionError("暂停失败")
            self._audio_paused = True
        return f"已暂停 {self._audio_name or '音频'}"

    def audio_resume(self) -> str:
        with self._lock:
            alias = self._audio_alias
            if not alias:
                raise ActionError("电脑当前没有播放手表推过去的音频")
            if not self._audio_paused:
                return "正在播放中"
            if not self._mci(f"resume {alias}"):
                raise ActionError("继续播放失败")
            self._audio_paused = False
        return f"继续播放 {self._audio_name or '音频'}"

    def audio_status(self) -> str:
        with self._lock:
            alias = self._audio_alias
            if not alias:
                return "电脑当前没有播放音频"
            mode = self._mci_mode(alias)
            if mode == "playing":
                return f"正在播放 {self._audio_name}"
            if mode == "paused":
                return f"已暂停 {self._audio_name}"
            # MCI 自己放完了
            self._mci_close_locked()
            self._audio_paused = False
            name = self._audio_name
            self._audio_name = ""
            return f"{name} 已播放完毕" if name else "播放已结束"

    def _mci_close_locked(self) -> None:
        if self._audio_alias:
            self._mci(f"close {self._audio_alias}")
            self._audio_alias = None

    @staticmethod
    def _mci(command: str) -> bool:
        try:
            winmm = ctypes.windll.winmm
            err = winmm.mciSendStringW(ctypes.c_wchar_p(command), None, 0, None)
            return err == 0
        except Exception:
            return False

    # ------------------------------------------------------------------
    # 媒体控制：模拟多媒体按键，作用于当前播放器
    # ------------------------------------------------------------------

    def media_control(self, action: str) -> str:
        if not IS_WINDOWS:
            raise ActionError("媒体控制目前仅支持 Windows")
        if action not in VK_MEDIA:
            raise ActionError(f"不支持的媒体动作: {action}")

        # 电脑正在播手表推过去的音频时，直接控制它。
        # MCI 不是系统媒体会话，全局媒体键对它无效 ——
        # 这就是"推了 mp3 过去却暂停不了"的原因。
        if action in ("playpause", "stop"):
            handled = self._control_pushed_audio(action)
            if handled:
                return handled

        vk = VK_MEDIA.get(action)
        try:
            user32 = ctypes.windll.user32
            user32.keybd_event(vk, 0, 0, 0)
            user32.keybd_event(vk, 0, KEYEVENTF_KEYUP, 0)
        except Exception as exc:
            raise ActionError(f"媒体按键注入失败: {exc}") from exc
        return f"已执行 {action}"

    # ------------------------------------------------------------------
    # 剪贴板
    # ------------------------------------------------------------------

    def clip_set(self, text: str) -> str:
        def do():
            self.ui.root.clipboard_clear()
            self.ui.root.clipboard_append(text)
            self.ui.root.update_idletasks()
        self.ui.ui_call(do)
        self.store.set("last_clip", text)
        self._remember_clip(text)
        return f"剪贴板已更新（{len(text)} 字符）"

    def clip_get(self) -> str:
        result = {"text": ""}

        def do():
            try:
                result["text"] = self.ui.root.clipboard_get()
            except Exception:
                result["text"] = ""
        self.ui.ui_call(do)
        return result["text"]

    # ------------------------------------------------------------------
    # 电源
    # ------------------------------------------------------------------

    def power(self, action: str) -> str:
        if action not in POWER_ACTIONS:
            raise ActionError(f"不支持的电源动作: {action}")
        if not self.store.get("allow_power", True):
            raise ActionError("被控端已禁用电源操作")
        if not IS_WINDOWS:
            raise ActionError("电源控制目前仅支持 Windows")

        if action == "shutdown":
            cmd = ["shutdown", "/s", "/t", "5"]
            note = "将在 5 秒后关机"
        elif action == "restart":
            cmd = ["shutdown", "/r", "/t", "5"]
            note = "将在 5 秒后重启"
        elif action == "abort":
            cmd = ["shutdown", "/a"]
            note = "已取消关机"
        elif action == "lock":
            cmd = ["rundll32.exe", "user32.dll,LockWorkStation"]
            note = "已锁屏"
        else:  # sleep
            cmd = ["rundll32.exe", "powrprof.dll,SetSuspendState", "0,1,0"]
            note = "已进入睡眠"
        try:
            subprocess.Popen(
                cmd, shell=False,
                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                stdin=subprocess.DEVNULL,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        except Exception as exc:
            raise ActionError(f"执行失败: {exc}") from exc
        return note

    # ------------------------------------------------------------------
    # 自定义命令
    # ------------------------------------------------------------------

    def run_command(self, command: str, timeout: int = 60) -> dict:
        if not self.store.get("allow_run", True):
            raise ActionError("被控端已禁用自定义命令")
        if not command or not command.strip():
            raise ActionError("命令为空")
        self.ui.log(f"[自定义命令] {command}")
        try:
            # 打包成 --windowed 后没有控制台，cmd 的输出是 ANSI 代码页（本机 936）
            # 而不是 UTF-8 —— 用 utf-8 解会全是乱码。和本文件其它地方统一用 mbcs。
            # 另外必须加 CREATE_NO_WINDOW，否则每跑一条远程命令都会闪一个黑窗。
            proc = subprocess.run(
                command, shell=True, capture_output=True, text=True,
                timeout=timeout,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
                encoding="mbcs", errors="replace")
        except subprocess.TimeoutExpired:
            raise ActionError(f"命令超时（>{timeout}s）")
        except Exception as exc:
            raise ActionError(f"命令执行失败: {exc}") from exc
        return {
            "code": proc.returncode,
            "stdout": (proc.stdout or "")[-4000:],
            "stderr": (proc.stderr or "")[-2000:],
        }


    def _control_pushed_audio(self, action: str) -> str | None:
        """如果正在播推过去的音频，就控制它；否则返回 None 走系统媒体键。"""
        with self._lock:
            alias = self._audio_alias
            if not alias:
                return None
            mode = self._mci_mode(alias)
            if mode not in ("playing", "paused"):
                # MCI 已经放完了
                self._mci_close_locked()
                self._audio_paused = False
                self._audio_name = ""
                return None
            if action == "stop":
                self._mci_close_locked()
                self._audio_paused = False
                self._audio_name = ""
                return "已停止播放"
            # playpause
            if mode == "paused" or self._audio_paused:
                if self._mci(f"resume {alias}"):
                    self._audio_paused = False
                    return f"继续播放 {self._audio_name}"
                return None
            if self._mci(f"pause {alias}"):
                self._audio_paused = True
                return f"已暂停 {self._audio_name}"
            return None

    # ------------------------------------------------------------------
    # 语音播报（系统自带 TTS）
    # ------------------------------------------------------------------

    def list_voices(self) -> list[str]:
        """列出系统可用的 TTS 音色。"""
        if not IS_WINDOWS:
            return []
        script = (
            "Add-Type -AssemblyName System.Speech;"
            "(New-Object System.Speech.Synthesis.SpeechSynthesizer)"
            ".GetInstalledVoices() | ForEach-Object { $_.VoiceInfo.Name }"
        )
        try:
            proc = subprocess.run(
                ["powershell", "-NoProfile", "-Command", script],
                capture_output=True, timeout=20,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
                encoding="mbcs", errors="replace")
        except Exception:
            return []
        return [line.strip() for line in (proc.stdout or "").splitlines() if line.strip()]

    def say(self, text: str, voice: str = "", rate: int = 0) -> str:
        """语音播报。voice 为空则用系统默认（或设置里的默认音色）。"""
        if not text.strip():
            raise ActionError("播报内容为空")
        if not IS_WINDOWS:
            raise ActionError("语音播报目前仅支持 Windows")

        chosen = (voice or "").strip() or str(self.store.get("tts_voice") or "").strip()
        try:
            speed = int(rate) if rate else int(self.store.get("tts_rate", 0) or 0)
        except (TypeError, ValueError):
            speed = 0
        speed = max(-10, min(10, speed))

        safe = text.replace("'", "''")
        parts = [
            "Add-Type -AssemblyName System.Speech;",
            "$s = New-Object System.Speech.Synthesis.SpeechSynthesizer;",
        ]
        if chosen:
            parts.append(f"try {{ $s.SelectVoice('{chosen.replace(chr(39), chr(39) * 2)}') }} "
                         "catch { };")
        if speed:
            parts.append(f"$s.Rate = {speed};")
        parts.append(f"$s.Speak('{safe}')")
        script = "".join(parts)

        try:
            subprocess.Popen(
                ["powershell", "-NoProfile", "-WindowStyle", "Hidden",
                 "-Command", script],
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        except Exception as exc:
            raise ActionError(f"TTS 启动失败: {exc}") from exc
        return f"已播报（{chosen or '默认音色'}）"

    # ------------------------------------------------------------------
    # 播放本机音频文件（MP3 等）
    # ------------------------------------------------------------------

    AUDIO_EXTS = (".mp3", ".wav", ".m4a", ".flac", ".aac", ".ogg", ".wma")

    def audio_roots(self) -> list[Path]:
        """允许播放的目录。只认音乐相关目录，避免被用来播任意系统文件。"""
        roots: list[Path] = []
        configured = str(self.store.get("mp3_dir") or "").strip()
        if configured:
            path = Path(configured)
            if path.is_dir():
                roots.append(path)
        home = Path.home()
        for name in ("Music", "音乐", "Downloads", "下载"):
            path = home / name
            if path.is_dir():
                roots.append(path)
        return roots

    def list_local_audio(self, limit: int = 300) -> list[dict]:
        """列出本机可播放的音频文件。"""
        items: list[dict] = []
        seen: set[str] = set()
        for root in self.audio_roots():
            try:
                paths = sorted(root.rglob("*"))
            except OSError:
                continue
            for path in paths:
                if len(items) >= limit:
                    return items
                try:
                    if not path.is_file():
                        continue
                    if path.suffix.lower() not in self.AUDIO_EXTS:
                        continue
                    key = str(path).lower()
                    if key in seen:
                        continue
                    seen.add(key)
                    items.append({
                        "name": path.name,
                        "path": str(path),
                        "size": path.stat().st_size,
                        "dir": root.name,
                    })
                except OSError:
                    continue
        items.sort(key=lambda it: it["name"].lower())
        return items

    # ---- 音效（网络热门音效，手表上点一下电脑就响）----
    #
    #  素材放哪：data/sfx/
    #    内置的几个是程序**合成**的（正弦/方波现场算出来），
    #    所以零版权风险、也不占安装包体积。
    #    用户想用真正的一模一样的素材，把文件丢进 data/sfx/ 即可，
    #    列表会自动带上（见 sfx.py 顶部的说明）。

    def _sfx(self):
        """懒加载音效库。第一次调用时才生成内置 WAV。"""
        if getattr(self, "_sfx_obj", None) is None:
            from .sfx import Sfx
            self._sfx_obj = Sfx(self.store.dir, player=self.play_audio)
        return self._sfx_obj

    def list_sfx(self) -> list:
        return self._sfx().list_sfx()

    def sfx_dir_hint(self) -> str:
        return self._sfx().path_hint()

    def play_sfx(self, name: str) -> str:
        return self._sfx().play(name)

    def play_local(self, path: str) -> str:
        """播放本机某个音频文件。只允许音乐目录内的文件。"""
        target = Path(path)
        if not target.is_file():
            raise ActionError(f"文件不存在: {path}")
        if target.suffix.lower() not in self.AUDIO_EXTS:
            raise ActionError(f"不支持的音频格式: {target.suffix}")

        allowed = False
        for root in self.audio_roots():
            try:
                target.resolve().relative_to(root.resolve())
                allowed = True
                break
            except (ValueError, OSError):
                continue
        if not allowed:
            raise ActionError("只允许播放音乐目录里的文件")

        self.play_audio(str(target))
        return f"正在播放 {target.name}"

    # ------------------------------------------------------------------
    # 定时关机 / 关显示器
    # ------------------------------------------------------------------

    def shutdown_timer(self, minutes: int = 0, cancel: bool = False) -> str:
        """定时关机；cancel=True 时取消。minutes<=0 也会取消。"""
        if not IS_WINDOWS:
            raise ActionError("仅支持 Windows")
        if cancel or minutes <= 0:
            proc = subprocess.run(["shutdown", "/a"], capture_output=True,
                                  creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
            text = (proc.stdout or proc.stderr or b"").decode("mbcs", errors="replace")
            if proc.returncode != 0 and "1116" not in text and "没有" not in text:
                return "当前没有待执行的关机计划"
            return "已取消定时关机"
        minutes = max(1, min(1440, int(minutes)))
        subprocess.run(["shutdown", "/s", "/t", str(minutes * 60)],
                       capture_output=True,
                       creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        return f"将在 {minutes} 分钟后关机"

    def monitor_power(self, on: bool = False) -> str:
        """关屏 / 唤醒屏幕。"""
        if not IS_WINDOWS:
            raise ActionError("仅支持 Windows")
        import ctypes
        HWND_BROADCAST = 0xFFFF
        WM_SYSCOMMAND = 0x0112
        SC_MONITORPOWER = 0xF170
        # 2 = 关闭, -1 = 打开
        ctypes.windll.user32.SendMessageW(HWND_BROADCAST, WM_SYSCOMMAND,
                                          SC_MONITORPOWER, -1 if on else 2)
        return "显示器已打开" if on else "显示器已关闭"

    # ------------------------------------------------------------------
    # 打开网址 / 快捷命令 / 常用文件夹
    # ------------------------------------------------------------------

    def open_url(self, url: str) -> str:
        text = (url or "").strip()
        if not text:
            raise ActionError("网址为空")
        if "://" not in text:
            text = "https://" + text
        if not text.lower().startswith(("http://", "https://")):
            raise ActionError("只允许打开 http/https 网址")
        try:
            os.startfile(text)                            # noqa: S606
        except Exception as exc:                          # noqa: BLE001
            raise ActionError(f"打开失败: {exc}") from exc
        return f"已在电脑打开 {text}"

    QUICK_APPS = {
        "taskmgr": ("任务管理器", "taskmgr.exe"),
        "explorer": ("资源管理器", "explorer.exe"),
        "cmd": ("命令提示符", "cmd.exe"),
        "calc": ("计算器", "calc.exe"),
        "notepad": ("记事本", "notepad.exe"),
        "control": ("控制面板", "control.exe"),
        "mspaint": ("画图", "mspaint.exe"),
        "devmgmt": ("设备管理器", "devmgmt.msc"),
    }

    def quick_launch(self, key: str) -> str:
        item = self.QUICK_APPS.get((key or "").strip().lower())
        if item is None:
            raise ActionError(f"没有这个快捷程序: {key}")
        label, command = item
        try:
            subprocess.Popen(command, shell=True,
                             creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        except Exception as exc:                          # noqa: BLE001
            raise ActionError(f"启动失败: {exc}") from exc
        return f"已打开{label}"

    def open_folder(self, key: str) -> str:
        home = Path.home()
        folders = {
            "downloads": ("下载", home / "Downloads"),
            "documents": ("文档", home / "Documents"),
            "desktop": ("桌面", home / "Desktop"),
            "pictures": ("图片", home / "Pictures"),
            "music": ("音乐", home / "Music"),
            "received": ("接收目录", self.store.received_dir()),
            "app": ("程序目录", Path(__file__).resolve().parent.parent),
        }
        item = folders.get((key or "").strip().lower())
        if item is None:
            raise ActionError(f"没有这个文件夹: {key}")
        label, path = item
        if not path.is_dir():
            raise ActionError(f"{label}目录不存在: {path}")
        try:
            os.startfile(str(path))                       # noqa: S606
        except Exception as exc:                          # noqa: BLE001
            raise ActionError(f"打开失败: {exc}") from exc
        return f"已打开{label}"

    # ------------------------------------------------------------------
    # 进程
    # ------------------------------------------------------------------

    def list_processes(self, limit: int = 40) -> list[dict]:
        if not IS_WINDOWS:
            return []
        try:
            proc = subprocess.run(
                ["tasklist", "/FO", "CSV", "/NH"], capture_output=True,
                timeout=20, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
                encoding="mbcs", errors="replace")
        except Exception:                                 # noqa: BLE001
            return []
        items: list[dict] = []
        for raw in (proc.stdout or "").splitlines():
            parts = [p.strip().strip('"') for p in raw.split('","')]
            if len(parts) < 5:
                continue
            name = parts[0].strip('"')
            pid = parts[1]
            memory = parts[4].replace(",", "").replace(" K", "").strip()
            try:
                mem_kb = int(memory)
            except ValueError:
                mem_kb = 0
            if not pid.isdigit():
                continue
            items.append({"name": name, "pid": int(pid), "mem_kb": mem_kb})
        items.sort(key=lambda it: it["mem_kb"], reverse=True)
        return items[:limit]

    def kill_process(self, pid: int = 0, name: str = "") -> str:
        if not IS_WINDOWS:
            raise ActionError("仅支持 Windows")
        args = ["taskkill", "/F"]
        if pid:
            args += ["/PID", str(int(pid))]
            label = f"PID {pid}"
        elif name:
            safe = "".join(ch for ch in name if ch.isalnum() or ch in "._- ")
            if not safe:
                raise ActionError("进程名不合法")
            args += ["/IM", safe]
            label = safe
        else:
            raise ActionError("要指定 PID 或进程名")
        proc = subprocess.run(args, capture_output=True,
                              creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
                              encoding="mbcs", errors="replace")
        if proc.returncode != 0:
            raise ActionError(f"结束失败: {(proc.stdout or proc.stderr or '').strip()[:80]}")
        return f"已结束 {label}"

    # ------------------------------------------------------------------
    # 状态：电量 / 磁盘 / 性能
    # ------------------------------------------------------------------

    def battery(self) -> dict:
        if not IS_WINDOWS:
            return {"present": False}
        import ctypes

        class PowerStatus(ctypes.Structure):
            _fields_ = [("ACLineStatus", ctypes.c_byte),
                        ("BatteryFlag", ctypes.c_byte),
                        ("BatteryLifePercent", ctypes.c_byte),
                        ("SystemStatusFlag", ctypes.c_byte),
                        ("BatteryLifeTime", ctypes.c_ulong),
                        ("BatteryFullLifeTime", ctypes.c_ulong)]

        status = PowerStatus()
        if not ctypes.windll.kernel32.GetSystemPowerStatus(ctypes.byref(status)):
            return {"present": False}
        if status.BatteryLifePercent == 255 or status.BatteryFlag == 128:
            return {"present": False, "note": "台式机，没有电池"}
        return {
            "present": True,
            "percent": int(status.BatteryLifePercent),
            "charging": bool(status.ACLineStatus),
            "seconds_left": int(status.BatteryLifeTime)
            if status.BatteryLifeTime not in (0xFFFFFFFF, 0) else 0,
        }

    def disk(self) -> list[dict]:
        if not IS_WINDOWS:
            return []
        import ctypes
        import string

        out: list[dict] = []
        bitmask = ctypes.windll.kernel32.GetLogicalDrives()
        free = ctypes.c_ulonglong(0)
        total = ctypes.c_ulonglong(0)
        for i, letter in enumerate(string.ascii_uppercase):
            if not (bitmask >> i) & 1:
                continue
            root = f"{letter}:\\"
            try:
                ok = ctypes.windll.kernel32.GetDiskFreeSpaceExW(
                    ctypes.c_wchar_p(root), None, ctypes.byref(total),
                    ctypes.byref(free))
            except Exception:                             # noqa: BLE001
                continue
            if not ok or total.value == 0:
                continue
            out.append({
                "drive": f"{letter}:",
                "total_gb": round(total.value / 1073741824, 1),
                "free_gb": round(free.value / 1073741824, 1),
                "used_percent": int((total.value - free.value) * 100 / total.value),
            })
        return out

    def perf(self) -> dict:
        if not IS_WINDOWS:
            return {}
        import ctypes

        class MemoryStatus(ctypes.Structure):
            _fields_ = [("dwLength", ctypes.c_ulong),
                        ("dwMemoryLoad", ctypes.c_ulong),
                        ("ullTotalPhys", ctypes.c_ulonglong),
                        ("ullAvailPhys", ctypes.c_ulonglong),
                        ("ullTotalPageFile", ctypes.c_ulonglong),
                        ("ullAvailPageFile", ctypes.c_ulonglong),
                        ("ullTotalVirtual", ctypes.c_ulonglong),
                        ("ullAvailVirtual", ctypes.c_ulonglong),
                        ("ullAvailExtendedVirtual", ctypes.c_ulonglong)]

        mem = MemoryStatus()
        mem.dwLength = ctypes.sizeof(MemoryStatus)
        ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(mem))

        idle = ctypes.c_ulonglong()
        kernel = ctypes.c_ulonglong()
        user = ctypes.c_ulonglong()
        ctypes.windll.kernel32.GetSystemTimes(ctypes.byref(idle),
                                              ctypes.byref(kernel),
                                              ctypes.byref(user))
        cpu = 0
        try:
            total = kernel.value + user.value
            if total and self._last_cpu_total:
                delta_total = total - self._last_cpu_total
                delta_idle = idle.value - self._last_cpu_idle
                if delta_total > 0:
                    cpu = int(round((1 - delta_idle / delta_total) * 100))
        except Exception:                                 # noqa: BLE001
            cpu = 0
        self._last_cpu_total = kernel.value + user.value
        self._last_cpu_idle = idle.value

        import platform as _platform
        return {
            "cpu_percent": max(0, min(100, cpu)),
            "mem_percent": int(mem.dwMemoryLoad),
            "mem_used_gb": round((mem.ullTotalPhys - mem.ullAvailPhys) / 1073741824, 1),
            "mem_total_gb": round(mem.ullTotalPhys / 1073741824, 1),
            "uptime": self._uptime(),
            "os": f"{_platform.system()} {_platform.release()}",
        }

    def _uptime(self) -> str:
        try:
            import ctypes
            # 不设 restype 的话 ctypes 默认按 32 位处理，
            # 开机超过 24.8 天会被截断成负数，时长显示错乱
            ctypes.windll.kernel32.GetTickCount64.restype = ctypes.c_ulonglong
            ms = ctypes.windll.kernel32.GetTickCount64()
            seconds = int(ms / 1000)
            days, rem = divmod(seconds, 86400)
            hours, rem = divmod(rem, 3600)
            minutes = rem // 60
            if days:
                return f"{days} 天 {hours} 小时"
            if hours:
                return f"{hours} 小时 {minutes} 分"
            return f"{minutes} 分钟"
        except Exception:                                 # noqa: BLE001
            return "-"

    # ------------------------------------------------------------------
    # 剪贴板历史
    # ------------------------------------------------------------------

    def clip_history(self) -> list[dict]:
        return list(self._clip_history)

    def _remember_clip(self, text: str) -> None:
        text = (text or "").strip()
        if not text:
            return
        with self._lock:
            self._clip_history = [x for x in self._clip_history
                                  if x.get("text") != text]
            self._clip_history.insert(0, {
                "text": text[:2000],
                "at": time.strftime("%H:%M:%S"),
            })
            del self._clip_history[20:]

    def say_clipboard(self) -> str:
        # 之前读的是 store["last_clip"]（那是手表推过来的内容），
        # 手表没推过就永远报"剪贴板是空的"，可电脑剪贴板明明有东西。
        # 应该读电脑本机的剪贴板。
        try:
            text = str(self.clip_get() or "").strip()
        except Exception:                                 # noqa: BLE001
            text = ""
        if not text:
            text = str(self.store.get("last_clip") or "").strip()
        if not text:
            raise ActionError("电脑剪贴板是空的")
        return self.say(text[:500] + ("…" if len(text) > 500 else ""))

    # ------------------------------------------------------------------
    # 壁纸
    # ------------------------------------------------------------------

    def set_wallpaper(self, path: str = "") -> str:
        """把某张图片设为桌面壁纸。默认用最近收到的那张图。"""
        if not IS_WINDOWS:
            raise ActionError("仅支持 Windows")
        import ctypes

        target = Path(path) if path else None
        if target is None or not target.is_file():
            received = self.store.received_dir()
            images = sorted(
                [p for p in received.iterdir()
                 if p.is_file() and p.suffix.lower() in (".png", ".jpg", ".jpeg", ".bmp")],
                key=lambda p: p.stat().st_mtime, reverse=True)
            if not images:
                raise ActionError("接收目录里还没有图片")
            target = images[0]

        SPI_SETDESKWALLPAPER = 0x0014
        SPIF_UPDATEINIFILE = 0x01
        SPIF_SENDCHANGE = 0x02
        ok = ctypes.windll.user32.SystemParametersInfoW(
            SPI_SETDESKWALLPAPER, 0, ctypes.c_wchar_p(str(target)),
            SPIF_UPDATEINIFILE | SPIF_SENDCHANGE)
        if not ok:
            raise ActionError("设置壁纸失败")
        return f"壁纸已设为 {target.name}"

    # ------------------------------------------------------------------
    # 通知（托盘气泡，可见）
    # ------------------------------------------------------------------

    def notify(self, title: str, text: str) -> str:
        self.ui.notify(title or "局域网遥控", text or "")
        return "已通知"

    # ------------------------------------------------------------------
    # 系统信息
    # ------------------------------------------------------------------

    def info(self) -> dict:
        import platform
        screen_ok = False
        screen_size = (0, 0)
        try:
            from .screen import ScreenCapturer
            capturer = ScreenCapturer()
            screen_ok = capturer.available()
            if screen_ok:
                screen_size = capturer.screen_size()
        except Exception:
            screen_ok = False
        return {
            "server_name": self.store.get("server_name", ""),
            "os": f"{platform.system()} {platform.release()}",
            "python": platform.python_version(),
            "port": self.store.get("port", 0),
            "allow_run": bool(self.store.get("allow_run", True)),
            "allow_power": bool(self.store.get("allow_power", True)),
            "allow_screen": bool(self.store.get("allow_screen", True)),
            "screen_supported": screen_ok,
            "screen_width": screen_size[0],
            "screen_height": screen_size[1],
            "received_dir": str(self.store.received_dir()),
        }


def open_path(path: str) -> None:
    """用系统默认程序打开文件/文件夹（仅本机 UI 按钮使用）。"""
    if IS_WINDOWS:
        os.startfile(path)                                # noqa: S606
    elif sys.platform == "darwin":
        subprocess.Popen(["open", path])
    else:
        subprocess.Popen(["xdg-open", path])