仓库初始化

This commit is contained in:
lwt
2026-05-22 00:16:08 +08:00
commit 95c96dce42
118 changed files with 23005 additions and 0 deletions
View File
+154
View File
@@ -0,0 +1,154 @@
# -*- coding: utf-8 -*-
"""
独立进程外更新器 (Out-of-Process Updater)
无第三方依赖(仅标准库 + 可选 ctypes 探测 PID),
打包为 mini_updater.exe 后置于安装目录,由主程序在“确认重启”后拉起。
负责:等待主进程退出 → 重命名替换 exe → 以正确 cwd 与 DETACHED_PROCESS 启动新进程。
"""
import argparse
import os
import subprocess
import sys
import time
# 仅标准库:PID 检测用 ctypesWindows)或 os.killUnix
if sys.platform == "win32":
import ctypes
try:
_kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined]
except Exception:
_kernel32 = None
else:
_kernel32 = None
def _log_error(install_dir: str, message: str, exc: Exception = None) -> None:
"""将错误写入安装目录下的 logs/update_error.log,便于排查。"""
try:
log_dir = os.path.join(install_dir, "logs")
os.makedirs(log_dir, exist_ok=True)
log_path = os.path.join(log_dir, "update_error.log")
with open(log_path, "a", encoding="utf-8") as f:
from datetime import datetime
line = "[%s] %s" % (datetime.now().isoformat(), message)
if exc is not None:
line += " (%s)" % exc
f.write(line + "\n")
except Exception:
pass
def is_process_alive(pid: int) -> bool:
"""
跨平台判断指定 PID 是否仍存活。
Windows: ctypes 调用 OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION)
Unix: os.kill(pid, 0)。
"""
if sys.platform == "win32" and _kernel32 is not None:
# PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
handle = _kernel32.OpenProcess(0x1000, False, pid)
if handle is None or handle == 0:
return False
try:
_kernel32.CloseHandle(handle)
except Exception:
pass
return True
try:
os.kill(pid, 0)
return True
except (OSError, ProcessLookupError, PermissionError):
return False
def main() -> None:
parser = argparse.ArgumentParser(description="独立更新守护:等主进程退出后替换 exe 并重启")
parser.add_argument("--pid", type=int, required=True, help="主程序进程 ID,必须等其完全退出")
parser.add_argument("--install-dir", type=str, required=True, help="主程序安装根目录")
parser.add_argument("--new-exe-path", type=str, required=True, help="已准备好的新版本 main.exe 的完整路径(通常在 TEMP)")
parser.add_argument("--target-exe-name", type=str, default="main.exe", help="主程序文件名,如 main.exe")
args = parser.parse_args()
install_dir = os.path.abspath(args.install_dir)
new_exe_path = os.path.abspath(args.new_exe_path)
target_exe_name = args.target_exe_name.strip() or "main.exe"
target_exe_full = os.path.join(install_dir, target_exe_name)
old_exe_full = target_exe_full + ".old"
# -------------------------------------------------------------------------
# 1. 死神凝视:每隔 0.5 秒检测 --pid 是否存活,必须等到进程彻底消失
# -------------------------------------------------------------------------
while is_process_alive(args.pid):
time.sleep(0.5)
# -------------------------------------------------------------------------
# 2. 障眼法替换 (The Swap)
# -------------------------------------------------------------------------
import shutil
try:
os.chdir(install_dir)
except Exception as e:
_log_error(install_dir, "切换到安装目录失败", e)
sys.exit(1)
# 若存在上次遗留的 .old,先静默删除
if os.path.isfile(old_exe_full):
try:
os.remove(old_exe_full)
except Exception as e:
_log_error(install_dir, "删除旧 .old 文件失败: %s" % old_exe_full, e)
# 将当前主程序重命名为 .old,腾出位置
if not os.path.isfile(target_exe_full):
_log_error(install_dir, "目标 exe 不存在: %s" % target_exe_full)
sys.exit(1)
try:
os.rename(target_exe_full, old_exe_full)
except Exception as e:
_log_error(install_dir, "重命名当前 exe 为 .old 失败", e)
sys.exit(1)
# 将新 exe 移入安装目录并命名为 target-exe-name
if not os.path.isfile(new_exe_path):
_log_error(install_dir, "新 exe 不存在: %s" % new_exe_path)
sys.exit(1)
try:
shutil.move(new_exe_path, target_exe_full)
except Exception as e:
_log_error(install_dir, "移动新 exe 到安装目录失败", e)
try:
os.rename(old_exe_full, target_exe_full)
except Exception:
pass
sys.exit(1)
# -------------------------------------------------------------------------
# 3. 干净的重生:Popen 新程序,强制 cwd、close_fds、DETACHED_PROCESS
# -------------------------------------------------------------------------
try:
creationflags = 0
if sys.platform == "win32":
# DETACHED_PROCESS = 0x00000008,不继承控制台与父进程句柄
creationflags = 0x00000008
p = subprocess.Popen(
[target_exe_full],
cwd=install_dir,
close_fds=True,
creationflags=creationflags,
shell=False,
)
# 不等待子进程,更新器功成身退
except Exception as e:
_log_error(install_dir, "拉起新主程序失败", e)
sys.exit(1)
# -------------------------------------------------------------------------
# 4. 功成身退
# -------------------------------------------------------------------------
sys.exit(0)
if __name__ == "__main__":
main()
+434
View File
@@ -0,0 +1,434 @@
# -*- coding: utf-8 -*-
"""
仿 Cursor / VS Code 的无感增量更新模块(Tkinter 入口用)
【打包前提】采用 PyInstaller 文件夹模式 (-D)。安装目录含 main.exe 与 _internal 等依赖。
【增量范围】.patch 仅针对 main.exe:在 TEMP 合成 main_new.exe 并校验 SHA256
替换时只覆盖安装目录下的 main.exe,绝不修改 _internal 或其它文件。
【临时清理】替换脚本会清理 TEMP 中的 .patch 与 main_new.exe,不留下垃圾。
依赖:bsdiff4, hashlib, requests, threading, os, sys, subprocess, tkinter。
远端 version.json 支持两种键名:delta 或 delta_updates(兼容旧配置)。
"""
import hashlib
import json
import os
import subprocess
import sys
import threading
import tkinter as tk
from pathlib import Path
from tkinter import messagebox
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
from urllib.parse import urlparse
try:
import requests
_HAS_REQUESTS = True
except ImportError:
_HAS_REQUESTS = False
try:
import bsdiff4
_HAS_BSDIFF = True
except ImportError:
_HAS_BSDIFF = False
# 远端 version.json 地址
VERSION_JSON_URL = "http://172.18.180.94:3000/api/file?path=server_manager/build/output/version.json"
REQUEST_TIMEOUT = 15
DOWNLOAD_TIMEOUT = 120
def clean_up_old_version():
"""
启动时调用:删除安装目录(或当前 exe 所在目录)下以 .old 结尾的残留文件,
即上一次「重命名大法」更新退下来的旧程序,静默删除不报错。
"""
if getattr(sys, "frozen", False):
install_dir = os.path.dirname(os.path.abspath(sys.executable))
else:
install_dir = os.path.dirname(os.path.abspath(__file__)) or os.getcwd()
try:
for name in os.listdir(install_dir) or []:
if name.endswith(".old"):
path = os.path.join(install_dir, name)
if os.path.isfile(path):
try:
os.remove(path)
except Exception:
pass
except Exception:
pass
def _write_update_log(install_dir: str, message: str) -> None:
"""向安装目录下的 logs/update_launcher.log 写入,便于确认是否走了 mini_updater。"""
try:
from datetime import datetime
log_dir = os.path.join(install_dir, "logs")
os.makedirs(log_dir, exist_ok=True)
log_path = os.path.join(log_dir, "update_launcher.log")
with open(log_path, "a", encoding="utf-8") as f:
f.write("[%s] %s\n" % (datetime.now().isoformat(), message))
except Exception:
pass
def apply_update_and_restart(
new_exe_path: str,
target_exe_name: str = None,
) -> None:
"""
独立进程外更新:拉起 mini_updater.exe,由其等待本进程退出后替换 exe 并以正确 cwd 重启。
若 mini_updater.exe 不存在则回退到“死循环重试”的 .bat。
调用后本进程会立即 os._exit(0),不会返回。
"""
new_exe_path = os.path.abspath(new_exe_path)
current_exe_path = os.path.abspath(sys.executable)
if not os.path.isfile(new_exe_path):
return
install_dir = os.path.normpath(os.path.dirname(current_exe_path))
target_name = (target_exe_name or os.path.basename(current_exe_path) or "main.exe").strip() or "main.exe"
# 优先使用安装目录下的 mini_updater.exe(与 main.exe 同目录),必须用绝对路径
mini_updater_exe = os.path.normpath(os.path.join(install_dir, "mini_updater.exe"))
if sys.platform == "win32" and os.path.isfile(mini_updater_exe):
try:
pid = os.getpid()
creationflags = 0x00000008 # DETACHED_PROCESS
subprocess.Popen(
[
mini_updater_exe,
"--pid", str(pid),
"--install-dir", install_dir,
"--new-exe-path", new_exe_path,
"--target-exe-name", target_name,
],
cwd=install_dir,
close_fds=True,
creationflags=creationflags,
shell=False,
)
_write_update_log(install_dir, "已拉起 mini_updater.exe (pid=%s)" % pid)
os._exit(0)
return
except Exception as e:
_write_update_log(install_dir, "拉起 mini_updater.exe 失败: %s,改用 .bat" % e)
else:
_write_update_log(
install_dir,
"mini_updater.exe 未找到 (路径: %s),使用 .bat 回退" % mini_updater_exe,
)
# 回退:.bat 死循环重试(仅 Windows)或 shell 脚本(Unix
temp_dir = os.environ.get("TEMP") or os.path.expandvars("%TEMP%")
bat_path = os.path.join(temp_dir, "apply_update_%s.bat" % os.getpid())
bat_lines = [
"@echo off",
":retry",
'move /y "%s" "%s" >nul 2>&1' % (new_exe_path, current_exe_path),
"if errorlevel 1 (",
" ping 127.0.0.1 -n 2 >nul",
" goto retry",
")",
'cd /d "%s"' % install_dir,
'start "" "%s"' % current_exe_path,
'del "%~f0"',
]
try:
with open(bat_path, "w", encoding="utf-8") as f:
f.write("\r\n".join(bat_lines))
if sys.platform == "win32":
subprocess.Popen(
["cmd", "/c", bat_path],
creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0,
shell=False,
)
else:
script = (
"while ! mv -f '%s' '%s' 2>/dev/null; do sleep 1; done; cd '%s' && exec '%s'"
% (new_exe_path, current_exe_path, install_dir, current_exe_path)
)
subprocess.Popen(["sh", "-c", script], shell=False)
except Exception:
pass
os._exit(0)
def _parse_version(v: str):
"""'x.y.z' -> (x, y, z)"""
try:
parts = v.strip().split(".")
return (
int(parts[0]) if len(parts) > 0 else 0,
int(parts[1]) if len(parts) > 1 else 0,
int(parts[2]) if len(parts) > 2 else 0,
)
except (ValueError, IndexError):
return (0, 0, 0)
def _version_less(a: str, b: str) -> bool:
"""True 表示 远端 b > 本地 a"""
return _parse_version(a) < _parse_version(b)
def _get_delta_map(manifest: dict) -> dict:
"""兼容 delta 与 delta_updates 两种键名"""
d = manifest.get("delta") or manifest.get("delta_updates") or {}
return d if isinstance(d, dict) else {}
class SeamlessUpdater:
"""
隐形增量更新:后台守护线程静默准备,就绪后主线程弹一次「重启以更新」,
用户确认后极速替换 main.exe(或静默安装完整包)并退出。
"""
def __init__(self, root: tk.Tk, local_version: str, version_url: str = None):
self.root = root
self.local_version = (local_version or "").strip()
self.version_url = (version_url or VERSION_JSON_URL).strip()
self._ready_delta_path = None # TEMP 下 main_new.exe 路径(增量就绪)
self._ready_full_path = None # TEMP 下 Setup.exe 路径(全量就绪)
self._ready_remote_version = None
self._lock = threading.Lock()
def _fetch_manifest(self) -> dict:
"""请求远端 version.json,失败返回空 dict"""
try:
if _HAS_REQUESTS:
r = requests.get(
self.version_url,
headers={"User-Agent": "ServerManager-Updater/1.0"},
timeout=REQUEST_TIMEOUT,
)
r.raise_for_status()
return r.json() if isinstance(r.json(), dict) else {}
req = Request(self.version_url, headers={"User-Agent": "ServerManager-Updater/1.0"})
with urlopen(req, timeout=REQUEST_TIMEOUT) as resp:
data = json.loads(resp.read().decode("utf-8"))
return data if isinstance(data, dict) else {}
except (URLError, HTTPError, OSError, json.JSONDecodeError, Exception):
return {}
def _download_to_path(self, url: str, dest_path: str) -> bool:
"""静默下载 url 到 dest_path,无 UI。"""
dest_path = Path(dest_path)
try:
dest_path.parent.mkdir(parents=True, exist_ok=True)
if _HAS_REQUESTS:
r = requests.get(
url,
headers={"User-Agent": "ServerManager-Updater/1.0"},
stream=True,
timeout=DOWNLOAD_TIMEOUT,
)
r.raise_for_status()
with open(dest_path, "wb") as f:
for chunk in r.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
else:
req = Request(url, headers={"User-Agent": "ServerManager-Updater/1.0"})
with urlopen(req, timeout=DOWNLOAD_TIMEOUT) as resp:
dest_path.write_bytes(resp.read())
return dest_path.is_file()
except Exception:
return False
def _try_delta_path(self, manifest: dict) -> bool:
"""
若存在当前版本的 delta:在 TEMP 静默下载 .patch -> bsdiff4 合成 main_new.exe -> 校验 SHA256。
仅针对 main.exe,不涉及 _internal。合成后删除 TEMP 中的 .patch。
成功则设置 self._ready_delta_path 并返回 True;任何一步失败返回 False。
"""
delta_map = _get_delta_map(manifest)
delta = delta_map.get(self.local_version)
if not isinstance(delta, dict):
return False
patch_url = (delta.get("patch_url") or "").strip()
new_exe_sha256 = (delta.get("new_exe_sha256") or "").strip().lower()
if not patch_url or not new_exe_sha256 or len(new_exe_sha256) != 64:
return False
if not _HAS_BSDIFF or not getattr(sys, "frozen", False):
return False
temp_dir = os.environ.get("TEMP") or os.path.expandvars("%TEMP%")
if not temp_dir or not os.path.isdir(temp_dir):
return False
patch_path = os.path.join(temp_dir, "ServerManager_update.patch")
new_exe_path = os.path.join(temp_dir, "ServerManager_new.exe")
if not self._download_to_path(patch_url, patch_path):
return False
if not os.path.isfile(patch_path):
return False
try:
with open(sys.executable, "rb") as f:
old_data = f.read()
with open(patch_path, "rb") as f:
patch_data = f.read()
new_data = bsdiff4.patch(old_data, patch_data)
got_sha = hashlib.sha256(new_data).hexdigest().lower()
if got_sha != new_exe_sha256:
return False
with open(new_exe_path, "wb") as f:
f.write(new_data)
with self._lock:
self._ready_delta_path = new_exe_path
return True
except Exception:
return False
finally:
try:
if os.path.isfile(patch_path):
os.remove(patch_path)
except Exception:
pass
def _try_full_path(self, manifest: dict) -> bool:
"""静默下载 full_installer_url 到 TEMP,成功则设置 _ready_full_path。"""
full_url = (
(manifest.get("full_installer_url") or manifest.get("download_url") or "").strip()
)
if not full_url:
return False
temp_dir = os.environ.get("TEMP") or os.path.expandvars("%TEMP%")
if not temp_dir or not os.path.isdir(temp_dir):
return False
name = os.path.basename(urlparse(full_url).path) or "ServerManager_Setup.exe"
local_path = os.path.join(temp_dir, name)
if not self._download_to_path(full_url, local_path):
return False
if not os.path.isfile(local_path):
return False
with self._lock:
self._ready_full_path = local_path
return True
def _worker(self):
"""后台线程:检测 -> 优先增量静默准备,失败则全量静默准备 -> 就绪后主线程弹窗"""
manifest = self._fetch_manifest()
if not manifest:
return
remote_version = (manifest.get("version") or "").strip()
if not remote_version or not _version_less(self.local_version, remote_version):
return
with self._lock:
self._ready_remote_version = remote_version
self._ready_delta_path = None
self._ready_full_path = None
# 优先增量:静默下载 + 合成 + 校验
if self._try_delta_path(manifest):
self.root.after(0, self._show_restart_dialog)
return
# 兜底:静默下载完整安装包
if self._try_full_path(manifest):
self.root.after(0, self._show_restart_dialog)
return
def _show_restart_dialog(self):
"""主线程:仅当有 _ready_delta_path 或 _ready_full_path 时弹出「退出并重启」"""
with self._lock:
remote = self._ready_remote_version or "?"
delta_path = self._ready_delta_path
full_path = self._ready_full_path
if not delta_path and not full_path:
return
msg = "新版本 v{} 已经准备就绪。点击「是」退出并重启以应用更新。".format(remote)
if not messagebox.askyesno("更新就绪", msg, default="yes"):
return
if delta_path and os.path.isfile(delta_path):
self._apply_delta_and_exit(delta_path)
elif full_path and os.path.isfile(full_path):
self._run_silent_install_and_exit(full_path)
def _apply_delta_and_exit(self, temp_new_exe_path: str):
"""增量路径:优先调用独立进程外更新器 mini_updater.exe,否则回退到 .bat 重试。"""
apply_update_and_restart(temp_new_exe_path, target_exe_name=os.path.basename(sys.executable))
def _run_silent_install_and_exit(self, setup_exe_path: str):
"""全量路径:静默参数启动 Setup.exe,主进程立即退出"""
try:
subprocess.Popen(
[
setup_exe_path,
"/VERYSILENT",
"/SUPPRESSMSGBOXES",
"/FORCECLOSEAPPLICATIONS",
],
shell=False,
)
except Exception:
pass
os._exit(0)
def start_background_worker(self):
"""启动后台守护线程(静默检测 + 静默准备),不阻塞主线程。"""
t = threading.Thread(target=self._worker, daemon=True)
t.start()
# ---------------------------------------------------------------------------
# entrypoints/tk_update_demo.py 调用示例(软件启动时调用)
# ---------------------------------------------------------------------------
"""
# 在 entrypoints/tk_update_demo.py 中(Tkinter 入口):
import tkinter as tk
from scripts.update.tk_updater import SeamlessUpdater, VERSION_JSON_URL
# 从 version.json 或常量读取本地版本号
LOCAL_VERSION = "1.1.3" # 需与 version.json / 安装包一致
def main():
root = tk.Tk()
root.withdraw() # 若主界面不是 Tk,可先隐藏,仅用其 messagebox/after
# 启动无感更新:后台静默准备,就绪后弹「重启以更新」
SeamlessUpdater(root, LOCAL_VERSION, VERSION_JSON_URL).start_background_worker()
# 你的主界面(Tk 或其它)
root.deiconify()
root.mainloop()
if __name__ == "__main__":
main()
"""
# ---------------------------------------------------------------------------
# 开发者:如何生成 .patch 文件(发版流程)
# ---------------------------------------------------------------------------
"""
【重要】采用 PyInstaller 文件夹模式 (-D)。增量仅针对 main.exe,不包含 _internal
用户端替换时只覆盖 main.exe,绝不修改 _internal 或其它文件。
1. 打包:使用 PyInstaller 文件夹模式生成当前版本:
pyinstaller -D -w your_main.py
得到 dist/your_main/main.exe(或 dist/main.exe)。
2. 保存上一版本的 main.exe(仅此单文件,无需 _internal):
将上一发版时的 main.exe 保存为 build/output/main_1.0.2.exe(版本号与 version.json 中 delta 的键一致)。
3. 生成差分补丁(需安装 bsdiff4pip install bsdiff4):
import bsdiff4
old = open('build/output/main_1.0.2.exe', 'rb').read()
new = open('dist/your_main/main.exe', 'rb').read()
patch = bsdiff4.diff(old, new)
open('build/output/patches/v1.0.2_to_v1.0.3.patch', 'wb').write(patch)
4. 计算新 exe 的 SHA256 并写入 version.json
import hashlib
new_sha256 = hashlib.sha256(new).hexdigest().lower()
# 将 new_sha256 填入 version.json 的 delta["1.0.2"].new_exe_sha256
5. 上传 build/output/patches/*.patch 与 version.json 到服务器;用户端即可通过增量路径静默更新。
"""