Danil Iarosh
35c9693740
- Исправлен перезапуск приложения после замены файлов и добавлено подтверждение запуска новой версии. - Обеспечены откат при ошибке и корректное завершение фоновых задач. - Исключены посторонние DLL из Windows-сборки, обновлены тесты и инструкция ручного перехода с v2.0.1.
649 lines
25 KiB
Python
649 lines
25 KiB
Python
"""Проверка, загрузка и установка обновлений Plate Config App."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import platform
|
||
import re
|
||
import shutil
|
||
import stat
|
||
import subprocess
|
||
import sys
|
||
import uuid
|
||
import zipfile
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass
|
||
from pathlib import Path, PurePosixPath
|
||
from typing import BinaryIO
|
||
from urllib.parse import quote, unquote, urlparse
|
||
from urllib.request import Request, urlopen
|
||
|
||
LATEST_RELEASE_URL = (
|
||
"https://git.robotail.pro/Robotail/plate-config-app/releases/latest"
|
||
)
|
||
RELEASE_DOWNLOAD_BASE_URL = (
|
||
"https://git.robotail.pro/Robotail/plate-config-app/releases/download"
|
||
)
|
||
UPDATE_HOST = "git.robotail.pro"
|
||
HTTP_TIMEOUT_SECONDS = 10.0
|
||
DOWNLOAD_TIMEOUT_SECONDS = 30.0
|
||
DOWNLOAD_CHUNK_SIZE = 1024 * 1024
|
||
MAX_DOWNLOAD_BYTES = 300 * 1024 * 1024
|
||
MAX_UNPACKED_BYTES = 600 * 1024 * 1024
|
||
MAX_ARCHIVE_MEMBERS = 10_000
|
||
UPDATE_READY_ENV = "PLATE_CONFIG_UPDATE_READY_FILE"
|
||
VERSION_PATTERN = re.compile(
|
||
r"[vV]?(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)"
|
||
)
|
||
TOP_LEVEL_NAME_PATTERN = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]*")
|
||
|
||
|
||
class UpdateError(RuntimeError):
|
||
"""Ошибка проверки, загрузки или запуска обновления."""
|
||
|
||
|
||
class UpdateCancelled(UpdateError):
|
||
"""Загрузка обновления отменена пользователем или закрытием приложения."""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class UpdateInfo:
|
||
tag: str
|
||
version: str
|
||
asset_name: str
|
||
download_url: str
|
||
|
||
|
||
def normalize_version(value: str) -> str:
|
||
normalized = value.strip()
|
||
if normalized[:1].casefold() == "v":
|
||
normalized = normalized[1:]
|
||
return normalized
|
||
|
||
|
||
def parse_version(value: str) -> tuple[int, int, int]:
|
||
"""Преобразовать версию X.Y.Z или vX.Y.Z в числовой кортеж."""
|
||
match = VERSION_PATTERN.fullmatch(value.strip())
|
||
if match is None:
|
||
raise UpdateError(
|
||
f"Некорректная версия {value!r}: ожидается формат vX.Y.Z"
|
||
)
|
||
return tuple(int(component) for component in match.groups())
|
||
|
||
|
||
def platform_executable_name(
|
||
*,
|
||
system: str | None = None,
|
||
machine: str | None = None,
|
||
) -> str:
|
||
system_name = (system or platform.system()).casefold()
|
||
machine_name = (machine or platform.machine()).casefold()
|
||
if machine_name not in {"amd64", "x86_64"}:
|
||
raise UpdateError(f"Автообновление не поддерживает архитектуру {machine_name}")
|
||
if system_name == "windows":
|
||
return "plate_config_app-windows-x86_64.exe"
|
||
if system_name == "linux":
|
||
return "plate_config_app-linux-x86_64"
|
||
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
|
||
|
||
|
||
def platform_asset_name(
|
||
*,
|
||
system: str | None = None,
|
||
machine: str | None = None,
|
||
) -> str:
|
||
executable_name = platform_executable_name(system=system, machine=machine)
|
||
if executable_name.casefold().endswith(".exe"):
|
||
executable_name = executable_name[:-4]
|
||
return f"{executable_name}.zip"
|
||
|
||
|
||
def archive_executable_name(asset_name: str) -> str:
|
||
names = {
|
||
"plate_config_app-windows-x86_64.zip": (
|
||
"plate_config_app-windows-x86_64.exe"
|
||
),
|
||
"plate_config_app-linux-x86_64.zip": "plate_config_app-linux-x86_64",
|
||
}
|
||
try:
|
||
return names[asset_name]
|
||
except KeyError as exc:
|
||
raise UpdateError("Неизвестное имя архива обновления") from exc
|
||
|
||
|
||
def extract_release_tag(final_url: str) -> str:
|
||
parsed = urlparse(final_url)
|
||
if parsed.scheme != "https" or parsed.hostname != UPDATE_HOST:
|
||
raise UpdateError("Сервер релизов перенаправил запрос на неизвестный адрес")
|
||
parts = [unquote(part) for part in parsed.path.split("/") if part]
|
||
try:
|
||
release_index = parts.index("releases")
|
||
except ValueError as exc:
|
||
raise UpdateError("Не удалось определить тег последнего релиза") from exc
|
||
if parts[release_index + 1:release_index + 2] != ["tag"]:
|
||
raise UpdateError("Не удалось определить тег последнего релиза")
|
||
tag_parts = parts[release_index + 2:release_index + 3]
|
||
if not tag_parts or not tag_parts[0].strip():
|
||
raise UpdateError("Сервер релизов вернул пустой тег")
|
||
return tag_parts[0].strip()
|
||
|
||
|
||
def build_download_url(tag: str, asset_name: str) -> str:
|
||
encoded_tag = quote(tag, safe="")
|
||
encoded_asset = quote(asset_name, safe="")
|
||
return f"{RELEASE_DOWNLOAD_BASE_URL}/{encoded_tag}/{encoded_asset}"
|
||
|
||
|
||
def check_for_update(
|
||
current_version: str,
|
||
*,
|
||
timeout: float = HTTP_TIMEOUT_SECONDS,
|
||
opener: Callable[..., BinaryIO] | None = None,
|
||
system: str | None = None,
|
||
machine: str | None = None,
|
||
) -> UpdateInfo | None:
|
||
"""Перейти по latest URL и вернуть данные отличающегося релиза."""
|
||
request = Request(
|
||
LATEST_RELEASE_URL,
|
||
headers={
|
||
"Accept": "text/html,application/xhtml+xml",
|
||
"User-Agent": f"PlateConfigApp/{normalize_version(current_version)}",
|
||
},
|
||
)
|
||
open_request = opener or urlopen
|
||
try:
|
||
with open_request(request, timeout=timeout) as response:
|
||
final_url = response.geturl()
|
||
except Exception as exc:
|
||
raise UpdateError(f"Не удалось проверить обновления: {exc}") from exc
|
||
|
||
tag = extract_release_tag(final_url)
|
||
remote_version = parse_version(tag)
|
||
if remote_version <= parse_version(current_version):
|
||
return None
|
||
|
||
version = normalize_version(tag)
|
||
asset_name = platform_asset_name(system=system, machine=machine)
|
||
return UpdateInfo(
|
||
tag=tag,
|
||
version=version,
|
||
asset_name=asset_name,
|
||
download_url=build_download_url(tag, asset_name),
|
||
)
|
||
|
||
|
||
def _validate_executable_signature(signature: bytes, executable_name: str) -> None:
|
||
if executable_name.casefold().endswith(".exe"):
|
||
valid = signature.startswith(b"MZ")
|
||
else:
|
||
valid = signature == b"\x7fELF"
|
||
if not valid:
|
||
raise UpdateError("Сервер вернул файл, не похожий на исполняемый артефакт")
|
||
|
||
|
||
def _validated_archive_members(
|
||
archive: zipfile.ZipFile,
|
||
executable_name: str,
|
||
) -> list[zipfile.ZipInfo]:
|
||
members = archive.infolist()
|
||
if len(members) > MAX_ARCHIVE_MEMBERS:
|
||
raise UpdateError("В архиве обновления слишком много файлов")
|
||
|
||
total_size = 0
|
||
normalized_names: set[str] = set()
|
||
executable_member: zipfile.ZipInfo | None = None
|
||
for member in members:
|
||
raw_name = member.filename
|
||
if not raw_name or "\\" in raw_name or "\x00" in raw_name:
|
||
raise UpdateError("Архив обновления содержит небезопасный путь")
|
||
relative_path = PurePosixPath(raw_name.rstrip("/"))
|
||
if (
|
||
relative_path.is_absolute()
|
||
or not relative_path.parts
|
||
or any(part in {"", ".", ".."} for part in relative_path.parts)
|
||
):
|
||
raise UpdateError("Архив обновления содержит небезопасный путь")
|
||
if not TOP_LEVEL_NAME_PATTERN.fullmatch(relative_path.parts[0]):
|
||
raise UpdateError("Архив обновления содержит недопустимое имя")
|
||
|
||
normalized_name = relative_path.as_posix().casefold()
|
||
if normalized_name in normalized_names:
|
||
raise UpdateError("Архив обновления содержит повторяющиеся пути")
|
||
normalized_names.add(normalized_name)
|
||
|
||
unix_mode = member.external_attr >> 16
|
||
file_type = stat.S_IFMT(unix_mode)
|
||
if file_type not in {0, stat.S_IFREG, stat.S_IFDIR}:
|
||
raise UpdateError("Архив обновления содержит ссылку или специальный файл")
|
||
if member.flag_bits & 0x1:
|
||
raise UpdateError("Зашифрованные архивы обновления не поддерживаются")
|
||
|
||
if not member.is_dir():
|
||
total_size += member.file_size
|
||
if total_size > MAX_UNPACKED_BYTES:
|
||
raise UpdateError("Распакованный размер обновления превышает предел")
|
||
if relative_path.as_posix() == executable_name:
|
||
if member.is_dir():
|
||
raise UpdateError("Исполняемый файл обновления имеет неверный тип")
|
||
executable_member = member
|
||
|
||
if executable_member is None:
|
||
raise UpdateError("В архиве обновления не найден исполняемый файл")
|
||
with archive.open(executable_member, "r") as executable:
|
||
_validate_executable_signature(executable.read(4), executable_name)
|
||
return members
|
||
|
||
|
||
def _validate_update_archive(path: Path, executable_name: str) -> None:
|
||
try:
|
||
with zipfile.ZipFile(path, "r") as archive:
|
||
_validated_archive_members(archive, executable_name)
|
||
bad_member = archive.testzip()
|
||
except (OSError, zipfile.BadZipFile, RuntimeError) as exc:
|
||
raise UpdateError(f"Не удалось проверить ZIP-архив обновления: {exc}") from exc
|
||
if bad_member is not None:
|
||
raise UpdateError(f"Повреждён файл в архиве обновления: {bad_member}")
|
||
|
||
|
||
def _extract_update_archive(
|
||
archive_path: Path,
|
||
staging_directory: Path,
|
||
executable_name: str,
|
||
) -> Path:
|
||
try:
|
||
staging_directory.mkdir()
|
||
with zipfile.ZipFile(archive_path, "r") as archive:
|
||
members = _validated_archive_members(archive, executable_name)
|
||
for member in members:
|
||
relative_path = PurePosixPath(member.filename.rstrip("/"))
|
||
destination = staging_directory.joinpath(*relative_path.parts)
|
||
if member.is_dir():
|
||
destination.mkdir(parents=True, exist_ok=True)
|
||
continue
|
||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||
with (
|
||
archive.open(member, "r") as source,
|
||
destination.open("xb") as output,
|
||
):
|
||
shutil.copyfileobj(source, output)
|
||
|
||
executable_path = staging_directory / executable_name
|
||
with executable_path.open("rb") as executable:
|
||
_validate_executable_signature(executable.read(4), executable_name)
|
||
if not executable_name.casefold().endswith(".exe"):
|
||
executable_path.chmod(0o755)
|
||
return executable_path
|
||
except Exception:
|
||
shutil.rmtree(staging_directory, ignore_errors=True)
|
||
raise
|
||
|
||
|
||
def download_update(
|
||
info: UpdateInfo,
|
||
target_directory: Path,
|
||
*,
|
||
timeout: float = DOWNLOAD_TIMEOUT_SECONDS,
|
||
opener: Callable[..., BinaryIO] | None = None,
|
||
progress: Callable[[int], None] | None = None,
|
||
cancelled: Callable[[], bool] | None = None,
|
||
) -> Path:
|
||
"""Скачать и проверить ZIP-пакет обновления рядом с приложением."""
|
||
target_directory = target_directory.resolve()
|
||
if not target_directory.is_dir():
|
||
raise UpdateError("Каталог приложения не найден")
|
||
if not os.access(target_directory, os.W_OK):
|
||
raise UpdateError("Нет прав на запись в каталог приложения")
|
||
|
||
temporary_path = target_directory / (
|
||
f".{info.asset_name}.download-{uuid.uuid4().hex}"
|
||
)
|
||
request = Request(
|
||
info.download_url,
|
||
headers={
|
||
"Accept": "application/octet-stream",
|
||
"User-Agent": f"PlateConfigAppUpdater/{info.version}",
|
||
},
|
||
)
|
||
open_request = opener or urlopen
|
||
received = 0
|
||
try:
|
||
with open_request(request, timeout=timeout) as response:
|
||
final_url = urlparse(response.geturl())
|
||
if final_url.scheme != "https" or final_url.hostname != UPDATE_HOST:
|
||
raise UpdateError("Загрузка перенаправлена на неизвестный сервер")
|
||
if final_url.path.rstrip("/") == "/user/login":
|
||
raise UpdateError("Release-артефакты недоступны без авторизации")
|
||
content_length = response.headers.get("Content-Length")
|
||
total = int(content_length) if content_length else 0
|
||
if total > MAX_DOWNLOAD_BYTES:
|
||
raise UpdateError("Размер обновления превышает допустимый предел")
|
||
|
||
with temporary_path.open("wb") as output:
|
||
while True:
|
||
if cancelled is not None and cancelled():
|
||
raise UpdateCancelled("Загрузка обновления отменена")
|
||
chunk = response.read(DOWNLOAD_CHUNK_SIZE)
|
||
if not chunk:
|
||
break
|
||
received += len(chunk)
|
||
if received > MAX_DOWNLOAD_BYTES:
|
||
raise UpdateError("Размер обновления превышает допустимый предел")
|
||
output.write(chunk)
|
||
if progress is not None and total > 0:
|
||
progress(min(received * 100 // total, 100))
|
||
output.flush()
|
||
os.fsync(output.fileno())
|
||
|
||
if received == 0:
|
||
raise UpdateError("Сервер вернул пустой файл обновления")
|
||
if total > 0 and received != total:
|
||
raise UpdateError("Файл обновления скачан не полностью")
|
||
executable_name = archive_executable_name(info.asset_name)
|
||
_validate_update_archive(temporary_path, executable_name)
|
||
if progress is not None:
|
||
progress(100)
|
||
return temporary_path
|
||
except Exception:
|
||
temporary_path.unlink(missing_ok=True)
|
||
raise
|
||
|
||
|
||
def current_executable() -> Path:
|
||
if not getattr(sys, "frozen", False):
|
||
raise UpdateError("Самообновление доступно только в собранной версии")
|
||
return Path(sys.executable).resolve()
|
||
|
||
|
||
def signal_update_ready() -> None:
|
||
ready_file = os.environ.pop(UPDATE_READY_ENV, None)
|
||
if ready_file:
|
||
try:
|
||
Path(ready_file).write_text(str(os.getpid()), encoding="ascii")
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _write_windows_updater(script_path: Path) -> None:
|
||
script = r'''param(
|
||
[int]$ApplicationProcessId,
|
||
[string]$Staging,
|
||
[string]$Target,
|
||
[string]$Backup,
|
||
[string]$ReadyFile
|
||
)
|
||
$ErrorActionPreference = "Stop"
|
||
$applicationDirectory = Split-Path -Parent $Target
|
||
$errorLog = Join-Path $applicationDirectory "update-error.log"
|
||
$installed = [System.Collections.Generic.List[string]]::new()
|
||
$backedUp = [System.Collections.Generic.List[string]]::new()
|
||
$restarted = $null
|
||
try {
|
||
Wait-Process -Id $ApplicationProcessId -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
|
||
New-Item -ItemType Directory -Path $Backup | Out-Null
|
||
$entries = @(Get-ChildItem -LiteralPath $Staging -Force)
|
||
foreach ($entry in $entries) {
|
||
$destination = Join-Path $applicationDirectory $entry.Name
|
||
if (Test-Path -LiteralPath $destination) {
|
||
$deadline = (Get-Date).AddSeconds(30)
|
||
while ($true) {
|
||
try {
|
||
Move-Item -LiteralPath $destination -Destination $Backup -Force -ErrorAction Stop
|
||
break
|
||
} catch {
|
||
if ((Get-Date) -ge $deadline) { throw }
|
||
Start-Sleep -Milliseconds 250
|
||
}
|
||
}
|
||
$backedUp.Add($entry.Name)
|
||
}
|
||
Move-Item -LiteralPath $entry.FullName -Destination $destination -Force -ErrorAction Stop
|
||
$installed.Add($entry.Name)
|
||
}
|
||
if (-not (Test-Path -LiteralPath $Target -PathType Leaf)) {
|
||
throw "В пакете обновления отсутствует исполняемый файл"
|
||
}
|
||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||
$previousReadyFile = $env:PLATE_CONFIG_UPDATE_READY_FILE
|
||
$previousResetEnvironment = $env:PYINSTALLER_RESET_ENVIRONMENT
|
||
try {
|
||
$env:PLATE_CONFIG_UPDATE_READY_FILE = $ReadyFile
|
||
$env:PYINSTALLER_RESET_ENVIRONMENT = "1"
|
||
$restarted = Start-Process -FilePath $Target -WorkingDirectory $applicationDirectory -PassThru
|
||
} finally {
|
||
$env:PLATE_CONFIG_UPDATE_READY_FILE = $previousReadyFile
|
||
$env:PYINSTALLER_RESET_ENVIRONMENT = $previousResetEnvironment
|
||
}
|
||
$deadline = (Get-Date).AddSeconds(60)
|
||
while (-not (Test-Path -LiteralPath $ReadyFile -PathType Leaf)) {
|
||
if ($restarted.HasExited) { throw "Новая версия завершилась до запуска интерфейса" }
|
||
if ((Get-Date) -ge $deadline) { throw "Новая версия не подтвердила запуск" }
|
||
Start-Sleep -Milliseconds 250
|
||
}
|
||
if ($restarted.HasExited) { throw "Новая версия завершилась сразу после запуска" }
|
||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force
|
||
} catch {
|
||
$failure = $_
|
||
if ($restarted -and -not $restarted.HasExited) {
|
||
& taskkill.exe /PID $restarted.Id /T /F *> $null
|
||
Start-Sleep -Milliseconds 500
|
||
}
|
||
$rollbackErrors = [System.Collections.Generic.List[string]]::new()
|
||
foreach ($name in $installed) {
|
||
$destination = Join-Path $applicationDirectory $name
|
||
if (Test-Path -LiteralPath $destination) {
|
||
try {
|
||
Remove-Item -LiteralPath $destination -Recurse -Force -ErrorAction Stop
|
||
} catch {
|
||
$rollbackErrors.Add($_.ToString())
|
||
}
|
||
}
|
||
}
|
||
foreach ($name in $backedUp) {
|
||
$backupItem = Join-Path $Backup $name
|
||
$destination = Join-Path $applicationDirectory $name
|
||
if (Test-Path -LiteralPath $backupItem) {
|
||
try {
|
||
Move-Item -LiteralPath $backupItem -Destination $destination -Force -ErrorAction Stop
|
||
} catch {
|
||
$rollbackErrors.Add($_.ToString())
|
||
}
|
||
}
|
||
}
|
||
Remove-Item -LiteralPath $ReadyFile -Force -ErrorAction SilentlyContinue
|
||
Remove-Item -LiteralPath $Staging -Recurse -Force -ErrorAction SilentlyContinue
|
||
if ($rollbackErrors.Count -eq 0) {
|
||
Remove-Item -LiteralPath $Backup -Recurse -Force -ErrorAction SilentlyContinue
|
||
}
|
||
@($failure.ToString()) + @($rollbackErrors) | Out-File -LiteralPath $errorLog -Encoding utf8
|
||
Remove-Item -LiteralPath $MyInvocation.MyCommand.Path -Force -ErrorAction SilentlyContinue
|
||
exit 1
|
||
}
|
||
'''
|
||
script_path.write_text(script, encoding="utf-8-sig", newline="\r\n")
|
||
|
||
|
||
def _write_linux_updater(script_path: Path) -> None:
|
||
script = r'''#!/bin/sh
|
||
set -eu
|
||
application_pid="$1"
|
||
staging_path="$2"
|
||
target_path="$3"
|
||
backup_path="$4"
|
||
ready_path="$5"
|
||
application_dir="$(dirname -- "$target_path")"
|
||
error_log="$application_dir/update-error.log"
|
||
installed_file="$backup_path/.installed"
|
||
backed_up_file="$backup_path/.backed_up"
|
||
new_pid=""
|
||
rollback() {
|
||
exit_code="$?"
|
||
trap - EXIT HUP INT TERM
|
||
set +e
|
||
if [ -n "$new_pid" ]; then
|
||
kill "$new_pid" 2>/dev/null
|
||
wait "$new_pid" 2>/dev/null
|
||
fi
|
||
rollback_failed=0
|
||
if [ -f "$installed_file" ]; then
|
||
while IFS= read -r name; do
|
||
rm -rf -- "$application_dir/$name" || rollback_failed=1
|
||
done < "$installed_file"
|
||
fi
|
||
if [ -f "$backed_up_file" ]; then
|
||
while IFS= read -r name; do
|
||
if [ -e "$backup_path/$name" ]; then
|
||
mv -f -- "$backup_path/$name" "$application_dir/$name" || rollback_failed=1
|
||
fi
|
||
done < "$backed_up_file"
|
||
fi
|
||
rm -f -- "$ready_path"
|
||
rm -rf -- "$staging_path"
|
||
if [ "$rollback_failed" -eq 0 ]; then
|
||
rm -rf -- "$backup_path"
|
||
fi
|
||
printf '%s\n' "Не удалось заменить приложение" >"$error_log"
|
||
rm -f -- "$0"
|
||
if [ "$exit_code" -eq 0 ]; then
|
||
exit_code=1
|
||
fi
|
||
exit "$exit_code"
|
||
}
|
||
trap rollback EXIT HUP INT TERM
|
||
while kill -0 "$application_pid" 2>/dev/null; do
|
||
sleep 0.2
|
||
done
|
||
rm -rf -- "$backup_path"
|
||
mkdir -p -- "$backup_path"
|
||
: >"$installed_file"
|
||
: >"$backed_up_file"
|
||
for source_path in "$staging_path"/*; do
|
||
name="$(basename -- "$source_path")"
|
||
destination="$application_dir/$name"
|
||
if [ -e "$destination" ]; then
|
||
mv -f -- "$destination" "$backup_path/$name"
|
||
printf '%s\n' "$name" >>"$backed_up_file"
|
||
fi
|
||
mv -f -- "$source_path" "$destination"
|
||
printf '%s\n' "$name" >>"$installed_file"
|
||
done
|
||
test -f "$target_path"
|
||
chmod 755 -- "$target_path"
|
||
rm -f -- "$ready_path"
|
||
PLATE_CONFIG_UPDATE_READY_FILE="$ready_path" PYINSTALLER_RESET_ENVIRONMENT=1 "$target_path" >/dev/null 2>&1 &
|
||
new_pid=$!
|
||
attempts=0
|
||
while [ ! -f "$ready_path" ]; do
|
||
if ! kill -0 "$new_pid" 2>/dev/null; then
|
||
exit 1
|
||
fi
|
||
attempts=$((attempts + 1))
|
||
if [ "$attempts" -ge 240 ]; then
|
||
exit 1
|
||
fi
|
||
sleep 0.25
|
||
done
|
||
if ! kill -0 "$new_pid" 2>/dev/null; then
|
||
exit 1
|
||
fi
|
||
trap - EXIT HUP INT TERM
|
||
rm -f -- "$ready_path"
|
||
rm -rf -- "$backup_path" "$staging_path"
|
||
rm -f -- "$0"
|
||
'''
|
||
script_path.write_text(script, encoding="utf-8", newline="\n")
|
||
script_path.chmod(0o700)
|
||
|
||
|
||
def launch_update(
|
||
downloaded_path: Path,
|
||
*,
|
||
executable_path: Path | None = None,
|
||
process_id: int | None = None,
|
||
system: str | None = None,
|
||
) -> None:
|
||
"""Распаковать обновление и запустить helper для замены пакета."""
|
||
archive_path = downloaded_path.resolve()
|
||
target = (executable_path or current_executable()).resolve()
|
||
if not archive_path.is_file():
|
||
raise UpdateError("Скачанный архив обновления не найден")
|
||
if not os.access(target.parent, os.W_OK):
|
||
raise UpdateError("Нет прав на замену приложения")
|
||
|
||
updater_id = uuid.uuid4().hex
|
||
application_pid = process_id or os.getpid()
|
||
system_name = (system or platform.system()).casefold()
|
||
if system_name not in {"windows", "linux"}:
|
||
raise UpdateError(f"Автообновление не поддерживает платформу {system_name}")
|
||
|
||
staging_path = target.parent / f".plate_config_update-{updater_id}"
|
||
backup_path = target.parent / f".plate_config_backup-{updater_id}"
|
||
ready_path = target.parent / f".plate_config_ready-{updater_id}"
|
||
script_path: Path | None = None
|
||
try:
|
||
_extract_update_archive(archive_path, staging_path, target.name)
|
||
archive_path.unlink()
|
||
except Exception:
|
||
shutil.rmtree(staging_path, ignore_errors=True)
|
||
raise
|
||
|
||
if system_name == "windows":
|
||
script_path = target.parent / f".plate_config_updater-{updater_id}.ps1"
|
||
try:
|
||
_write_windows_updater(script_path)
|
||
creation_flags = (
|
||
getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
||
| getattr(subprocess, "CREATE_NO_WINDOW", 0)
|
||
)
|
||
subprocess.Popen(
|
||
[
|
||
"powershell.exe",
|
||
"-NoProfile",
|
||
"-ExecutionPolicy",
|
||
"Bypass",
|
||
"-File",
|
||
str(script_path),
|
||
str(application_pid),
|
||
str(staging_path),
|
||
str(target),
|
||
str(backup_path),
|
||
str(ready_path),
|
||
],
|
||
close_fds=True,
|
||
creationflags=creation_flags,
|
||
)
|
||
except Exception:
|
||
script_path.unlink(missing_ok=True)
|
||
shutil.rmtree(staging_path, ignore_errors=True)
|
||
raise
|
||
return
|
||
if system_name == "linux":
|
||
script_path = target.parent / f".plate_config_updater-{updater_id}.sh"
|
||
try:
|
||
_write_linux_updater(script_path)
|
||
subprocess.Popen(
|
||
[
|
||
"/bin/sh",
|
||
str(script_path),
|
||
str(application_pid),
|
||
str(staging_path),
|
||
str(target),
|
||
str(backup_path),
|
||
str(ready_path),
|
||
],
|
||
cwd=str(target.parent),
|
||
close_fds=True,
|
||
start_new_session=True,
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=subprocess.DEVNULL,
|
||
stderr=subprocess.DEVNULL,
|
||
)
|
||
except Exception:
|
||
script_path.unlink(missing_ok=True)
|
||
shutil.rmtree(staging_path, ignore_errors=True)
|
||
raise
|
||
return
|