136 lines
5.3 KiB
Python
136 lines
5.3 KiB
Python
"""API-слой для /api/updates/* (статус, принудительная проверка, установка по выбору пользователя)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
from flask import current_app
|
|
|
|
from app.services.auto_update_runtime import get_auto_updater
|
|
from app.services.update_state_store import read_update_state
|
|
|
|
FORCED_CHECK_MIN_INTERVAL_SEC = 60
|
|
_BODY_PREVIEW_LEN = 500
|
|
|
|
|
|
def _iso(dt: Optional[datetime]) -> Optional[str]:
|
|
if dt is None:
|
|
return None
|
|
return dt.replace(microsecond=0).isoformat() + "Z"
|
|
|
|
|
|
def build_updates_status_payload() -> Dict[str, Any]:
|
|
updater = get_auto_updater()
|
|
ver = str(current_app.config.get("SYNC_CLIENT_VERSION", "2.0.0") or "2.0.0")
|
|
restart_cmd = str(current_app.config.get("UPDATE_RESTART_CMD", "") or "").strip()
|
|
|
|
if updater is None:
|
|
base = str(current_app.config.get("WESP_BASE_DIR") or "")
|
|
last_state = read_update_state(base) if base else {"status": "idle"}
|
|
return {
|
|
"initialized": False,
|
|
"check_enabled": False,
|
|
"current_version": ver,
|
|
"update_available": False,
|
|
"pending_version": None,
|
|
"pending_name": None,
|
|
"pending_body": None,
|
|
"published_at": None,
|
|
"last_check_at": None,
|
|
"is_running": False,
|
|
"is_updating": False,
|
|
"update_progress": None,
|
|
"last_update_state": last_state,
|
|
"gitea_configured": False,
|
|
"restart_configured": bool(restart_cmd),
|
|
"gitea_url": None,
|
|
"gitea_repo": None,
|
|
}
|
|
|
|
pending = updater.get_pending_update()
|
|
gitea_ok = bool(getattr(updater, "gitea_url", "") and getattr(updater, "gitea_owner", "") and getattr(updater, "gitea_repo", ""))
|
|
body = (pending or {}).get("body") or ""
|
|
if isinstance(body, str) and len(body) > _BODY_PREVIEW_LEN:
|
|
body = body[:_BODY_PREVIEW_LEN].rstrip() + "…"
|
|
|
|
progress = None
|
|
if getattr(updater, "is_updating", False):
|
|
progress = getattr(updater, "get_update_progress", lambda: None)()
|
|
|
|
last_state = read_update_state(getattr(updater, "base_dir", "") or "")
|
|
|
|
return {
|
|
"initialized": True,
|
|
"check_enabled": bool(getattr(updater, "enabled", False)),
|
|
"current_version": getattr(updater, "current_version", ver) or ver,
|
|
"update_available": pending is not None,
|
|
"pending_version": (pending or {}).get("version"),
|
|
"pending_name": (pending or {}).get("name"),
|
|
"pending_body": body if pending else None,
|
|
"published_at": (pending or {}).get("published_at"),
|
|
"last_check_at": _iso(getattr(updater, "last_check_at", None)),
|
|
"is_running": bool(getattr(updater, "is_running", False)),
|
|
"is_updating": bool(getattr(updater, "is_updating", False)),
|
|
"update_progress": progress,
|
|
"last_update_state": last_state,
|
|
"gitea_configured": gitea_ok,
|
|
"restart_configured": bool(restart_cmd or getattr(updater, "restart_cmd", "")),
|
|
"gitea_url": getattr(updater, "gitea_url", None) if gitea_ok else None,
|
|
"gitea_repo": (
|
|
f"{updater.gitea_owner}/{updater.gitea_repo}"
|
|
if gitea_ok and getattr(updater, "gitea_owner", None)
|
|
else None
|
|
),
|
|
}
|
|
|
|
|
|
def force_check_updates() -> Dict[str, Any]:
|
|
updater = get_auto_updater()
|
|
if updater is None:
|
|
payload = build_updates_status_payload()
|
|
payload["message"] = "Проверка обновлений не инициализирована"
|
|
return payload
|
|
|
|
if not getattr(updater, "enabled", False):
|
|
payload = build_updates_status_payload()
|
|
payload["message"] = "Проверка обновлений отключена"
|
|
return payload
|
|
|
|
now = time.monotonic()
|
|
if now - float(getattr(updater, "_last_forced_check_mono", 0.0)) >= FORCED_CHECK_MIN_INTERVAL_SEC:
|
|
updater._last_forced_check_mono = now
|
|
updater.check_for_updates()
|
|
|
|
payload = build_updates_status_payload()
|
|
payload["forced_check"] = True
|
|
return payload
|
|
|
|
|
|
def install_pending_update() -> Tuple[Dict[str, Any], int]:
|
|
updater = get_auto_updater()
|
|
if updater is None:
|
|
return {"error": "Проверка обновлений не инициализирована", "status": "error"}, 500
|
|
|
|
if not getattr(updater, "enabled", False):
|
|
return {"error": "Проверка обновлений отключена", "status": "error"}, 400
|
|
|
|
if getattr(updater, "is_updating", False):
|
|
return {"error": "Обновление уже выполняется", "status": "error"}, 409
|
|
|
|
pending = updater.get_pending_update()
|
|
if not pending:
|
|
return {"error": "Нет доступного обновления для установки", "status": "error"}, 404
|
|
|
|
success = updater.update(pending)
|
|
if not success:
|
|
return {"error": "Не удалось применить обновление", "status": "error"}, 500
|
|
|
|
return {
|
|
"status": "success",
|
|
"success": True,
|
|
"message": "Обновление применено. Система перезапускается…",
|
|
"new_version": getattr(updater, "current_version", None),
|
|
}, 200
|