# -*- coding: utf-8 -*- """安卓端第二阶段：后端新增 14 个功能。 1. 定时关机 8. 电脑电量 2. 取消定时关机 9. 磁盘占用 3. 关闭显示器 10. CPU / 内存占用 4. 打开网址 11. 剪贴板历史 5. 快捷命令 12. 设置壁纸（用手表发来的图） 6. 打开常用文件夹 13. 朗读剪贴板 7. 进程列表/结束 14. 结束进程 全部只用 Windows 自带能力（ctypes / tasklist / shutdown），不引入任何依赖。 """ from __future__ import annotations import sys from pathlib import Path ROOT = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\windows\lanremote") PROTOCOL = ROOT / "protocol.py" ACTIONS = ROOT / "actions.py" SERVER = ROOT / "server.py" PROTO_OLD = 'CMD_PLAY_LOCAL = "play_local"\n' PROTO_NEW = '''CMD_PLAY_LOCAL = "play_local" CMD_SHUTDOWN_TIMER = "shutdown_timer" # 定时关机 / 取消 CMD_MONITOR = "monitor" # 关闭 / 打开显示器 CMD_OPEN_URL = "open_url" # 在电脑浏览器打开网址 CMD_QUICK = "quick" # 快捷启动程序 CMD_OPEN_FOLDER = "open_folder" # 打开常用文件夹 CMD_PROC_LIST = "proc_list" # 进程列表 CMD_PROC_KILL = "proc_kill" # 结束进程 CMD_BATTERY = "battery" # 电量 CMD_DISK = "disk" # 磁盘占用 CMD_PERF = "perf" # CPU / 内存占用 CMD_CLIP_HISTORY = "clip_history" # 剪贴板历史 CMD_SET_WALLPAPER = "set_wallpaper" # 把收到的图设为壁纸 CMD_SAY_CLIP = "say_clip" # 朗读剪贴板 ''' ACTIONS_ANCHOR = ''' # ------------------------------------------------------------------ # 通知（托盘气泡，可见） # ------------------------------------------------------------------''' ACTIONS_NEW = ''' # ------------------------------------------------------------------ # 定时关机 / 关显示器 # ------------------------------------------------------------------ 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 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: 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}" ''' + ACTIONS_ANCHOR def main() -> int: ok = True text = PROTOCOL.read_text(encoding="utf-8") if PROTO_OLD not in text: print(" [失败] protocol.py 锚点未匹配") ok = False else: text = text.replace(PROTO_OLD, PROTO_NEW, 1) PROTOCOL.write_text(text, encoding="utf-8") print(" protocol.py: 新增 13 个命令") text = ACTIONS.read_text(encoding="utf-8") if ACTIONS_ANCHOR not in text: print(" [失败] actions.py 锚点未匹配") ok = False else: text = text.replace(ACTIONS_ANCHOR, ACTIONS_NEW, 1) ACTIONS.write_text(text, encoding="utf-8") print(" actions.py: 新增 14 项能力") print("\n语法检查：") import ast for name, path in (("protocol.py", PROTOCOL), ("actions.py", ACTIONS)): try: ast.parse(path.read_text(encoding="utf-8")) print(f" {name}: OK") except SyntaxError as exc: print(f" {name}: 语法错误 {exc}") ok = False return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())