xmz7891/ykxmz
1
0
Fork 0
mirror of https://gitee.com/xmz7891/ykxmz.git synced 2026-10-18 13:40:45 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
41bf110e22dbb27171fa2a4c2eb8f9bccd14402c
ykxmz/_build/patch_backend_14.py
T
xmz7891 b62ac865ea ykxmz 1.5.0：修复手表端 TLS 证书校验 + MTU 修复网络稳定性 
被控端
- 修复 MTU 问题：夹 MSS + 路径 MTU 探测（公网可用率 48% -> 100%）
- 复用 HTTP 连接（免掉每次新建 TLS 的 30% 握手失败）
- 每分钟回头找局域网，自动切回快 400 倍的路径
- 画面心跳帧：静止时省流量，客户端仍知道流是活的
- 锁屏时给出可操作的提示

手表端
- 内置 GTS Root R4 / GlobalSign Root CA，修 SSLHandshakeException
- 连接错误翻译成人话（证书/时间/解析/超时分别提示）
- 经中转连上后用服务器报的地址刷新已配对记录
- 新增局域网连接页面、服务器延迟检测

服务器端
- 支持 chunked 请求体（修每 16 秒一次的连接错乱与大面积 502）
- 网络调优脚本 net-tune.sh（开机自动应用）
- 管理页：批量删除、网络体检、下载区
- 隧道改 HTTP/2 + 双连接

文档
- 许可协议 / 安装前必读 / 安装后须知（作者：小麦子）
2026-10-06 19:01:05 +08:00

417 lines
17 KiB
Python
Raw Blame History

# -*- 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())
Reference in New Issue View Git Blame Copy Permalink