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

2.1.6：静默自动更新 + 远程鼠标控制 + 一键显示桌面

Browse Source 
用户这次提了一整套升级。这个提交先落地三块（其余几块在做）。

== 一、静默自动更新 ==

用户原话："手表端点击一个更新按钮，电脑端就可以自动进行静默更新，
其他什么设置都不变 …… 不弹任何窗口出来"。

链路：

    手表点按钮 -> 被控端问中转"最新是哪版"
               -> 下载到 %TEMP%\ykxmz-update\
               -> 校验 sha256
               -> 起一个**分离的 .cmd 帮手**
               -> 自己退出
               -> 帮手等它退出 -> 静默跑安装器 -> 重新拉起

关键设计说明：

  * **为什么要有那个 .cmd 帮手**：Windows 上正在运行的 exe 是锁着的，
    安装器覆盖不了，所以必须"先退出、再装、再起来"，而这三步
    不能由被控端自己做（它一退出就没人执行后面的了）。
    用 DETACHED_PROCESS 起的助手不受父进程退出影响。

  * **配置为什么不会丢**：配置在 %ProgramData%\ykxmz\data\，
    **不在程序目录里**。安装器升级只替换 {app} 下的文件，
    不碰 ProgramData —— 中转地址、配对码、各种开关原样保留。
    这是这个功能能做的前提。

  * **更新源是用户自己的中转站**，不是作者的服务器，也不是任何
    写死的公网地址。被控端问的是它自己配置的那个地址。
    中转没有这个接口（老版本服务端）时会明确提示"先手动升级一次服务端"。

  * **sha256 必须校验**：自更新最大的风险是下到一半断了或被换了，
    装上去是个坏程序、而且它还会自己再更新 —— 螺旋下去很难救。
    对不上就**放弃这一轮**，保持现状（现状至少能用）。

中转新增：GET /v1/update/latest?platform=windows
返回 version / name / url / size / sha256。版本比较按**数字段**
（不能按字符串，否则 "1.9.0" > "2.1.4"）。sha256 按
(文件名,大小,mtime) 缓存，避免每次重算 9MB。

手表端："电脑端版本：点这里查看" 按钮，查完显示
"电脑端 2.1.4 → 2.1.5（点这里更新）"，确认后才真的更新。

== 二、远程鼠标控制 ==

新增 windows/lanremote/input.py，用 SendInput 模拟鼠标键盘。

  * **为什么用 SendInput 而不是 mouse_event**：后者是 Win95 时代的
    API，会被 UIPI 拦，而且没有 dwExtraInfo（多个注入源会互相干扰）。
    SendInput 还能一次打包多个事件，按下+抬起之间不会插进别的输入。

  * **坐标是 0..65535 归一化**，而且**必须除以 (屏幕宽-1)**。
    少了那个 -1，最右边和最下边几像素点不到 —— 表现为
    "鼠标到不了屏幕边缘"，任务栏在底下的机器上尤其明显。

  * **INPUT 结构在 32/64 位下大小不同**（28 / 40），
    ctypes 里按进程位数算，不能写死。

  * 权限说明：目标窗口如果是管理员权限的（UAC 提升过的），
    普通权限的注入会被 UIPI 拦掉。这不是 bug，是 Windows 的设计。
    被控端以管理员运行就都能点。

== 三、一键显示桌面 ==

用户要求快捷栏里那个"不能改"的按钮就是这个功能，等于 Win+D。
和鼠标控制共用 input.py。

== 顺手修的一个自己造的坑 ==

往 server.py 插方法时，**两次**把 dispatch 的兜底
`return {"ok": False, "error": f"未知命令: {cmd}"}` 一起卷进了新方法里：

  * pyflakes 报 "undefined name 'cmd'"
  * 而且 dispatch 丢了兜底分支（未知命令会返回 None）

第一次靠 pyflakes 抓到，第二次也是。为了以后不再犯，
`_tests/test_update_flow.py` 里加了**结构断言**：
兜底必须在带 CMD_UPDATE 的那个方法体里，而各新方法里不能出现它。

（那条断言第一版写得太死 —— 要求"紧跟 CMD_UPDATE 分支"，
 后来往 dispatch 里插了新分支就误报了。改成验结构真相。）

== 验证 ==

    全量回归（含新增的 3a10 / 3a11 两关）全部 PASS
    静默自动更新：协议 + updater 15 项 + 服务端结构 8 项
    远程鼠标：坐标归一化 / 滚轮夹取 / Win+D 事件顺序 / 非法参数 22 项
    610 个 Python 文件语法通过
    实际注入鼠标成功（move / click / scroll / show_desktop）

== 还没做的（下一批） ==

    手表顶部快捷栏（可自定义最多 4 个，1 个固定"显示桌面"）
    屏幕镜像去掉提示条、去掉所有弹窗
    托盘图标改成设置项
    「允许新设备连接」开关（中转后台 + 安卓端）
    弹文字窗口位置/大小可自定义
    中转后台设备列表显示版本 + 远程点更新

== 顺手清理 ==

_tests/out_app/ 下面那堆 .class 是 Java 编译检查的产物，
以前误提交进仓库了（152 个文件）。这次从索引里移除并加进 .gitignore。
This commit is contained in:
xmz7891
2026-10-09 22:12:09 +08:00
parent 31a2360031
commit 4d97e40280
169 changed files with 2820 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
windows/lanremote/__init__.py
+1 -1
View File
@@ -1,5 +1,5 @@
# -*- coding: utf-8 -*-
"""局域网遥控 · Windows 被控端。"""
__version__ = "2.1.5"
__version__ = "2.1.6"
APP_NAME = "ykxmz"
windows/lanremote/input.py
+284
View File
@@ -0,0 +1,284 @@
# -*- coding: utf-8 -*-
r'''被控端的**输入注入**：模拟鼠标和键盘。
== 用来干什么 ==
1. 远程鼠标控制：移动 / 左键 / 右键 / 中键 / 滚轮
2. 「一键显示桌面」：Win+D
== 为什么用 SendInput 而不是 mouse_event ==
`mouse_event` / `keybd_event` 是 Win95 时代的旧 API，
微软文档明确说"新程序请用 SendInput"：
* 旧的会被 UIPI 拦（不同完整性级别的窗口之间）
* 旧的没有 `dwExtraInfo`，多个注入源会互相干扰
* SendInput 一次可以打包多个事件，**原子性好** ——
比如"按下+抬起"中间不会插进别的输入
== 坐标怎么算 ==
SendInput 的绝对坐标是 **0..65535 归一化**的，不是像素。
而且它是**按整个虚拟屏幕**（多显示器算成一个大方块）算的。
所以：
归一化 x = 像素 x * 65535 / (虚拟屏宽 - 1)
归一化 y = 像素 y * 65535 / (虚拟屏高 - 1)
⚠ 那个 `-1` 不能省。微软文档写得很清楚：0 和 65535 分别对应
屏幕最左/最右**像素**，所以要除以 `宽-1`。少了这个 -1，
最右边和最下边几像素点不到 —— 表现为"鼠标到不了屏幕边缘"，
任务栏在底下的机器上尤其明显。
== 权限 ==
被控端跑在普通用户下也能注入（同一会话内）。
但如果目标窗口是**管理员权限**的（比如 UAC 提升过的程序），
普通权限的 SendInput 会被 UIPI 拦掉，鼠标点不动。
这不是 bug，是 Windows 的安全设计（防止低权限程序操控高权限窗口）。
被控端如果以管理员运行就都能点。README 里说清楚这一点。
'''
from __future__ import annotations
import ctypes
import ctypes.wintypes as wt
user32 = ctypes.WinDLL("user32", use_last_error=True)
# ---- SendInput 的结构体 ----
#
# ⚠ INPUT 结构在 32 位和 64 位下**大小不一样**：
# 64 位下 union 前面有 4 字节的对齐填充，sizeof(INPUT) == 40；
# 32 位下是 28。
# 所以必须让 ctypes 按本进程的位数算，不能写死。
# （我们用 Python 3.8 **32 位**打包，所以实际是 28 ——
# 但代码要两边都对，免得以后换 64 位打包时静默出错。）
ULONG_PTR = ctypes.c_ulonglong if ctypes.sizeof(ctypes.c_void_p) == 8 \
else ctypes.c_ulong
class MOUSEINPUT(ctypes.Structure):
_fields_ = [("dx", wt.LONG),
("dy", wt.LONG),
("mouseData", wt.DWORD),
("dwFlags", wt.DWORD),
("time", wt.DWORD),
("dwExtraInfo", ULONG_PTR)]
class KEYBDINPUT(ctypes.Structure):
_fields_ = [("wVk", wt.WORD),
("wScan", wt.WORD),
("dwFlags", wt.DWORD),
("time", wt.DWORD),
("dwExtraInfo", ULONG_PTR)]
class HARDWAREINPUT(ctypes.Structure):
_fields_ = [("uMsg", wt.DWORD),
("wParamL", wt.WORD),
("wParamH", wt.WORD)]
class _INPUTunion(ctypes.Union):
_fields_ = [("mi", MOUSEINPUT),
("ki", KEYBDINPUT),
("hi", HARDWAREINPUT)]
class INPUT(ctypes.Structure):
_fields_ = [("type", wt.DWORD),
("union", _INPUTunion)]
INPUT_MOUSE = 0
INPUT_KEYBOARD = 1
# 鼠标标志
MOUSEEVENTF_MOVE = 0x0001
MOUSEEVENTF_LEFTDOWN = 0x0002
MOUSEEVENTF_LEFTUP = 0x0004
MOUSEEVENTF_RIGHTDOWN = 0x0008
MOUSEEVENTF_RIGHTUP = 0x0010
MOUSEEVENTF_MIDDLEDOWN = 0x0020
MOUSEEVENTF_MIDDLEUP = 0x0040
MOUSEEVENTF_WHEEL = 0x0800
MOUSEEVENTF_HWHEEL = 0x1000
MOUSEEVENTF_ABSOLUTE = 0x8000
MOUSEEVENTF_VIRTUALDESK = 0x4000
# 键盘标志
KEYEVENTF_KEYUP = 0x0002
KEYEVENTF_EXTENDEDKEY = 0x0001
VK_LWIN = 0x5B
VK_D = 0x44
VK_ESCAPE = 0x1B
WHEEL_DELTA = 120
# 屏幕指标
SM_XVIRTUALSCREEN = 76
SM_YVIRTUALSCREEN = 77
SM_CXVIRTUALSCREEN = 78
SM_CYVIRTUALSCREEN = 79
# 屏幕宽高在进程生命周期里基本不变，缓存一下
_screen_cache: dict = {}
class InputError(Exception):
"""注入失败（被系统拦了、参数不对等）。"""
def _screen() -> tuple:
"""整个虚拟屏幕的 (左, 上, 宽, 高)。多显示器时是包住所有屏的矩形。"""
if not _screen_cache:
_screen_cache["v"] = (
user32.GetSystemMetrics(SM_XVIRTUALSCREEN),
user32.GetSystemMetrics(SM_YVIRTUALSCREEN),
user32.GetSystemMetrics(SM_CXVIRTUALSCREEN),
user32.GetSystemMetrics(SM_CYVIRTUALSCREEN),
)
return _screen_cache["v"]
def refresh_screen() -> None:
"""显示器分辨率变了之后调一下，让缓存重新读。"""
_screen_cache.clear()
def _send(*inputs: INPUT) -> None:
"""把一批事件一次性发出去。"""
if not inputs:
return
n = len(inputs)
arr = (INPUT * n)(*inputs)
sent = user32.SendInput(n, ctypes.byref(arr), ctypes.sizeof(INPUT))
if sent != n:
err = ctypes.get_last_error()
# 0x57 = ERROR_INVALID_PARAMETER，多半是被 UIPI 拦了
if err == 87:
raise InputError("系统拒绝了这次输入（目标窗口权限比被控端高？"
"把被控端用管理员身份运行就能解决）")
raise InputError("注入输入失败（SendInput 返回 %d/%d，错误码 %d）"
% (sent, n, err))
def _mouse(dx: int, dy: int, data: int, flags: int) -> INPUT:
return INPUT(type=INPUT_MOUSE,
union=_INPUTunion(mi=MOUSEINPUT(dx, dy, data, flags, 0, 0)))
def _key(vk: int, up: bool = False) -> INPUT:
flags = KEYEVENTF_KEYUP if up else 0
# 左右 Win / 方向键这些是扩展键（不是必须，但更准确）
if vk in (VK_LWIN, 0x5C, 0x25, 0x26, 0x27, 0x28):
flags |= KEYEVENTF_EXTENDEDKEY
return INPUT(type=INPUT_KEYBOARD,
union=_INPUTunion(ki=KEYBDINPUT(vk, 0, flags, 0, 0)))
# ---------------------------------------------------------------- 鼠标
def move_to(x: int, y: int) -> None:
"""把鼠标移到**屏幕像素坐标** (x, y)。
坐标是相对虚拟屏幕左上角的像素值；超出范围会被夹到边界内。
"""
vx, vy, vw, vh = _screen()
if vw <= 1 or vh <= 1:
raise InputError("拿不到屏幕尺寸")
# 夹到边界，免得归一化后溢出
px = max(vx, min(vx + vw - 1, int(x)))
py = max(vy, min(vy + vh - 1, int(y)))
# ⚠ 除以 (宽-1)，见文件头说明
nx = int(round((px - vx) * 65535 / (vw - 1)))
ny = int(round((py - vy) * 65535 / (vh - 1)))
_send(_mouse(nx, ny, 0,
MOUSEEVENTF_MOVE | MOUSEEVENTF_ABSOLUTE
| MOUSEEVENTF_VIRTUALDESK))
def move_by(dx: int, dy: int) -> None:
"""相对移动（有些场景比绝对坐标好用，比如拖拽）。"""
_send(_mouse(int(dx), int(dy), 0, MOUSEEVENTF_MOVE))
_BUTTONS = {
"left": (MOUSEEVENTF_LEFTDOWN, MOUSEEVENTF_LEFTUP),
"right": (MOUSEEVENTF_RIGHTDOWN, MOUSEEVENTF_RIGHTUP),
"middle": (MOUSEEVENTF_MIDDLEDOWN, MOUSEEVENTF_MIDDLEUP),
}
def click(button: str = "left", double: bool = False) -> None:
"""点一下（按下+抬起一次发出去，中间不会插进别的输入）。"""
b = _BUTTONS.get(str(button).lower())
if not b:
raise InputError("不认识的鼠标键：%s（只能 left/right/middle）"
% button)
down, up = b
seq = [_mouse(0, 0, 0, down), _mouse(0, 0, 0, up)]
if double:
seq = seq + seq
_send(*seq)
def down(button: str = "left") -> None:
b = _BUTTONS.get(str(button).lower())
if not b:
raise InputError("不认识的鼠标键：%s" % button)
_send(_mouse(0, 0, 0, b[0]))
def up(button: str = "left") -> None:
b = _BUTTONS.get(str(button).lower())
if not b:
raise InputError("不认识的鼠标键：%s" % button)
_send(_mouse(0, 0, 0, b[1]))
def scroll(amount: int, horizontal: bool = False) -> None:
"""滚轮。amount 正数向上/向右，负数向下/向左。
一个"格"按 120 算（Windows 的 WHEEL_DELTA）。
手表那边传 1/-1 这种格子数就行，这里乘 120。
"""
try:
n = int(amount)
except (TypeError, ValueError):
raise InputError("滚轮量必须是整数")
if n == 0:
return
# 夹一下，防止一次滚几百格把页面甩飞
n = max(-50, min(50, n))
flag = MOUSEEVENTF_HWHEEL if horizontal else MOUSEEVENTF_WHEEL
_send(_mouse(0, 0, (n * WHEEL_DELTA) & 0xFFFFFFFF, flag))
# ---------------------------------------------------------------- 键盘
def key(vk: int) -> None:
"""按一下某个虚拟键码。"""
_send(_key(int(vk)), _key(int(vk), up=True))
def combo(*vks: int) -> None:
"""组合键，比如 combo(VK_LWIN, VK_D) 就是 Win+D。
顺序：全部按下 -> 逆序抬起。这样修饰键不会卡住。
"""
if not vks:
return
seq = [_key(v) for v in vks]
seq += [_key(v, up=True) for v in reversed(vks)]
_send(*seq)
def show_desktop() -> None:
"""显示桌面（等于 Win+D）。
用户要求快捷栏里那个不能改的按钮就是这个功能。
"""
combo(VK_LWIN, VK_D)
windows/lanremote/protocol.py
+7
View File
@@ -61,6 +61,13 @@ CMD_FILE_PUSH = "file_push"
CMD_FILE_LIST = "file_list"
CMD_FILE_PULL = "file_pull"
CMD_INFO = "info"
# 静默自动更新：从**用户自己的中转站**取新版本包，静默装好并重启。
# action=check 只看不动；action=start 真的更新。
CMD_UPDATE = "update"
# 远程鼠标控制：{"action":"move|click|down|up|scroll", ...}
CMD_MOUSE = "mouse"
# 一键显示桌面（Win+D）。用户要求的快捷栏固定按钮。
CMD_SHOW_DESKTOP = "show_desktop"
CMD_SCREEN_START = "screen_start"
CMD_SCREEN_STOP = "screen_stop"
CMD_SCREENSHOT = "screenshot"
windows/lanremote/server.py
+176
View File
@@ -10,7 +10,11 @@ import threading
import time
from pathlib import Path
import tempfile
import sys
from . import protocol as P
from . import updater as U
from . import input as IN
from .actions import ActionError
from .config import Store
from .discovery import DISCOVERY_PORT, DiscoveryResponder
@@ -693,8 +697,180 @@ class RemoteServer:
if cmd == P.CMD_INFO:
return {"ok": True, "info": self.actions.info()}
if cmd == P.CMD_UPDATE:
return self._do_update(msg)
if cmd == P.CMD_MOUSE:
return self._do_mouse(msg)
if cmd == P.CMD_SHOW_DESKTOP:
try:
IN.show_desktop()
except IN.InputError as exc:
return {"ok": False, "error": str(exc)}
return {"ok": True, "done": "show_desktop"}
return {"ok": False, "error": f"未知命令: {cmd}"}
# ---------------- 远程鼠标 / 键盘注入 ----------------
def _do_mouse(self, msg: dict) -> dict:
"""处理 {"cmd":"mouse","action":...}。
坐标系是**屏幕像素**（相对虚拟屏幕左上角），由手表按比例映射好
再发过来 —— 电脑端不做缩放猜测，免得两边算法不一致导致漂移。
"""
action = str(msg.get("action") or "").lower()
try:
if action == "move":
# 分辨率可能被改过（投屏切换等），每次移动前刷一下缓存。
# 不刷的话，插上投影仪之后鼠标会跑到屏幕外。
IN.refresh_screen()
IN.move_to(int(msg.get("x", 0)), int(msg.get("y", 0)))
return {"ok": True}
if action == "moveby":
IN.move_by(int(msg.get("dx", 0)), int(msg.get("dy", 0)))
return {"ok": True}
if action == "click":
IN.click(str(msg.get("button") or "left"),
double=bool(msg.get("double")))
return {"ok": True}
if action == "down":
IN.down(str(msg.get("button") or "left"))
return {"ok": True}
if action == "up":
IN.up(str(msg.get("button") or "left"))
return {"ok": True}
if action == "scroll":
IN.scroll(int(msg.get("amount", 0)),
horizontal=bool(msg.get("horizontal")))
return {"ok": True}
if action == "desktop":
IN.show_desktop()
return {"ok": True}
return {"ok": False, "error": "不认识的鼠标动作：%s" % action}
except IN.InputError as exc:
return {"ok": False, "error": str(exc)}
except (TypeError, ValueError) as exc:
return {"ok": False, "error": "参数不对：%s" % exc}
# ---------------- 静默自动更新 ----------------
def _do_update(self, msg: dict) -> dict:
"""处理 {"cmd":"update","action":...}。
取包的地方是**用户自己配的中转站**（store 里的 relay_url），
不是任何写死的公网地址 —— 这一点是刻意的，见 updater.py 的说明。
"""
from . import __version__ as my_version
action = str(msg.get("action") or "check").lower()
relay = (self.store.relay_url() or "").strip()
if action == "check":
try:
info = U.check(relay, my_version)
except U.UpdateError as exc:
return {"ok": False, "error": str(exc)}
if not info.get("supported"):
return {"ok": False,
"error": info.get("error") or "不支持自动更新"}
if "error" in info:
return {"ok": False, "error": info["error"]}
return {
"ok": True,
"current": my_version,
"latest": info["latest"],
"needUpdate": bool(info["needUpdate"]),
"size": info.get("size", 0),
"name": info.get("name", ""),
}
if action != "start":
return {"ok": False, "error": "action 只能是 check 或 start"}
# ---- 真的更新 ----
# 先查一遍（在调用方线程里查，失败能立刻报回去）
try:
info = U.check(relay, my_version)
except U.UpdateError as exc:
return {"ok": False, "error": str(exc)}
if not info.get("supported") or "error" in info:
return {"ok": False,
"error": info.get("error") or "不支持自动更新"}
if not info["needUpdate"]:
return {"ok": True, "alreadyLatest": True, "current": my_version,
"message": "电脑端已经是最新版本 %s" % my_version}
# 下载和应用放到后台线程，先回一句让手表知道开始了
threading.Thread(
target=self._run_update,
args=(info, my_version),
name="ykxmz-update", daemon=True).start()
return {
"ok": True,
"started": True,
"from": my_version,
"to": info["latest"],
"size": info.get("size", 0),
"message": "正在下载 %s（%.1f MB），下完会自动装好并重启，"
"设置和配对码都不变"
% (info["latest"], (info.get("size") or 0) / 1048576.0),
}
def _run_update(self, info: dict, my_version: str) -> None:
"""后台线程：下载 -> 校验 -> 起帮手 -> 自己退出。"""
def log(text: str) -> None:
try:
self._log(text)
except Exception: # noqa: BLE001
pass
try:
log("[更新] 开始下载 %s（%.1f MB）"
% (info["latest"], (info.get("size") or 0) / 1048576.0))
work = Path(tempfile.gettempdir()) / "ykxmz-update"
last = [0]
def progress(got: int, total: int) -> None:
# 每 25% 报一次，别把日志刷爆
if total:
pct = int(got * 100 / total)
if pct - last[0] >= 25:
last[0] = pct
log("[更新] 已下载 %d%%（%.1f/%.1f MB）"
% (pct, got / 1048576.0, total / 1048576.0))
pkg = U.download(info["url"], work,
info.get("sha256", ""), info.get("size", 0),
progress=progress)
log("[更新] 下载完成并校验通过：%s" % pkg.name)
# 应用：起分离的 .cmd 帮手，然后**本进程必须退出**
app_exe = Path(sys.executable if getattr(sys, "frozen", False)
else sys.argv[0]).resolve()
logfile = U.apply_update(pkg, app_exe)
log("[更新] 已交给更新帮手（%s），被控端即将退出并重启" % logfile)
U.cleanup_old_downloads()
# 给日志和网络一点时间落完，再退出
time.sleep(2)
os._exit(0)
except U.UpdateError as exc:
log("[更新] 失败：%s（保持当前版本 %s）" % (exc, my_version))
except Exception as exc: # noqa: BLE001
log("[更新] 意外失败：%s: %s" % (type(exc).__name__, exc))
# ---------------- 握手实现 ----------------
def _do_pair(self, session: Session, msg: dict) -> dict:
windows/lanremote/updater.py
+294
View File
@@ -0,0 +1,294 @@
# -*- coding: utf-8 -*-
r'''被控端**静默自动更新**。
== 用户要的效果 ==
"手表端点击一个更新按钮，电脑端就自动进行静默更新，
其他什么设置都不变 …… 不弹任何窗口出来。"
== 流程 ==
1. 问中转：最新是哪版、在哪下、sha256 多少
GET <中转>/v1/update/latest?platform=windows
2. 和本机版本比，一样就直接返回"已是最新"
3. 下载安装包到 %TEMP%\ykxmz-update\
4. **校验 sha256**（下坏了/被换了都必须拦住）
5. 写一个 .cmd 帮手，**分离进程**启动它，然后自己退出
6. 帮手等本进程退出 -> 静默跑安装器 -> 重新拉起被控端
== 为什么要有第 5 步那个 .cmd ==
Windows 上**正在运行的 exe 文件是锁着的**，安装器覆盖不了。
所以必须"先退出、再安装、再起来"，而这三步不能由本进程做
（它一退出就没人执行后面的了）。
用 `DETACHED_PROCESS` 起的 .cmd 不依赖本进程存活，
本进程退出后它继续跑 —— 这是 Windows 自更新的标准做法。
== 配置为什么不会丢 ==
配置在 `%ProgramData%\ykxmz\data\`（见 config.py 的 data_dir），
**不在程序目录里**。安装器升级时只替换 `{app}` 下的文件，
不碰 ProgramData —— 所以中转地址、配对码、各种开关原样保留。
这一点是这个功能能做的前提。要是配置在程序目录里，
每次更新都会被清掉，那"设置不变"就无从谈起。
== 为什么 sha256 必须校验 ==
自更新的最大风险是"下到一半断了"或"下到的不是我们要的东西"，
装上去就是一个坏掉的程序，而且它还会自己再更新 —— 螺旋下去很难救。
sha256 对不上就**直接放弃这一轮**，保持现状（现状至少是能用的）。
== 安全边界（写清楚，免得以后自己忘了）==
这个功能**只能从中转站取包**，而中转站地址是用户自己配的。
所以：
* 用户自己搭的中转，就只从那里更新 —— 不存在"被第三方投毒"
* 中转上的包是用户自己上传的
* 校验和不匹配就拒绝，不做"绕过校验"的降级路径
换句话说：**更新源 = 用户自己的服务器**，不是作者的服务器，
也不是任何写死的公网地址。这一点是刻意的。
'''
from __future__ import annotations
import hashlib
import json
import os
import subprocess
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
# 中转没有这个接口时（老版本服务端）会返回 404/其它，直接当"不支持"处理
TIMEOUT_CHECK = 20
TIMEOUT_DOWNLOAD = 600
CHUNK = 256 * 1024
class UpdateError(Exception):
"""更新过程中出的问题（会被上层翻成中文提示）。"""
def _ver_tuple(v: str) -> tuple:
"""'2.1.5' -> (2,1,5)。解析不了就当 (0,0,0)。"""
try:
parts = [int(x) for x in str(v).strip().split(".")]
while len(parts) < 3:
parts.append(0)
return tuple(parts[:3])
except (ValueError, AttributeError):
return (0, 0, 0)
def _ua() -> str:
# 用自家标识。Cloudflare 会把 Python-urllib 判为可疑客户端直接 403
# （报 error code 1010），这个坑仓库里记过两次了。
return "LanRemote-updater/1.0"
def check(relay_url: str, current_version: str,
timeout: int = TIMEOUT_CHECK) -> dict:
"""问中转最新版。返回：
{"supported": bool, 中转支持不支持这个接口
"latest": "2.1.5",
"url": "https://.../download/ykxmz-setup-2.1.5.exe",
"name": "...", "size": 123, "sha256": "...",
"needUpdate": bool}
"""
base = (relay_url or "").strip().rstrip("/")
if not base:
raise UpdateError("还没配中转地址 —— 自动更新要从你自己的中转站取包")
if not base.startswith(("http://", "https://")):
base = "https://" + base
url = base + "/v1/update/latest?platform=windows"
req = urllib.request.Request(url)
req.add_header("User-Agent", _ua())
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
if exc.code == 404:
return {"supported": False,
"error": "中转站版本太旧，不支持自动更新（先手动升级一次服务端）"}
raise UpdateError("查询更新失败：HTTP %s" % exc.code)
except Exception as exc: # noqa: BLE001
raise UpdateError("连不上中转站：%s" % exc)
if not data.get("ok"):
return {"supported": True,
"error": data.get("error") or "中转站上还没有可下载的安装包"}
latest = str(data.get("version") or "")
url_path = str(data.get("url") or "")
if not latest or not url_path:
raise UpdateError("中转返回的信息不完整")
full = url_path if url_path.startswith("http") else base + url_path
return {
"supported": True,
"latest": latest,
"current": current_version,
"url": full,
"name": data.get("name") or "",
"size": int(data.get("size") or 0),
"sha256": (data.get("sha256") or "").lower(),
"needUpdate": _ver_tuple(latest) > _ver_tuple(current_version),
}
def download(url: str, dest_dir: Path, expect_sha256: str = "",
expect_size: int = 0,
progress=None) -> Path:
"""下载安装包并校验。返回下载好的路径。
progress: 可选回调 (已下载字节, 总字节)
"""
dest_dir.mkdir(parents=True, exist_ok=True)
name = url.rsplit("/", 1)[-1].split("?")[0] or "ykxmz-update.exe"
# 只留文件名，别让 URL 里的东西跑到路径里（防目录穿越）
name = os.path.basename(urllib.parse.unquote(name))
if not name or name.startswith("."):
name = "ykxmz-update.exe"
dest = dest_dir / name
req = urllib.request.Request(url)
req.add_header("User-Agent", _ua())
try:
with urllib.request.urlopen(req, timeout=TIMEOUT_DOWNLOAD) as resp:
total = int(resp.headers.get("Content-Length") or 0)
h = hashlib.sha256()
got = 0
with dest.open("wb") as f:
while True:
chunk = resp.read(CHUNK)
if not chunk:
break
f.write(chunk)
h.update(chunk)
got += len(chunk)
if progress:
try:
progress(got, total)
except Exception: # noqa: BLE001
pass
except Exception as exc: # noqa: BLE001
try:
dest.unlink()
except OSError:
pass
raise UpdateError("下载失败：%s" % exc)
if expect_size and got != expect_size:
dest.unlink(missing_ok=True)
raise UpdateError("下载大小对不上（期望 %d，实际 %d）"
% (expect_size, got))
if expect_sha256 and h.hexdigest() != expect_sha256:
dest.unlink(missing_ok=True)
raise UpdateError("校验和不匹配 —— 包可能下坏了，已丢弃（保持当前版本）")
return dest
WAITER_TEMPLATE = r'''@echo off
rem ===================================================================
rem ykxmz 自动更新的"帮手"脚本
rem
rem 被控端自己要退出了，没法再做后面的事，所以把这三步交给这个脚本：
rem 1. 等被控端真的退出（不然安装器覆盖不了正在运行的 exe）
rem 2. 静默跑安装器
rem 3. 重新拉起被控端
rem
rem 它由被控端用 DETACHED_PROCESS 启动，不受被控端退出影响。
rem 跑完自己删掉。
rem ===================================================================
setlocal
set "PID={pid}"
set "SETUP={setup}"
set "APPEXE={appexe}"
set "LOGFILE={log}"
echo [%date% %time%] 等待进程 %PID% 退出... >> "%LOGFILE%"
set /a TRIES=0
:waitloop
tasklist /FI "PID eq %PID%" 2>nul | find "%PID%" >nul
if errorlevel 1 goto gone
set /a TRIES+=1
if %TRIES% GEQ 150 goto gone
timeout /t 2 /nobreak >nul
goto waitloop
:gone
echo [%date% %time%] 进程已退出，开始静默安装 >> "%LOGFILE%"
rem /VERYSILENT 全静默、不弹任何窗口
rem /SUPPRESSMSGBOXES 连错误框也不弹
rem /NORESTART 不自动重启电脑
rem 不带 /DIR，让它按注册表里记的上次安装位置原地升级
start "" /wait "%SETUP%" /VERYSILENT /SUPPRESSMSGBOXES /NORESTART
echo [%date% %time%] 安装器退出码 %ERRORLEVEL% >> "%LOGFILE%"
rem 给安装器一点时间把文件落完
timeout /t 3 /nobreak >nul
echo [%date% %time%] 重新拉起被控端 >> "%LOGFILE%"
if exist "%APPEXE%" start "" "%APPEXE%" --silent
timeout /t 3 /nobreak >nul
rem 清理：删掉安装包和自己
del /f /q "%SETUP%" >nul 2>&1
(goto) 2>nul & del /f /q "%~f0"
'''
def apply_update(installer: Path, app_exe: Path, pid: int = None) -> str:
"""写帮手脚本、分离启动它。调用方**紧接着必须自己退出**。
返回日志文件路径（写进被控端日志，方便事后查）。
"""
pid = pid or os.getpid()
work = Path(tempfile.gettempdir()) / "ykxmz-update"
work.mkdir(parents=True, exist_ok=True)
log = work / "update.log"
script = work / "do-update.cmd"
body = (WAITER_TEMPLATE
.replace("{pid}", str(pid))
.replace("{setup}", str(installer))
.replace("{appexe}", str(app_exe))
.replace("{log}", str(log)))
# .bat/.cmd 用系统 ANSI 编码（GBK），不能让中文变乱码
script.write_bytes(body.replace("\n", "\r\n").encode("mbcs", "replace"))
# DETACHED_PROCESS: 不继承控制台、不随父进程退出
# CREATE_NEW_PROCESS_GROUP: 不受父进程的 Ctrl 事件影响
flags = 0
flags |= getattr(subprocess, "DETACHED_PROCESS", 0x00000008)
flags |= getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200)
subprocess.Popen(["cmd.exe", "/c", str(script)],
creationflags=flags,
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True)
return str(log)
def cleanup_old_downloads(keep_days: int = 3) -> None:
"""清掉临时目录里过期的更新包（失败的残留）。"""
work = Path(tempfile.gettempdir()) / "ykxmz-update"
if not work.is_dir():
return
cutoff = time.time() - keep_days * 86400
try:
for f in work.iterdir():
if f.is_file() and f.stat().st_mtime < cutoff:
f.unlink()
except OSError:
pass