Commit 3a8f8760 by 周田

实现可复用更新器与统一依赖配置

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
parent f38a1ccd
......@@ -2,9 +2,11 @@
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
# Local distribution artifacts and database
distribution-server/data/
[project]
name = "program-updater"
version = "0.1.0"
description = "Add your description here"
readme = "README.md"
version = "261028.0.0"
description = "Lightweight updater for Windows PyInstaller applications"
readme = "updater/README.md"
authors = [
{ name = "ZutJoe", email = "zhout50@163.com" }
]
requires-python = ">=3.12"
dependencies = []
[project.optional-dependencies]
server = [
"fastapi>=0.115.0,<1.0.0",
"python-multipart>=0.0.20",
"uvicorn>=0.34.0,<1.0.0",
]
test = [
"httpx2>=0.1.0",
]
[project.scripts]
program-updater = "program_updater:main"
program-updater = "updater:main"
program-updater-helper = "updater.platforms.windows.helper:main"
[build-system]
requires = ["uv_build>=0.12.13,<0.13.0"]
build-backend = "uv_build"
[tool.uv.build-backend]
module-name = "updater"
module-root = "."
def main() -> None:
print("Hello from program-updater!")
# program-updater
面向 Windows、Python 3.12 和 PyInstaller `--onefile` 应用的更新客户端。基础安装无第三方运行时依赖。局域网分发管理后台是同一仓库中的独立项目,位于 `../distribution-server/`。
## 包结构
```text
updater/
├─ __init__.py # 稳定的公共导入入口
├─ core/
│ ├─ client.py # 检查、下载、校验和应用更新流程
│ ├─ manifest.py # manifest 校验
│ ├─ models.py # 更新信息模型
│ ├─ errors.py # 更新异常
│ ├─ files.py # 文件哈希等共享工具
│ └─ health.py # 新版本启动确认
└─ platforms/
└─ windows/
├─ helper.py # Windows exe 替换和回滚
└─ helper_entry.py
```
应用使用的公开入口是 `from updater import UpdateClient, confirm_update_started`;平台相关实现与通用更新逻辑分离。包名由 `program_updater` 改为 `updater`,集成旧包名的应用需要同步更新 import。
## 接入步骤
以下示例假设目标应用在 `self-strategy/`,updater 仓库在同一父目录下,
目标环境使用 Windows、Python 3.12 和 PyInstaller。
### 1. 构建并安装 wheel
在 updater 仓库根目录构建 wheel:
```powershell
cd C:\path\to\program-updater
uv build --wheel
```
将 wheel 安装到**目标应用用于 PyInstaller 的同一个 Python 环境**。例如在
`self-strategy/` 根目录执行:
```powershell
cd C:\path\to\self-strategy
python -m pip install ..\program-updater\dist\program_updater-261028.0.0-py3-none-any.whl
python -c "from updater import UpdateClient, confirm_update_started; print('updater installed')"
```
wheel 的发行包名是 `program-updater`,Python 导入名是 `updater`。基础安装没有
FastAPI 等 Web 依赖。
### 2. 构建 Windows helper
在同一个 Python 3.12 环境中确保已安装 PyInstaller(它是构建工具,不是 updater
运行时依赖),然后从 `self-strategy/` 根目录构建 helper:
```powershell
python -m PyInstaller --onefile --paths ..\program-updater --name UpdaterHelper ..\program-updater\updater\platforms\windows\helper_entry.py
```
构建结果为 `self-strategy\dist\UpdaterHelper.exe`。helper 是一个独立的小型 exe,
负责等待旧程序退出、备份并替换主 exe、启动新版以及在启动失败时回滚。
### 3. 将 helper 加入目标程序的 PyInstaller 配置
在目标程序的 `.spec` 中定义 helper 二进制及其在 onefile 包内的目标目录:
```python
UPDATER_HELPER = os.path.join(ROOT, "dist", "UpdaterHelper.exe")
updater_binaries = [
(UPDATER_HELPER, "updater/platforms/windows/bin"),
]
```
把 `updater_binaries` 传给 `Analysis`:
```python
a = Analysis(
...,
pathex=[ROOT],
binaries=updater_binaries,
...
)
```
如果使用 PyInstaller 命令行而非 `.spec`,对应写法是:
```powershell
pyinstaller --onefile --add-binary "dist\UpdaterHelper.exe;updater\platforms\windows\bin" --name MyProgram path\to\your_app.py
```
目标程序静态导入 `updater` 后,PyInstaller 会自动收集所需的 Python 模块;helper
exe 则必须按上述方式额外嵌入。
### 4. 在应用启动时检查更新
分发端为每个程序使用独立的 `app_id`,并提供如下清单:
```json
{
"app_id": "margin-guard",
"version": "1.2.0",
"url": "http://192.168.1.20:8000/packages/margin-guard/1.2.0/app.exe",
"size": 12345678,
"sha256": "填入 exe 的 64 位 SHA-256"
}
```
目标程序在启动早期、创建主窗口和初始化业务逻辑之前调用更新器。下面是 PyQt
程序的示例:
```python
import logging
import sys
from pathlib import Path
from PyQt5.QtWidgets import QApplication, QMessageBox
from updater import UpdateClient, UpdaterError, confirm_update_started
APP_ID = "margin-guard"
APP_VERSION = "1.1.0"
MANIFEST_URL = "http://192.168.1.20:8000/updates/margin-guard/latest.json"
logger = logging.getLogger(__name__)
def main() -> int:
app = QApplication(sys.argv)
if getattr(sys, "frozen", False):
bundle_dir = Path(getattr(sys, "_MEIPASS", Path(sys.executable).parent))
updater = UpdateClient(
app_id=APP_ID,
current_version=APP_VERSION,
manifest_url=MANIFEST_URL,
executable_path=sys.executable,
helper_resource=(
bundle_dir
/ "updater"
/ "platforms"
/ "windows"
/ "bin"
/ "UpdaterHelper.exe"
),
)
def confirm_update(update) -> bool:
answer = QMessageBox.question(
None,
"发现新版本",
f"发现版本 {update.version},现在更新吗?选择“否”可稍后更新。",
QMessageBox.Yes | QMessageBox.No,
QMessageBox.Yes,
)
return answer == QMessageBox.Yes
try:
handed_off = updater.update_before_startup(
confirm_update=confirm_update,
restart_args=sys.argv[1:],
)
except UpdaterError:
logger.exception("自动更新失败,将启动当前版本")
handed_off = False
if handed_off:
return 0
# 放入目标应用原有的窗口、host 和业务初始化代码。
window = create_main_window()
window.show()
# 在程序完成关键初始化后确认更新成功;helper 收到此信号后不再回滚。
confirm_update_started()
return app.exec_()
if __name__ == "__main__":
raise SystemExit(main())
```
`update_before_startup()` 会同步检查清单。选择“否”时本次继续启动旧版,下次启动会再次提醒;选择“是”时才下载和校验 exe、启动 helper,并返回 `True`。目标进程随后退出,helper 替换 exe 并自动启动新版。新版启动检查时发现已是最新版本,就会继续启动主窗口。LAN 服务不可用或更新失败时,上例记录错误并继续启动旧版;强制更新的程序可以改为提示错误并停止启动。
`confirm_update()` 是目标应用提供的回调,所以 updater 本身不依赖 PyQt。`confirm_update_started()` 应在新版完成必要初始化后调用。helper 默认等待确认 45 秒;若新版提前退出或没有确认,会尝试恢复旧 exe。数据库迁移前应备份数据,exe 回滚本身不能撤销数据库迁移。portable 安装目录必须允许当前用户写入,配置文件和数据库应放在 exe 之外。
## 限制
- helper 只支持 Windows;分发管理后台可在 Linux 上运行。
- 更新包是完整 exe,不做差分更新。
- 没有发布者签名验证;分发服务器必须可信。没有 Authenticode 证书时,Windows 可能提示未知发布者。
"""Public client-side updater API."""
from .core.client import UpdateClient
from .core.errors import (
DownloadError,
ManifestError,
UpdateApplyError,
UpdateCheckError,
UpdaterError,
)
from .core.health import confirm_update_started
from .core.models import UpdateInfo
__all__ = [
"DownloadError",
"ManifestError",
"UpdateApplyError",
"UpdateCheckError",
"UpdateClient",
"UpdateInfo",
"UpdaterError",
"confirm_update_started",
]
def main() -> int:
from .platforms.windows.helper import main as helper_main
return helper_main()
"""Platform-independent updater core."""
"""Exception types raised by the updater core."""
class UpdaterError(Exception):
"""Base exception for updater failures."""
class ManifestError(UpdaterError):
"""The update manifest is invalid or does not match this application."""
class UpdateCheckError(UpdaterError):
"""The update manifest could not be retrieved."""
class DownloadError(UpdaterError):
"""The update artifact could not be downloaded or verified."""
class UpdateApplyError(UpdaterError):
"""The update helper could not be launched."""
"""File utilities shared by updater workflows."""
import hashlib
from pathlib import Path
_CHUNK_SIZE = 64 * 1024
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as file:
while chunk := file.read(_CHUNK_SIZE):
digest.update(chunk)
return digest.hexdigest()
"""Startup confirmation shared with the update helper's rollback check."""
import os
import sys
from pathlib import Path
from uuid import uuid4
from .errors import UpdateApplyError
_HEALTH_FILE_ENV = "PROGRAM_UPDATER_HEALTH_FILE"
_HEALTH_TOKEN_ENV = "PROGRAM_UPDATER_HEALTH_TOKEN"
def confirm_update_started() -> bool:
marker_value = os.environ.get(_HEALTH_FILE_ENV)
token = os.environ.get(_HEALTH_TOKEN_ENV)
if marker_value is None and token is None:
return False
if marker_value is None or token is None:
raise UpdateApplyError("incomplete update health-check environment.")
if len(token) != 32 or any(character not in "0123456789abcdef" for character in token):
raise UpdateApplyError("invalid update health-check token.")
marker_path = Path(marker_value).resolve()
install_root = Path(sys.executable).resolve().parent
health_root = (install_root / ".updates" / "health").resolve()
if not marker_path.is_relative_to(health_root):
raise UpdateApplyError("update health-check marker is outside the update directory.")
marker_path.parent.mkdir(parents=True, exist_ok=True)
temporary_path = marker_path.with_name(f".{marker_path.name}.{uuid4().hex}.tmp")
try:
temporary_path.write_text(token, encoding="ascii")
os.replace(temporary_path, marker_path)
finally:
temporary_path.unlink(missing_ok=True)
return True
"""Validation for manifests shared with the distribution service."""
import json
import re
from urllib.parse import urlsplit
from .errors import ManifestError
from .models import UpdateInfo
_APP_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
_VERSION_PATTERN = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)$")
_SHA256_PATTERN = re.compile(r"^[A-Fa-f0-9]{64}$")
def validate_app_id(app_id: str) -> str:
if not isinstance(app_id, str) or not _APP_ID_PATTERN.fullmatch(app_id):
raise ManifestError("app_id must contain only letters, digits, '.', '_' or '-'.")
return app_id
def version_key(version: str) -> tuple[int, int, int]:
if not isinstance(version, str):
raise ManifestError("version must be a string in MAJOR.MINOR.PATCH format.")
match = _VERSION_PATTERN.fullmatch(version)
if match is None:
raise ManifestError(
f"unsupported version {version!r}; use MAJOR.MINOR.PATCH, for example 1.2.3."
)
return int(match.group(1)), int(match.group(2)), int(match.group(3))
def validate_url(url: str, *, field_name: str = "url") -> str:
if not isinstance(url, str):
raise ManifestError(f"{field_name} must be a string.")
parsed = urlsplit(url)
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
raise ManifestError(f"{field_name} must be an absolute HTTP or HTTPS URL.")
if parsed.username is not None or parsed.password is not None:
raise ManifestError(f"{field_name} must not contain embedded credentials.")
return url
def parse_manifest(raw: bytes, *, expected_app_id: str) -> UpdateInfo:
try:
manifest = json.loads(raw.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ManifestError("manifest is not valid UTF-8 JSON.") from exc
if not isinstance(manifest, dict):
raise ManifestError("manifest must be a JSON object.")
app_id = validate_app_id(manifest.get("app_id"))
if app_id != expected_app_id:
raise ManifestError(
f"manifest app_id {app_id!r} does not match {expected_app_id!r}."
)
version = manifest.get("version")
version_key(version)
url = validate_url(manifest.get("url"))
size = manifest.get("size")
if type(size) is not int or size <= 0:
raise ManifestError("size must be a positive integer.")
sha256 = manifest.get("sha256")
if not isinstance(sha256, str) or not _SHA256_PATTERN.fullmatch(sha256):
raise ManifestError("sha256 must contain exactly 64 hexadecimal characters.")
return UpdateInfo(
app_id=app_id,
version=version,
url=url,
size=size,
sha256=sha256.lower(),
)
"""Data models used by the updater core."""
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class UpdateInfo:
app_id: str
version: str
url: str
size: int
sha256: str
"""Platform-specific updater implementations."""
"""Windows-specific updater implementation."""
"""PyInstaller entry point for the standalone Windows update helper."""
from updater.platforms.windows.helper import main
if __name__ == "__main__":
raise SystemExit(main())
"""Tests for the client-side updater."""
import hashlib
import io
import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from updater import UpdateClient, UpdateInfo, confirm_update_started
from updater.core.errors import DownloadError, UpdateCheckError
class FakeResponse(io.BytesIO):
def __enter__(self) -> "FakeResponse":
return self
def __exit__(self, *_: object) -> None:
self.close()
class UpdateClientTests(unittest.TestCase):
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.root = Path(self.temp_dir.name)
self.executable = self.root / "MyProgram.exe"
self.executable.write_bytes(b"old executable")
self.helper = self.root / "UpdaterHelper.exe"
self.helper.write_bytes(b"helper")
def tearDown(self) -> None:
self.temp_dir.cleanup()
def client(self, *, current_version: str = "1.0.0") -> UpdateClient:
return UpdateClient(
app_id="my-program",
current_version=current_version,
manifest_url="http://127.0.0.1:8000/latest.json",
executable_path=self.executable,
helper_resource=self.helper,
)
def manifest(self, artifact: bytes, *, version: str = "1.1.0") -> bytes:
return json.dumps(
{
"app_id": "my-program",
"version": version,
"url": "http://127.0.0.1:8000/my-program.exe",
"size": len(artifact),
"sha256": hashlib.sha256(artifact).hexdigest(),
}
).encode("utf-8")
def test_check_and_download_update(self) -> None:
artifact = b"new executable"
responses = [
FakeResponse(self.manifest(artifact)),
FakeResponse(artifact),
]
progress: list[tuple[int, int]] = []
with patch(
"updater.core.client.urllib.request.urlopen",
side_effect=responses,
):
update = self.client().check_for_update()
self.assertIsNotNone(update)
assert update is not None
staged = self.client().download(
update,
on_progress=lambda downloaded, total: progress.append((downloaded, total)),
)
self.assertEqual(staged.read_bytes(), artifact)
self.assertEqual(progress[-1], (len(artifact), len(artifact)))
def test_check_returns_none_for_current_version(self) -> None:
artifact = b"same version"
with patch(
"updater.core.client.urllib.request.urlopen",
return_value=FakeResponse(self.manifest(artifact, version="1.0.0")),
):
self.assertIsNone(self.client().check_for_update())
def test_update_before_startup_returns_false_when_current(self) -> None:
client = self.client()
with (
patch.object(client, "check_for_update", return_value=None),
patch.object(client, "download") as download,
patch.object(client, "launch_update_helper") as launch_helper,
):
self.assertFalse(client.update_before_startup())
download.assert_not_called()
launch_helper.assert_not_called()
def test_update_before_startup_returns_false_when_user_defers(self) -> None:
client = self.client()
update = UpdateInfo(
app_id="my-program",
version="1.1.0",
url="http://127.0.0.1:8000/my-program.exe",
size=1,
sha256=hashlib.sha256(b"x").hexdigest(),
)
confirm_update = Mock(return_value=False)
with (
patch.object(client, "check_for_update", return_value=update),
patch.object(client, "download") as download,
patch.object(client, "launch_update_helper") as launch_helper,
):
self.assertFalse(
client.update_before_startup(confirm_update=confirm_update)
)
confirm_update.assert_called_once_with(update)
download.assert_not_called()
launch_helper.assert_not_called()
def test_update_before_startup_hands_off_to_helper(self) -> None:
client = self.client()
update = UpdateInfo(
app_id="my-program",
version="1.1.0",
url="http://127.0.0.1:8000/my-program.exe",
size=1,
sha256=hashlib.sha256(b"x").hexdigest(),
)
staged_executable = self.root / ".updates" / "staging" / "update.exe"
progress = Mock()
confirm_update = Mock(return_value=True)
restart_args = ("--safe-mode",)
with (
patch.object(client, "check_for_update", return_value=update),
patch.object(client, "download", return_value=staged_executable) as download,
patch.object(client, "launch_update_helper") as launch_helper,
):
self.assertTrue(
client.update_before_startup(
confirm_update=confirm_update,
restart_args=restart_args,
on_progress=progress,
health_timeout=60.0,
)
)
confirm_update.assert_called_once_with(update)
download.assert_called_once_with(update, on_progress=progress)
launch_helper.assert_called_once_with(
update,
staged_executable,
restart_args=restart_args,
health_timeout=60.0,
)
def test_update_before_startup_surfaces_check_errors(self) -> None:
client = self.client()
with patch.object(
client,
"check_for_update",
side_effect=UpdateCheckError("server unavailable"),
):
with self.assertRaisesRegex(UpdateCheckError, "server unavailable"):
client.update_before_startup()
def test_download_rejects_hash_mismatch_and_removes_partial_file(self) -> None:
expected = b"valid"
downloaded = b"wrong"
with patch(
"updater.core.client.urllib.request.urlopen",
return_value=FakeResponse(downloaded),
):
with self.assertRaises(DownloadError):
self.client().download(
UpdateInfo(
app_id="my-program",
version="1.1.0",
url="http://127.0.0.1:8000/my-program.exe",
size=len(expected),
sha256=hashlib.sha256(expected).hexdigest(),
)
)
staging_dir = self.root / ".updates" / "staging"
self.assertEqual(list(staging_dir.iterdir()), [])
def test_health_confirmation_is_noop_outside_an_update_restart(self) -> None:
with patch.dict(os.environ, {}, clear=True):
self.assertFalse(confirm_update_started())
def test_health_confirmation_writes_expected_token(self) -> None:
token = "a" * 32
marker = self.root / ".updates" / "health" / f"{token}.ready"
with (
patch("sys.executable", str(self.executable)),
patch.dict(
os.environ,
{
"PROGRAM_UPDATER_HEALTH_FILE": str(marker),
"PROGRAM_UPDATER_HEALTH_TOKEN": token,
},
clear=True,
),
):
self.assertTrue(confirm_update_started())
self.assertEqual(marker.read_text(encoding="ascii"), token)
if __name__ == "__main__":
unittest.main()
"""Tests for the updater manifest contract."""
import hashlib
import json
import unittest
from updater.core.errors import ManifestError
from updater.core.manifest import parse_manifest, version_key
class ManifestTests(unittest.TestCase):
def test_parse_valid_manifest_normalizes_sha256(self) -> None:
manifest = {
"app_id": "my-program",
"version": "1.2.3",
"url": "http://192.168.1.20/releases/app.exe",
"size": 42,
"sha256": "A" * 64,
}
update = parse_manifest(
json.dumps(manifest).encode("utf-8"),
expected_app_id="my-program",
)
self.assertEqual(update.app_id, "my-program")
self.assertEqual(update.version, "1.2.3")
self.assertEqual(update.sha256, "a" * 64)
self.assertEqual(update.size, 42)
def test_rejects_manifest_for_another_application(self) -> None:
manifest = {
"app_id": "other-app",
"version": "1.2.3",
"url": "http://192.168.1.20/app.exe",
"size": 42,
"sha256": hashlib.sha256(b"app").hexdigest(),
}
with self.assertRaises(ManifestError):
parse_manifest(
json.dumps(manifest).encode("utf-8"),
expected_app_id="my-program",
)
def test_rejects_non_http_urls(self) -> None:
manifest = {
"app_id": "my-program",
"version": "1.2.3",
"url": "file:///tmp/app.exe",
"size": 42,
"sha256": hashlib.sha256(b"app").hexdigest(),
}
with self.assertRaises(ManifestError):
parse_manifest(
json.dumps(manifest).encode("utf-8"),
expected_app_id="my-program",
)
def test_rejects_boolean_size(self) -> None:
manifest = {
"app_id": "my-program",
"version": "1.2.3",
"url": "http://192.168.1.20/app.exe",
"size": True,
"sha256": hashlib.sha256(b"app").hexdigest(),
}
with self.assertRaises(ManifestError):
parse_manifest(
json.dumps(manifest).encode("utf-8"),
expected_app_id="my-program",
)
def test_versions_compare_as_numeric_components(self) -> None:
self.assertGreater(version_key("1.10.0"), version_key("1.9.9"))
def test_rejects_prerelease_version_syntax(self) -> None:
with self.assertRaises(ManifestError):
version_key("1.2.3-rc.1")
if __name__ == "__main__":
unittest.main()
This diff is collapsed. Click to expand it.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment