261 lines
8.4 KiB
Python
261 lines
8.4 KiB
Python
"""Startup update-check helpers.
|
|
|
|
The first release only decides whether the app may enter the main window.
|
|
It never downloads, overwrites, deletes, or migrates local program/data files.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import urllib.request
|
|
from dataclasses import dataclass
|
|
|
|
from . import client_policy
|
|
from .version import APP_CODE_NAME, APP_UPDATE_CHECK_URL, APP_VERSION
|
|
|
|
DEFAULT_TIMEOUT_SECONDS = 3.0
|
|
|
|
|
|
class UpdateCheckError(RuntimeError):
|
|
"""Raised for invalid update-check inputs or server responses."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UpdateInfo:
|
|
latest_version: str = ""
|
|
min_supported_version: str = ""
|
|
force_update: bool = False
|
|
download_url: str = ""
|
|
sha256: str = ""
|
|
size_bytes: int = 0
|
|
package_format: str = ""
|
|
updater_protocol: int = 0
|
|
min_updater_protocol: int = 0
|
|
signature_algorithm: str = ""
|
|
manifest_signature: str = ""
|
|
message: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UpdateCheckResult:
|
|
current_version: str
|
|
checked: bool = False
|
|
forced: bool = False
|
|
latest_version: str = ""
|
|
min_supported_version: str = ""
|
|
download_url: str = ""
|
|
sha256: str = ""
|
|
size_bytes: int = 0
|
|
package_format: str = ""
|
|
updater_protocol: int = 0
|
|
min_updater_protocol: int = 0
|
|
signature_algorithm: str = ""
|
|
manifest_signature: str = ""
|
|
automatic_update_error: str = ""
|
|
message: str = ""
|
|
error: str = ""
|
|
client_policy: client_policy.ClientPolicy | None = None
|
|
client_policy_error: str = ""
|
|
|
|
@property
|
|
def can_enter(self) -> bool:
|
|
return not self.forced
|
|
|
|
|
|
def parse_version(version) -> tuple[int, ...]:
|
|
"""Parse semantic numeric version segments for comparison."""
|
|
|
|
text = str(version or "").strip()
|
|
if text.lower().startswith("v"):
|
|
text = text[1:].strip()
|
|
if not text:
|
|
raise UpdateCheckError("版本号不能为空")
|
|
|
|
segments = []
|
|
for part in text.split("."):
|
|
match = re.match(r"^(\d+)", part.strip())
|
|
if match is None:
|
|
raise UpdateCheckError(f"版本号格式不正确：{version}")
|
|
segments.append(int(match.group(1)))
|
|
return tuple(segments)
|
|
|
|
|
|
def compare_versions(left, right) -> int:
|
|
"""Return -1/0/1 for numeric semantic version comparison."""
|
|
|
|
left_segments = parse_version(left)
|
|
right_segments = parse_version(right)
|
|
length = max(len(left_segments), len(right_segments))
|
|
left_padded = left_segments + (0,) * (length - len(left_segments))
|
|
right_padded = right_segments + (0,) * (length - len(right_segments))
|
|
if left_padded < right_padded:
|
|
return -1
|
|
if left_padded > right_padded:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _as_bool(value) -> bool:
|
|
if isinstance(value, bool):
|
|
return value
|
|
if isinstance(value, (int, float)):
|
|
return value != 0
|
|
if isinstance(value, str):
|
|
return value.strip().lower() in {"1", "true", "yes", "y", "on"}
|
|
return False
|
|
|
|
|
|
def parse_update_info(payload) -> UpdateInfo:
|
|
if not isinstance(payload, dict):
|
|
raise UpdateCheckError("版本接口返回内容不是 JSON 对象")
|
|
|
|
release = payload.get("release")
|
|
if not isinstance(release, dict):
|
|
release = {}
|
|
|
|
latest_version = str(payload.get("latest_version") or release.get("version") or "").strip()
|
|
min_supported_version = str(
|
|
payload.get("min_supported_version") or release.get("min_supported_version") or ""
|
|
).strip()
|
|
if not latest_version and not min_supported_version:
|
|
raise UpdateCheckError("版本接口缺少 latest_version 或 min_supported_version")
|
|
|
|
return UpdateInfo(
|
|
latest_version=latest_version,
|
|
min_supported_version=min_supported_version,
|
|
force_update=_as_bool(payload.get("force_update", release.get("force_update"))),
|
|
download_url=str(payload.get("download_url") or release.get("download_url") or "").strip(),
|
|
sha256=str(payload.get("sha256") or release.get("sha256") or "").strip(),
|
|
size_bytes=int(payload.get("size_bytes") or release.get("size_bytes") or 0),
|
|
package_format=str(
|
|
payload.get("package_format") or release.get("package_format") or ""
|
|
).strip(),
|
|
updater_protocol=int(
|
|
payload.get("updater_protocol") or release.get("updater_protocol") or 0
|
|
),
|
|
min_updater_protocol=int(
|
|
payload.get("min_updater_protocol")
|
|
or release.get("min_updater_protocol")
|
|
or 0
|
|
),
|
|
signature_algorithm=str(
|
|
payload.get("signature_algorithm")
|
|
or release.get("signature_algorithm")
|
|
or ""
|
|
).strip(),
|
|
manifest_signature=str(
|
|
payload.get("manifest_signature")
|
|
or release.get("manifest_signature")
|
|
or ""
|
|
).strip(),
|
|
message=str(
|
|
payload.get("message")
|
|
or payload.get("release_notes")
|
|
or release.get("message")
|
|
or release.get("release_notes")
|
|
or ""
|
|
).strip(),
|
|
)
|
|
|
|
|
|
def is_forced_update(info: UpdateInfo, current_version: str) -> bool:
|
|
if info.min_supported_version and compare_versions(current_version, info.min_supported_version) < 0:
|
|
return True
|
|
if (
|
|
info.force_update
|
|
and info.latest_version
|
|
and compare_versions(current_version, info.latest_version) < 0
|
|
):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _decode_payload(raw_payload):
|
|
if isinstance(raw_payload, (bytes, bytearray)):
|
|
raw_payload = raw_payload.decode("utf-8")
|
|
if isinstance(raw_payload, str):
|
|
try:
|
|
return json.loads(raw_payload)
|
|
except ValueError as exc:
|
|
raise UpdateCheckError("版本接口返回内容不是合法 JSON") from exc
|
|
return raw_payload
|
|
|
|
|
|
def fetch_update_payload(url: str, timeout: float = DEFAULT_TIMEOUT_SECONDS):
|
|
request = urllib.request.Request(
|
|
url,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"User-Agent": f"{APP_CODE_NAME}/{APP_VERSION}",
|
|
},
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
return _decode_payload(response.read())
|
|
|
|
|
|
def check_for_update(
|
|
*,
|
|
current_version: str = APP_VERSION,
|
|
url: str = APP_UPDATE_CHECK_URL,
|
|
timeout: float = DEFAULT_TIMEOUT_SECONDS,
|
|
fetcher=None,
|
|
) -> UpdateCheckResult:
|
|
if not str(url or "").strip():
|
|
return UpdateCheckResult(current_version=current_version, checked=False)
|
|
|
|
try:
|
|
payload = _decode_payload((fetcher or fetch_update_payload)(url, timeout))
|
|
except Exception as exc:
|
|
return UpdateCheckResult(
|
|
current_version=current_version,
|
|
checked=True,
|
|
forced=False,
|
|
error=f"启动版本检查失败，已允许继续使用：{exc}",
|
|
)
|
|
|
|
resolved_policy = None
|
|
policy_error = ""
|
|
if isinstance(payload, dict) and "client_policy" in payload:
|
|
try:
|
|
resolved_policy = client_policy.parse_client_policy(
|
|
payload.get("client_policy")
|
|
)
|
|
policy_error = resolved_policy.warning
|
|
except client_policy.ClientPolicyError as exc:
|
|
policy_error = str(exc)
|
|
|
|
try:
|
|
if isinstance(payload, dict) and payload.get("release", object()) is None:
|
|
info = UpdateInfo()
|
|
else:
|
|
info = parse_update_info(payload)
|
|
forced = is_forced_update(info, current_version)
|
|
return UpdateCheckResult(
|
|
current_version=current_version,
|
|
checked=True,
|
|
forced=forced,
|
|
latest_version=info.latest_version,
|
|
min_supported_version=info.min_supported_version,
|
|
download_url=info.download_url,
|
|
sha256=info.sha256,
|
|
size_bytes=info.size_bytes,
|
|
package_format=info.package_format,
|
|
updater_protocol=info.updater_protocol,
|
|
min_updater_protocol=info.min_updater_protocol,
|
|
signature_algorithm=info.signature_algorithm,
|
|
manifest_signature=info.manifest_signature,
|
|
message=info.message,
|
|
client_policy=resolved_policy,
|
|
client_policy_error=policy_error,
|
|
)
|
|
except Exception as exc:
|
|
return UpdateCheckResult(
|
|
current_version=current_version,
|
|
checked=True,
|
|
forced=False,
|
|
error=f"启动版本检查失败，已允许继续使用：{exc}",
|
|
client_policy=resolved_policy,
|
|
client_policy_error=policy_error,
|
|
)
|