Скачивание образов было доработано по итогам ревью: зависшие операции, гонка запусков, отмена при проверке и распаковке
- Проверка «уже скачивается» идёт по всем операциям каталога (glob), а не по последним 50; скачивания без живого потока закрываются при старте (reap_stale_downloads), их .part удаляются, а отмена такой операции завершает её сразу, не дожидаясь потока. - Два запуска одного файла больше не пишут в один .part: проверки и создание операции под замком, файл создаётся монопольно (O_EXCL), вторая копия завершается ошибкой, не трогая чужой файл; осиротевший .part без владельца убирается перед новым запуском. - Отмена во время проверки контрольной суммы и распаковки .img.xz останавливает работу: операция завершается как отменённая, недокачанный файл, архив и недораспакованный образ удаляются. - Битый .img.xz после неудачной распаковки удаляется с понятным сообщением, повторное скачивание разрешено; проверка «файл уже есть» учитывает распакованный .img и qcow2 после конвертации. - Ссылки с кириллицей (домен IDNA, путь и параметры в процентах) открываются, некорректный порт и непонятные символы отклоняются сразу; ошибки сети описаны словами, исходный текст — в журнале. - Прогресс раз в секунду перечитывает JSON операции, чтобы не затирать cancel_requested и замечать отмену из другого процесса; операции читаются один раз на страницу; отмена в карточке — не ошибка. - Формулировки карточек и сообщения о недоступной архитектуре упрощены; добавлены тесты на каждое поведение. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019LEUXMcdZo7b87mjkUsPgg
Этот коммит содержится в:
@@ -7,15 +7,19 @@
|
||||
образов из каталога). Скачивание можно отменить — недокачанный файл удаляется.
|
||||
"""
|
||||
import hashlib
|
||||
import json
|
||||
import lzma
|
||||
import re
|
||||
import shutil
|
||||
import socket
|
||||
import ssl
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.parse import unquote, urlsplit
|
||||
from typing import Any, Callable
|
||||
from urllib.parse import quote, unquote, urlsplit, urlunsplit
|
||||
|
||||
from fastapi import APIRouter, Form, Request
|
||||
from fastapi.responses import JSONResponse
|
||||
@@ -34,7 +38,6 @@ from core import (
|
||||
require_auth,
|
||||
run_cmd,
|
||||
run_operation,
|
||||
running_operations,
|
||||
update_operation,
|
||||
)
|
||||
|
||||
@@ -52,16 +55,44 @@ KINDS: dict[str, dict[str, Any]] = {
|
||||
"disk": {"suffixes": (".qcow2", ".img", ".raw", ".img.xz"), "page": "/disk-images", "label": "образ диска", "hint": "Подойдут файлы .qcow2, .img, .raw и сжатые .img.xz"},
|
||||
}
|
||||
|
||||
# Отмена: идентификаторы операций, которые попросили остановить. Поток
|
||||
# загрузки проверяет множество между кусками файла.
|
||||
CANCEL_LOCK = threading.Lock()
|
||||
# Состояние скачиваний этого процесса (под одним замком):
|
||||
# - CANCEL_REQUESTS — операции, которые попросили остановить; поток загрузки
|
||||
# проверяет множество между кусками файла, при проверке суммы и распаковке;
|
||||
# - LIVE_DOWNLOADS — операция → путь к файлу для скачиваний, которыми владеет
|
||||
# живой поток. Операция без потока (после перезапуска панели) — сирота: её
|
||||
# закрывают сразу, ничего не ожидая.
|
||||
# Множества живут в памяти одного процесса: панель запускается с одним воркером
|
||||
# uvicorn. На случай нескольких процессов поток раз в секунду читает флаг
|
||||
# cancel_requested и из JSON операции.
|
||||
STATE_LOCK = threading.Lock()
|
||||
CANCEL_REQUESTS: set[str] = set()
|
||||
LIVE_DOWNLOADS: dict[str, str] = {}
|
||||
# Проверки «файл уже есть» / «уже скачивается» и создание операции идут под одним
|
||||
# замком, чтобы два запуска одного файла не прошли проверку одновременно.
|
||||
START_LOCK = threading.Lock()
|
||||
|
||||
# Понятные объяснения частых ошибок HTTP; остальные коды показываем числом.
|
||||
HTTP_ERRORS = {
|
||||
401: "Сервер требует вход по паролю (ошибка 401)",
|
||||
403: "Сервер запретил скачивание (ошибка 403)",
|
||||
404: "Файл по ссылке не найден (ошибка 404)",
|
||||
410: "Файл по ссылке больше не выдаётся (ошибка 410)",
|
||||
429: "Сервер просит подождать: слишком много запросов (ошибка 429)",
|
||||
500: "На сервере произошла ошибка (500)",
|
||||
502: "Сервер временно недоступен (ошибка 502)",
|
||||
503: "Сервер временно недоступен (ошибка 503)",
|
||||
504: "Сервер временно недоступен (ошибка 504)",
|
||||
}
|
||||
|
||||
|
||||
class DownloadError(ValueError):
|
||||
"""Понятная пользователю ошибка проверки ссылки или файла."""
|
||||
|
||||
|
||||
class DownloadCancelled(Exception):
|
||||
"""Пользователь отменил скачивание во время проверки суммы или распаковки."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- formatting
|
||||
def human_size(value: float | int | None) -> str:
|
||||
"""1234567 → «1,2 МБ». Десятичная запятая, как принято в русском."""
|
||||
@@ -107,14 +138,40 @@ def validate_url(url: str) -> str:
|
||||
raise DownloadError("Вставьте ссылку на файл")
|
||||
if len(url) > 2048 or any(ch.isspace() for ch in url):
|
||||
raise DownloadError("Ссылка выглядит некорректно")
|
||||
parts = urlsplit(url)
|
||||
try:
|
||||
parts = urlsplit(url)
|
||||
except ValueError as exc:
|
||||
raise DownloadError("Ссылка выглядит некорректно") from exc
|
||||
if parts.scheme.lower() not in ("http", "https"):
|
||||
raise DownloadError("Ссылка должна начинаться с http:// или https://")
|
||||
if not parts.hostname:
|
||||
raise DownloadError("В ссылке не указан адрес сервера")
|
||||
encode_url(url) # кириллица в домене или пути, порт — проверяем сразу, а не в фоне
|
||||
return url
|
||||
|
||||
|
||||
def encode_url(url: str) -> str:
|
||||
"""Ссылка в виде, который понимает urllib: домен с кириллицей — в IDNA, не-ASCII
|
||||
символы в пути и параметрах — в проценты. Уже закодированные ссылки не меняются."""
|
||||
parts = urlsplit(url)
|
||||
try:
|
||||
port = parts.port
|
||||
except ValueError as exc:
|
||||
raise DownloadError("Ссылка выглядит некорректно") from exc
|
||||
try:
|
||||
host = (parts.hostname or "").encode("idna").decode("ascii")
|
||||
except UnicodeError as exc:
|
||||
raise DownloadError("Ссылка содержит символы, которые сервер не понимает") from exc
|
||||
netloc = f"[{host}]" if ":" in host else host
|
||||
if port:
|
||||
netloc = f"{netloc}:{port}"
|
||||
if "@" in parts.netloc:
|
||||
netloc = f"{parts.netloc.rsplit('@', 1)[0]}@{netloc}"
|
||||
path = quote(parts.path, safe="/%:@!$&'()*+,;=~")
|
||||
query = quote(parts.query, safe="/%:@!$&'()*+,;=?~")
|
||||
return urlunsplit((parts.scheme.lower(), netloc, path, query, ""))
|
||||
|
||||
|
||||
def filename_from_url(url: str, kind: str) -> str:
|
||||
"""Имя файла из ссылки, приведённое к безопасному виду (как при загрузке с компьютера)."""
|
||||
if kind not in KINDS:
|
||||
@@ -205,6 +262,63 @@ def file_digest(path: Path, algo: str, on_progress: Any = None) -> str:
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- operations
|
||||
def unfinished_downloads() -> list[dict[str, Any]]:
|
||||
"""Незавершённые скачивания по всему каталогу операций, а не по последним 50,
|
||||
как running_operations: зависшая старая операция иначе не попадёт в проверку."""
|
||||
core.ensure_operations_dir()
|
||||
found = []
|
||||
for path in core.OPERATIONS_DIR.glob("*.json"):
|
||||
try:
|
||||
data = json.loads(path.read_text())
|
||||
except (OSError, ValueError):
|
||||
continue
|
||||
if data.get("type") == "download" and data.get("status") in ("queued", "running"):
|
||||
found.append(data)
|
||||
found.sort(key=lambda op: str(op.get("created_at") or ""))
|
||||
return found
|
||||
|
||||
|
||||
def is_live(operation_id: str) -> bool:
|
||||
with STATE_LOCK:
|
||||
return operation_id in LIVE_DOWNLOADS
|
||||
|
||||
|
||||
def release_download(operation_id: str) -> None:
|
||||
with STATE_LOCK:
|
||||
LIVE_DOWNLOADS.pop(operation_id, None)
|
||||
CANCEL_REQUESTS.discard(operation_id)
|
||||
|
||||
|
||||
def reap_stale_downloads() -> int:
|
||||
"""Скачивания без живого потока (панель перезапускалась) закрываются, а их
|
||||
недокачанные .part удаляются — иначе они лежали бы в каталоге образов вечно.
|
||||
|
||||
Операции всех типов закрывает core.interrupt_orphaned_operations() при старте;
|
||||
здесь остаётся убрать файлы .part, которых та не касается. Возвращает число
|
||||
закрытых операций."""
|
||||
count = 0
|
||||
for op in unfinished_downloads():
|
||||
if is_live(op["id"]):
|
||||
continue
|
||||
message = "Скачивание прервано: панель была перезапущена. Запустите его заново"
|
||||
if op.get("target_path"):
|
||||
part_path(Path(op["target_path"])).unlink(missing_ok=True)
|
||||
release_download(op["id"])
|
||||
finish_operation(op, False, message, interrupted=True)
|
||||
count += 1
|
||||
with STATE_LOCK:
|
||||
live_targets = set(LIVE_DOWNLOADS.values())
|
||||
for directory in (core.ISO_DIR, core.DISK_IMAGES_DIR):
|
||||
if not directory.is_dir():
|
||||
continue
|
||||
for part in directory.glob(".*.part"):
|
||||
if str(part.with_name(part.name[1:-5]).resolve()) in live_targets:
|
||||
continue
|
||||
part.unlink(missing_ok=True)
|
||||
return count
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- catalog
|
||||
def host_arch() -> str:
|
||||
arch = str(host_profile.load_host_profile().get("arch") or "").lower()
|
||||
@@ -224,10 +338,12 @@ def image_candidates(filename: str) -> list[str]:
|
||||
return names
|
||||
|
||||
|
||||
def catalog_entries(arch: str | None = None) -> list[dict[str, Any]]:
|
||||
"""Каталог готовых систем для процессора сервера с пометками «Загружен» / «Скачивается»."""
|
||||
def catalog_entries(arch: str | None = None, active: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
|
||||
"""Каталог готовых систем для процессора сервера с пометками «Загружен» / «Скачивается».
|
||||
Список идущих скачиваний можно передать готовым, чтобы не читать каталог операций дважды."""
|
||||
arch = arch or host_arch()
|
||||
active = {op.get("catalog_id"): op for op in running_operations("download") if op.get("catalog_id")}
|
||||
ops = unfinished_downloads() if active is None else active
|
||||
by_catalog_id = {op.get("catalog_id"): op for op in ops if op.get("catalog_id")}
|
||||
entries: list[dict[str, Any]] = []
|
||||
for item in catalog.CLOUD_IMAGES:
|
||||
source = item.get("arches", {}).get(arch)
|
||||
@@ -238,7 +354,7 @@ def catalog_entries(arch: str | None = None) -> list[dict[str, Any]]:
|
||||
except DownloadError:
|
||||
continue
|
||||
ready = next((name for name in reversed(image_candidates(filename)) if (core.DISK_IMAGES_DIR / name).exists()), None)
|
||||
operation = active.get(item["id"])
|
||||
operation = by_catalog_id.get(item["id"])
|
||||
entries.append({
|
||||
"id": item["id"], "title": item["title"], "vendor": item.get("vendor", ""), "hint": item.get("hint", ""),
|
||||
"recommended": bool(item.get("recommended")), "login": item.get("login", ""), "filename": filename,
|
||||
@@ -248,9 +364,9 @@ def catalog_entries(arch: str | None = None) -> list[dict[str, Any]]:
|
||||
return entries
|
||||
|
||||
|
||||
def active_downloads(kind: str | None = None) -> list[dict[str, Any]]:
|
||||
def active_downloads(kind: str | None = None, ops: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]:
|
||||
result = []
|
||||
for op in running_operations("download"):
|
||||
for op in unfinished_downloads() if ops is None else ops:
|
||||
if kind and op.get("download_kind") != kind:
|
||||
continue
|
||||
op["indeterminate"] = not op.get("total_bytes") and op.get("status") == "running"
|
||||
@@ -259,24 +375,29 @@ def active_downloads(kind: str | None = None) -> list[dict[str, Any]]:
|
||||
|
||||
|
||||
def downloads_panel(kind: str) -> dict[str, Any]:
|
||||
"""Данные для карточек скачивания на страницах /iso и /disk-images (глобальная функция шаблонов)."""
|
||||
"""Данные для карточек скачивания на страницах /iso и /disk-images (глобальная функция шаблонов).
|
||||
Каталог операций читается один раз на вызов; страница вызывает функцию один раз и передаёт результат во вставки."""
|
||||
kind = kind if kind in KINDS else "iso"
|
||||
return {"kind": kind, "page": KINDS[kind]["page"], "hint": KINDS[kind]["hint"], "active": active_downloads(kind), "catalog": catalog_entries() if kind == "disk" else []}
|
||||
ops = unfinished_downloads()
|
||||
return {"kind": kind, "page": KINDS[kind]["page"], "hint": KINDS[kind]["hint"], "active": active_downloads(kind, ops), "catalog": catalog_entries(active=ops) if kind == "disk" else []}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- cancel
|
||||
def cancel_requested(operation_id: str) -> bool:
|
||||
with CANCEL_LOCK:
|
||||
with STATE_LOCK:
|
||||
return operation_id in CANCEL_REQUESTS
|
||||
|
||||
|
||||
def clear_cancel(operation_id: str) -> None:
|
||||
with CANCEL_LOCK:
|
||||
with STATE_LOCK:
|
||||
CANCEL_REQUESTS.discard(operation_id)
|
||||
|
||||
|
||||
def request_cancel(operation_id: str) -> tuple[dict[str, Any] | None, str | None]:
|
||||
"""Попросить остановить скачивание. Возвращает (операция, ошибка)."""
|
||||
"""Попросить остановить скачивание. Возвращает (операция, ошибка).
|
||||
|
||||
Если операцией не владеет ни один поток (осталась от перезапуска панели),
|
||||
ждать некого — она закрывается сразу, а недокачанный файл удаляется."""
|
||||
operation = read_operation(operation_id)
|
||||
if not operation:
|
||||
return None, "Задача не найдена"
|
||||
@@ -284,9 +405,17 @@ def request_cancel(operation_id: str) -> tuple[dict[str, Any] | None, str | None
|
||||
return operation, "Отменить можно только скачивание"
|
||||
if operation.get("status") not in ("queued", "running"):
|
||||
return operation, "Скачивание уже завершено"
|
||||
with CANCEL_LOCK:
|
||||
CANCEL_REQUESTS.add(operation_id)
|
||||
operation.pop("log_tail", None)
|
||||
with STATE_LOCK:
|
||||
live = operation_id in LIVE_DOWNLOADS
|
||||
if live:
|
||||
CANCEL_REQUESTS.add(operation_id)
|
||||
if not live:
|
||||
if operation.get("target_path"):
|
||||
part_path(Path(operation["target_path"])).unlink(missing_ok=True)
|
||||
append_operation_log(operation_id, "Пользователь отменил скачивание. Поток загрузки не найден (панель перезапускалась), недокачанный файл удалён.")
|
||||
finish_operation(operation, False, "Скачивание отменено", cancelled=True, cancel_requested=True)
|
||||
return read_operation(operation_id) or operation, None
|
||||
update_operation(operation, cancel_requested=True, message="Останавливаем скачивание…")
|
||||
append_operation_log(operation_id, "Пользователь попросил отменить скачивание.")
|
||||
return operation, None
|
||||
@@ -300,22 +429,29 @@ def start_download(url: str, kind: str, *, title: str | None = None, sums_url: s
|
||||
url = validate_url(url)
|
||||
filename = filename_from_url(url, kind)
|
||||
target = target_path(kind, filename)
|
||||
if target.exists():
|
||||
raise DownloadError(f"Файл {filename} уже есть на сервере. Удалите его, если нужно скачать заново")
|
||||
if any(op.get("target_path") == str(target) for op in running_operations("download")):
|
||||
raise DownloadError(f"Файл {filename} уже скачивается")
|
||||
if sums_url:
|
||||
validate_url(sums_url)
|
||||
if algo not in hashlib.algorithms_available:
|
||||
raise DownloadError("Неизвестный алгоритм контрольной суммы")
|
||||
ok, free = free_space_ok(target.parent, None)
|
||||
if not ok:
|
||||
raise DownloadError(f"На сервере мало места: свободно {human_size(free)}")
|
||||
operation = new_operation(
|
||||
"download", title or f"Скачивание {filename}", message="Подключаемся к серверу…",
|
||||
download_kind=kind, url=url, filename=filename, target_path=str(target), source_path=str(target), page=KINDS[kind]["page"],
|
||||
sums_url=sums_url, algo=algo, catalog_id=catalog_id, total_bytes=None, downloaded_bytes=0, cancel_requested=False,
|
||||
)
|
||||
with START_LOCK:
|
||||
# Образ мог остаться под другим именем: распакованный .img или qcow2 после конвертации.
|
||||
existing = next((name for name in image_candidates(filename) if (target.parent / name).exists()), None)
|
||||
if existing:
|
||||
raise DownloadError(f"Файл {existing} уже есть на сервере. Удалите его, если нужно скачать заново")
|
||||
if any(op.get("target_path") == str(target) for op in unfinished_downloads()):
|
||||
raise DownloadError(f"Файл {filename} уже скачивается")
|
||||
# Владельца у .part нет (проверили выше) — это остаток прерванной загрузки.
|
||||
part_path(target).unlink(missing_ok=True)
|
||||
ok, free = free_space_ok(target.parent, None)
|
||||
if not ok:
|
||||
raise DownloadError(f"На сервере мало места: свободно {human_size(free)}")
|
||||
operation = new_operation(
|
||||
"download", title or f"Скачивание {filename}", message="Подключаемся к серверу…",
|
||||
download_kind=kind, url=url, filename=filename, target_path=str(target), source_path=str(target), page=KINDS[kind]["page"],
|
||||
sums_url=sums_url, algo=algo, catalog_id=catalog_id, total_bytes=None, downloaded_bytes=0, cancel_requested=False,
|
||||
)
|
||||
with STATE_LOCK:
|
||||
LIVE_DOWNLOADS[operation["id"]] = str(target)
|
||||
append_operation_log(operation["id"], f"Ссылка: {url}")
|
||||
append_operation_log(operation["id"], f"Файл: {target}")
|
||||
run_operation(operation, download_worker)
|
||||
@@ -323,20 +459,34 @@ def start_download(url: str, kind: str, *, title: str | None = None, sums_url: s
|
||||
|
||||
|
||||
def open_url(url: str) -> Any:
|
||||
request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"})
|
||||
request = urllib.request.Request(encode_url(url), headers={"User-Agent": USER_AGENT, "Accept": "*/*"})
|
||||
return urllib.request.urlopen(request, timeout=TIMEOUT) # noqa: S310 — схема проверена в validate_url
|
||||
|
||||
|
||||
def describe_network_error(exc: Exception) -> str:
|
||||
"""Ошибка сети словами для пользователя; исходный текст остаётся в журнале операции."""
|
||||
if isinstance(exc, urllib.error.HTTPError):
|
||||
return f"Сервер ответил ошибкой {exc.code} {exc.reason}"
|
||||
return HTTP_ERRORS.get(exc.code, f"Сервер ответил ошибкой {exc.code}")
|
||||
if isinstance(exc, urllib.error.URLError):
|
||||
reason = exc.reason
|
||||
if isinstance(reason, TimeoutError) or "timed out" in str(reason).lower():
|
||||
text = str(reason).lower()
|
||||
if isinstance(reason, TimeoutError) or "timed out" in text:
|
||||
return f"Сервер не отвечает (ждали {TIMEOUT} с)"
|
||||
if isinstance(reason, socket.gaierror) or "name or service not known" in text or "nodename nor servname" in text or "name resolution" in text:
|
||||
return "Не удалось найти сервер по этой ссылке — проверьте адрес"
|
||||
if isinstance(reason, ConnectionRefusedError) or "connection refused" in text:
|
||||
return "Сервер отказал в подключении"
|
||||
if isinstance(reason, ssl.SSLError) or "certificate" in text or "ssl" in text:
|
||||
return "У сервера проблема с сертификатом безопасности"
|
||||
if isinstance(reason, UnicodeError):
|
||||
return "Ссылка содержит символы, которые сервер не понимает"
|
||||
return f"Не удалось подключиться к серверу: {reason}"
|
||||
if isinstance(exc, UnicodeError):
|
||||
return "Ссылка содержит символы, которые сервер не понимает"
|
||||
if isinstance(exc, TimeoutError) or "timed out" in str(exc).lower():
|
||||
return f"Сервер перестал отвечать (ждали {TIMEOUT} с)"
|
||||
if isinstance(exc, (ConnectionResetError, BrokenPipeError, ConnectionAbortedError)):
|
||||
return "Сервер разорвал соединение"
|
||||
return str(exc)[:240]
|
||||
|
||||
|
||||
@@ -346,6 +496,7 @@ def fetch_expected_checksum(sums_url: str, filename: str, operation_id: str) ->
|
||||
with open_url(sums_url) as response:
|
||||
text = response.read(SUMS_MAX_BYTES).decode("utf-8", errors="replace")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
append_operation_log(operation_id, f"Ошибка: {exc}")
|
||||
raise DownloadError(f"Не удалось скачать файл контрольных сумм: {describe_network_error(exc)}") from exc
|
||||
expected = parse_checksums(text, filename)
|
||||
if not expected:
|
||||
@@ -354,6 +505,13 @@ def fetch_expected_checksum(sums_url: str, filename: str, operation_id: str) ->
|
||||
|
||||
|
||||
def download_worker(operation: dict[str, Any]) -> None:
|
||||
try:
|
||||
run_download(operation)
|
||||
finally:
|
||||
release_download(operation["id"])
|
||||
|
||||
|
||||
def run_download(operation: dict[str, Any]) -> None:
|
||||
operation_id = operation["id"]
|
||||
kind = operation["download_kind"]
|
||||
target = Path(operation["target_path"])
|
||||
@@ -364,27 +522,35 @@ def download_worker(operation: dict[str, Any]) -> None:
|
||||
def cancelled() -> bool:
|
||||
return cancel_requested(operation_id)
|
||||
|
||||
def fail(message: str, **meta: Any) -> None:
|
||||
part.unlink(missing_ok=True)
|
||||
clear_cancel(operation_id)
|
||||
def check_cancel(*_args: Any) -> None:
|
||||
if cancelled():
|
||||
raise DownloadCancelled
|
||||
|
||||
def fail(message: str, *, keep_part: bool = False, **meta: Any) -> None:
|
||||
if not keep_part:
|
||||
part.unlink(missing_ok=True)
|
||||
finish_operation(operation, False, message, **meta)
|
||||
|
||||
def stop() -> None:
|
||||
append_operation_log(operation_id, "Скачивание остановлено, недокачанный файл удалён.")
|
||||
def stop(note: str = "Скачивание остановлено, недокачанный файл удалён.") -> None:
|
||||
append_operation_log(operation_id, note)
|
||||
fail("Скачивание отменено", cancelled=True, cancel_requested=True)
|
||||
|
||||
def network_failure(prefix: str, exc: Exception) -> None:
|
||||
append_operation_log(operation_id, f"Ошибка: {exc}")
|
||||
fail(f"{prefix}{describe_network_error(exc)}")
|
||||
|
||||
if cancelled():
|
||||
stop()
|
||||
return
|
||||
try:
|
||||
response = open_url(operation["url"])
|
||||
except Exception as exc: # noqa: BLE001
|
||||
fail(describe_network_error(exc))
|
||||
network_failure("", exc)
|
||||
return
|
||||
|
||||
with response:
|
||||
final_url = response.geturl()
|
||||
if final_url != operation["url"]:
|
||||
if final_url != encode_url(operation["url"]):
|
||||
append_operation_log(operation_id, f"Перенаправление: {final_url}")
|
||||
length = response.headers.get("Content-Length")
|
||||
total = int(length) if length and length.isdigit() else None
|
||||
@@ -401,7 +567,9 @@ def download_worker(operation: dict[str, Any]) -> None:
|
||||
last_done = 0
|
||||
speed: float | None = None
|
||||
try:
|
||||
with part.open("wb") as handle:
|
||||
# Файл создаётся монопольно: если .part уже пишет другой поток, вторая загрузка
|
||||
# не должна ни писать в него, ни удалять его при ошибке.
|
||||
with part.open("xb") as handle:
|
||||
while True:
|
||||
if cancelled():
|
||||
stop()
|
||||
@@ -413,6 +581,15 @@ def download_worker(operation: dict[str, Any]) -> None:
|
||||
done += len(chunk)
|
||||
now = time.monotonic()
|
||||
if now - last_tick >= 1.0:
|
||||
# Раз в секунду перечитываем JSON операции: так запись прогресса не затирает
|
||||
# флаг cancel_requested, который выставил request_cancel, а отмена из другого
|
||||
# процесса панели (несколько воркеров uvicorn) тоже будет замечена.
|
||||
fresh = read_operation(operation_id) or {}
|
||||
fresh.pop("log_tail", None)
|
||||
operation.update(fresh)
|
||||
if cancelled() or operation.get("cancel_requested"):
|
||||
stop()
|
||||
return
|
||||
instant = (done - last_done) / (now - last_tick)
|
||||
speed = instant if speed is None else speed * 0.7 + instant * 0.3
|
||||
last_tick, last_done = now, done
|
||||
@@ -423,8 +600,12 @@ def download_worker(operation: dict[str, Any]) -> None:
|
||||
fail(f"На сервере закончилось место: свободно {human_size(free)}")
|
||||
return
|
||||
update_operation(operation, progress=min(progress, download_share), downloaded_bytes=done, speed_bps=int(speed), message=progress_message(done, total, speed))
|
||||
except FileExistsError:
|
||||
append_operation_log(operation_id, f"Файл {part.name} уже создан другой загрузкой — эта копия остановлена, чужой файл не тронут.")
|
||||
fail(f"Файл {target.name} уже скачивается", keep_part=True)
|
||||
return
|
||||
except Exception as exc: # noqa: BLE001
|
||||
fail(f"Загрузка прервалась: {describe_network_error(exc)}")
|
||||
network_failure("Загрузка прервалась: ", exc)
|
||||
return
|
||||
|
||||
if total is not None and done != total:
|
||||
@@ -434,37 +615,48 @@ def download_worker(operation: dict[str, Any]) -> None:
|
||||
append_operation_log(operation_id, f"Скачано {human_size(done)} за {human_eta(elapsed) or '1 с'} ({human_size(done / elapsed)}/с).")
|
||||
update_operation(operation, progress=download_share, downloaded_bytes=done, total_bytes=done if total is None else total, message="Файл скачан, проверяем…")
|
||||
|
||||
if verify:
|
||||
try:
|
||||
try:
|
||||
if verify:
|
||||
expected = fetch_expected_checksum(operation["sums_url"], operation["filename"], operation_id)
|
||||
except DownloadError as exc:
|
||||
fail(str(exc))
|
||||
return
|
||||
algo = operation.get("algo") or "sha256"
|
||||
update_operation(operation, progress=97, message=f"Проверяем контрольную сумму ({algo})…")
|
||||
actual = file_digest(part, algo)
|
||||
if actual != expected:
|
||||
append_operation_log(operation_id, f"Ожидали {algo}: {expected}")
|
||||
append_operation_log(operation_id, f"Получили {algo}: {actual}")
|
||||
fail("Контрольная сумма не совпала — файл повреждён или подменён, поэтому удалён. Попробуйте скачать ещё раз")
|
||||
return
|
||||
append_operation_log(operation_id, f"Контрольная сумма {algo} совпала.")
|
||||
check_cancel()
|
||||
algo = operation.get("algo") or "sha256"
|
||||
update_operation(operation, progress=97, message=f"Проверяем контрольную сумму ({algo})…")
|
||||
actual = file_digest(part, algo, on_progress=check_cancel)
|
||||
if actual != expected:
|
||||
append_operation_log(operation_id, f"Ожидали {algo}: {expected}")
|
||||
append_operation_log(operation_id, f"Получили {algo}: {actual}")
|
||||
fail("Контрольная сумма не совпала — файл повреждён или подменён, поэтому удалён. Попробуйте скачать ещё раз")
|
||||
return
|
||||
append_operation_log(operation_id, f"Контрольная сумма {algo} совпала.")
|
||||
check_cancel()
|
||||
except DownloadCancelled:
|
||||
stop("Скачивание остановлено во время проверки, файл удалён.")
|
||||
return
|
||||
except DownloadError as exc:
|
||||
fail(str(exc))
|
||||
return
|
||||
|
||||
if target.exists():
|
||||
fail(f"Файл {target.name} появился на сервере, пока шло скачивание. Скачанная копия удалена")
|
||||
return
|
||||
part.replace(target)
|
||||
clear_cancel(operation_id)
|
||||
append_operation_log(operation_id, f"Файл сохранён: {target}")
|
||||
try:
|
||||
result = after_download(operation, target, kind)
|
||||
result = after_download(operation, target, kind, cancelled)
|
||||
except DownloadCancelled:
|
||||
append_operation_log(operation_id, "Распаковка остановлена, скачанный архив и недораспакованный файл удалены.")
|
||||
finish_operation(operation, False, "Скачивание отменено", cancelled=True, cancel_requested=True)
|
||||
return
|
||||
except DownloadError as exc:
|
||||
finish_operation(operation, False, str(exc))
|
||||
return
|
||||
except Exception as exc: # noqa: BLE001
|
||||
finish_operation(operation, False, f"Файл скачан, но не удалось его подготовить: {str(exc)[:200]}", result_path=str(target))
|
||||
return
|
||||
finish_operation(operation, True, result["message"], **result["meta"])
|
||||
|
||||
|
||||
def after_download(operation: dict[str, Any], target: Path, kind: str) -> dict[str, Any]:
|
||||
def after_download(operation: dict[str, Any], target: Path, kind: str, should_stop: Callable[[], bool] | None = None) -> dict[str, Any]:
|
||||
"""Что делать со скачанным файлом: обновить пул ISO либо распаковать и конвертировать диск."""
|
||||
operation_id = operation["id"]
|
||||
if kind == "iso":
|
||||
@@ -476,7 +668,17 @@ def after_download(operation: dict[str, Any], target: Path, kind: str) -> dict[s
|
||||
if target.name.lower().endswith(".img.xz"):
|
||||
update_operation(operation, progress=98, message=f"Распаковываем {target.name}…")
|
||||
append_operation_log(operation_id, "Распаковываем сжатый образ…")
|
||||
path = extract_xz_image(target)
|
||||
try:
|
||||
path = extract_xz_image(target, should_stop)
|
||||
except DownloadCancelled:
|
||||
target.unlink(missing_ok=True)
|
||||
raise
|
||||
except Exception as exc: # noqa: BLE001
|
||||
# Битый архив со страницы не видно и не удалить (там только .img/.raw/.qcow2),
|
||||
# а повторное скачивание он бы блокировал — убираем сами.
|
||||
target.unlink(missing_ok=True)
|
||||
append_operation_log(operation_id, f"Ошибка распаковки: {exc}")
|
||||
raise DownloadError("Не удалось распаковать скачанный архив — файл повреждён и удалён. Попробуйте скачать ещё раз") from exc
|
||||
target.unlink(missing_ok=True)
|
||||
append_operation_log(operation_id, f"Распакован: {path}")
|
||||
convert = start_convert(path)
|
||||
@@ -488,10 +690,25 @@ def after_download(operation: dict[str, Any], target: Path, kind: str) -> dict[s
|
||||
return {"message": f"Готово: {path.name} — можно создавать машину", "meta": meta}
|
||||
|
||||
|
||||
def extract_xz_image(compressed: Path) -> Path:
|
||||
def extract_xz_image(compressed: Path, should_stop: Callable[[], bool] | None = None) -> Path:
|
||||
"""Распаковать .img.xz рядом с архивом (имя подбирает app, как при загрузке с компьютера).
|
||||
Между кусками проверяет should_stop: при отмене бросает DownloadCancelled, недораспакованный файл удаляется."""
|
||||
import app # локальный импорт: app.py подключает этот модуль в самом конце
|
||||
|
||||
return app.extract_xz_disk_image(compressed)
|
||||
extracted = app.unique_disk_image_path(compressed.name[:-3])
|
||||
try:
|
||||
with lzma.open(compressed, "rb") as source, extracted.open("wb") as handle:
|
||||
while True:
|
||||
if should_stop and should_stop():
|
||||
raise DownloadCancelled
|
||||
chunk = source.read(CHUNK_SIZE)
|
||||
if not chunk:
|
||||
break
|
||||
handle.write(chunk)
|
||||
except BaseException:
|
||||
extracted.unlink(missing_ok=True)
|
||||
raise
|
||||
return extracted
|
||||
|
||||
|
||||
def start_convert(path: Path) -> dict[str, Any] | None:
|
||||
@@ -529,7 +746,7 @@ def download_from_catalog(request: Request, image_id: str):
|
||||
arch = host_arch()
|
||||
source = entry.get("arches", {}).get(arch)
|
||||
if not source:
|
||||
return redirect_with_message("/disk-images", "download_error", f"{entry['title']} не выпускается для процессора {arch}")
|
||||
return redirect_with_message("/disk-images", "download_error", f"У {entry['title']} нет версии для этого сервера ({arch})")
|
||||
try:
|
||||
operation = start_download(source["url"], "disk", title=f"Скачивание {entry['title']}", sums_url=source.get("sums"), algo=source.get("algo"), catalog_id=image_id)
|
||||
except DownloadError as exc:
|
||||
@@ -546,7 +763,11 @@ def cancel_download(request: Request, operation_id: str):
|
||||
page = page_for(str(operation.get("download_kind"))) if operation else "/iso"
|
||||
if error:
|
||||
return redirect_with_message(page, "download_error", error)
|
||||
return redirect_with_message(page, "download_message", f"Останавливаем скачивание {operation.get('filename', '')}".rstrip())
|
||||
filename = operation.get("filename", "")
|
||||
if operation.get("status") == "error":
|
||||
# Потока не было — операция уже закрыта, ждать нечего.
|
||||
return redirect_with_message(page, "download_message", " ".join(word for word in ("Скачивание", filename, "отменено") if word))
|
||||
return redirect_with_message(page, "download_message", f"Останавливаем скачивание {filename}".rstrip())
|
||||
|
||||
|
||||
@router.post("/api/operations/{operation_id}/cancel")
|
||||
@@ -562,6 +783,12 @@ def api_cancel_operation(request: Request, operation_id: str):
|
||||
return {"ok": True, "operation": read_operation(operation_id) or operation}
|
||||
|
||||
|
||||
@router.on_event("startup")
|
||||
def close_stale_downloads() -> None:
|
||||
"""После перезапуска панели потоков загрузки нет: закрываем такие операции и убираем их .part."""
|
||||
reap_stale_downloads()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- templates
|
||||
def register_template_globals() -> None:
|
||||
"""Функция downloads_panel нужна шаблонам iso.html и disk_images.html, которые
|
||||
|
||||
Ссылка в новой задаче
Block a user