1344 lines
58 KiB
Python
1344 lines
58 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
Модуль автообновления
|
||
"""
|
||
|
||
import os
|
||
|
||
import wesp_runtime_env
|
||
|
||
wesp_runtime_env.apply_kiosk_headless_env()
|
||
|
||
from app.services.auto_update_db import (
|
||
load_auto_update_dict,
|
||
recipes_db_path,
|
||
upsert_auto_update_dict,
|
||
)
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
import threading
|
||
import shutil
|
||
import zipfile
|
||
import subprocess
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from typing import Dict, List, Optional, Tuple
|
||
import logging
|
||
from urllib.parse import urlparse, urlunparse
|
||
import base64
|
||
import hashlib
|
||
import shlex
|
||
|
||
# Обязательный импорт cryptography
|
||
try:
|
||
from cryptography.fernet import Fernet
|
||
from cryptography.hazmat.primitives import hashes
|
||
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
|
||
except ImportError:
|
||
# logger еще не определен, используем print
|
||
print("❌ КРИТИЧЕСКАЯ ОШИБКА: Библиотека cryptography обязательна для работы автообновления!")
|
||
print(" Установите: pip install cryptography")
|
||
raise ImportError("Библиотека cryptography обязательна. Установите: pip install cryptography")
|
||
|
||
# Опциональный импорт keyring для системных хранилищ секретов
|
||
try:
|
||
import keyring
|
||
|
||
KEYRING_AVAILABLE = True
|
||
except ImportError:
|
||
keyring = None # type: ignore[assignment,misc]
|
||
KEYRING_AVAILABLE = False
|
||
|
||
# Настраиваем logging ПЕРЕД использованием logger
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='%(asctime)s | %(levelname)-8s | %(name)-20s | %(funcName)-20s | %(lineno)-4d | %(message)s',
|
||
datefmt='%Y-%m-%d %H:%M:%S'
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def wesp_data_config_path(script_dir: Optional[str] = None) -> str:
|
||
"""Путь к data/config.json (единый с Flask и sync_client_state)."""
|
||
base = script_dir or os.path.dirname(os.path.abspath(__file__))
|
||
return os.path.join(base, "data", "config.json")
|
||
|
||
|
||
# Сообщение про keyring — только при загрузке секретов из файла (см. _load_secrets).
|
||
|
||
class SecretEncryption:
|
||
"""Класс для шифрования/расшифровки секретов с использованием соли"""
|
||
|
||
@staticmethod
|
||
def _collect_decrypt_passwords() -> list:
|
||
"""Уникальные ключи для расшифровки (порядок: keyring → WESP_MASTER_PASSWORD → SECRET_KEY)."""
|
||
seen = set()
|
||
passwords = []
|
||
|
||
def _add(candidate):
|
||
if isinstance(candidate, str) and len(candidate) >= 8 and candidate not in seen:
|
||
seen.add(candidate)
|
||
passwords.append(candidate)
|
||
|
||
if KEYRING_AVAILABLE:
|
||
try:
|
||
_add(keyring.get_password("WESP_AutoUpdate", "master_password"))
|
||
except Exception as e:
|
||
logger.warning("Системное хранилище секретов недоступно: %s", e)
|
||
|
||
_add((os.environ.get("WESP_MASTER_PASSWORD") or "").strip() or None)
|
||
|
||
try:
|
||
from config import get_config_class
|
||
|
||
cls = get_config_class()
|
||
sk = getattr(cls, "SECRET_KEY", None)
|
||
_add(sk if isinstance(sk, str) else None)
|
||
except Exception as exc:
|
||
logger.debug("SECRET_KEY как ключ расшифровки недоступен: %s", exc)
|
||
|
||
return passwords
|
||
|
||
@staticmethod
|
||
def _get_master_password() -> str:
|
||
"""
|
||
Первый доступный ключ шифрования (для encrypt без явного password).
|
||
При расшифровке decrypt_secret перебирает все источники через _collect_decrypt_passwords.
|
||
"""
|
||
passwords = SecretEncryption._collect_decrypt_passwords()
|
||
if passwords:
|
||
return passwords[0]
|
||
|
||
logger.error("КРИТИЧЕСКАЯ ОШИБКА: не задан ключ для расшифровки секретов Gitea!")
|
||
logger.error("Задайте WESP_SECRET_KEY (рекомендуется), WESP_MASTER_PASSWORD или keyring; либо сохраните секреты в админке.")
|
||
raise ValueError("Не задан ключ для расшифровки секретов Gitea")
|
||
|
||
@staticmethod
|
||
def _get_encryption_key(password: Optional[str] = None, salt: Optional[bytes] = None) -> bytes:
|
||
"""
|
||
Генерирует ключ шифрования из пароля и соли
|
||
Использует PBKDF2 для создания ключа
|
||
"""
|
||
if password is None:
|
||
password = SecretEncryption._get_master_password()
|
||
|
||
if salt is None:
|
||
# Генерируем соль из системной информации (стабильная для одного сервера)
|
||
system_info = f"{os.getenv('COMPUTERNAME', '')}{os.getenv('USERNAME', '')}{os.path.expanduser('~')}"
|
||
salt = hashlib.sha256(system_info.encode()).digest()[:16]
|
||
|
||
# Используем PBKDF2 для создания ключа
|
||
kdf = PBKDF2HMAC(
|
||
algorithm=hashes.SHA256(),
|
||
length=32,
|
||
salt=salt,
|
||
iterations=100000,
|
||
)
|
||
key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
|
||
return key
|
||
|
||
@staticmethod
|
||
def encrypt_secret(plaintext: str, password: Optional[str] = None) -> str:
|
||
"""Шифрует секрет с использованием соли (обязательно)"""
|
||
if not plaintext:
|
||
raise ValueError("Нельзя шифровать пустое значение")
|
||
|
||
key = SecretEncryption._get_encryption_key(password)
|
||
fernet = Fernet(key)
|
||
encrypted = fernet.encrypt(plaintext.encode())
|
||
return base64.urlsafe_b64encode(encrypted).decode()
|
||
|
||
@staticmethod
|
||
def decrypt_secret(ciphertext: str, password: Optional[str] = None) -> str:
|
||
"""Расшифровывает секрет (обязательно зашифрованный)."""
|
||
if not ciphertext:
|
||
raise ValueError("Нельзя расшифровать пустое значение")
|
||
|
||
passwords = [password] if password is not None else SecretEncryption._collect_decrypt_passwords()
|
||
if not passwords:
|
||
raise ValueError("Не задан ключ для расшифровки секретов Gitea")
|
||
|
||
last_err = None
|
||
encrypted = base64.urlsafe_b64decode(ciphertext.encode())
|
||
for pwd in passwords:
|
||
try:
|
||
key = SecretEncryption._get_encryption_key(pwd)
|
||
fernet = Fernet(key)
|
||
decrypted = fernet.decrypt(encrypted)
|
||
return decrypted.decode()
|
||
except Exception as e:
|
||
last_err = e
|
||
|
||
logger.error("Ошибка расшифровки (перепробованы все ключи): %s", last_err)
|
||
raise ValueError(
|
||
f"Не удалось расшифровать секрет. Сохраните секреты заново в админке или проверьте WESP_SECRET_KEY: {last_err}"
|
||
)
|
||
|
||
@staticmethod
|
||
def is_encrypted(value: str) -> bool:
|
||
"""Проверяет, зашифровано ли значение (Fernet или base64(Fernet) из encrypt_secret)."""
|
||
if not isinstance(value, str):
|
||
return False
|
||
s = value.strip()
|
||
if len(s) < 24:
|
||
return False
|
||
try:
|
||
decoded = base64.urlsafe_b64decode(s.encode())
|
||
if len(decoded) >= 16:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
return s.startswith("gAAAAAB") and len(s) > 50
|
||
|
||
|
||
class AutoUpdater:
|
||
"""Менеджер автообновления через Gitea API с защитой данных"""
|
||
|
||
def __init__(self):
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
self.base_dir = script_dir
|
||
self.data_dir = os.path.join(script_dir, "data")
|
||
os.makedirs(self.data_dir, exist_ok=True)
|
||
self.config_file = wesp_data_config_path(script_dir)
|
||
legacy_root_cfg = os.path.join(script_dir, "config.json")
|
||
if not os.path.exists(self.config_file) and os.path.exists(legacy_root_cfg):
|
||
try:
|
||
shutil.copy2(legacy_root_cfg, self.config_file)
|
||
logger.info("Скопирован legacy config.json из корня в data/config.json")
|
||
except OSError as exc:
|
||
logger.warning("Не удалось скопировать config.json в data/: %s", exc)
|
||
self.is_running = False
|
||
self.update_thread = None
|
||
self.is_updating = False
|
||
self.pending_release: Optional[Dict] = None
|
||
self.update_progress: Optional[Dict] = None
|
||
self.last_check_at: Optional[datetime] = None
|
||
self._state_lock = threading.Lock()
|
||
self._last_forced_check_mono: float = 0.0
|
||
self.restart_cmd = self._resolve_restart_cmd()
|
||
|
||
self.config = self.load_config()
|
||
self.update_config = self.config.get('auto_update', {})
|
||
if not isinstance(self.update_config, dict):
|
||
self.update_config = {}
|
||
try:
|
||
_ci = int(self.update_config.get('check_interval_sec', 3600) or 3600)
|
||
except (TypeError, ValueError):
|
||
_ci = 3600
|
||
self.check_interval = max(60, min(86400, _ci))
|
||
|
||
self.enabled = self.update_config.get('enabled', False)
|
||
self.gitea_url = ""
|
||
self.gitea_owner = ""
|
||
self.gitea_repo = ""
|
||
self.gitea_token = ""
|
||
self.gitea_username = ""
|
||
self.gitea_password = ""
|
||
|
||
secrets = self._load_secrets()
|
||
|
||
self.gitea_url = self.update_config.get('gitea_url', '') or secrets.get('gitea_url', '')
|
||
self.gitea_owner = self.update_config.get('gitea_owner', '') or secrets.get('gitea_owner', '')
|
||
self.gitea_repo = self.update_config.get('gitea_repo', '') or secrets.get('gitea_repo', '')
|
||
self.gitea_token = secrets.get('gitea_token', '') or self.update_config.get('gitea_token', '')
|
||
self.gitea_username = secrets.get('gitea_username', '') or self.update_config.get('gitea_username', '')
|
||
self.gitea_password = secrets.get('gitea_password', '') or self.update_config.get('gitea_password', '')
|
||
|
||
repository_url = self.update_config.get('repository_url', '') or secrets.get('repository_url', '')
|
||
if repository_url:
|
||
self._parse_repository_url(repository_url)
|
||
|
||
self.current_version = self._resolve_current_version()
|
||
|
||
self.protected_files = [
|
||
'data/recipes.db',
|
||
'data/recipes.db-shm',
|
||
'data/recipes.db-wal',
|
||
'data/reports.db',
|
||
'data/reports.db-shm',
|
||
'data/reports.db-wal',
|
||
'data/update_state.json',
|
||
'data/.pending_restart.json',
|
||
'credentials.json',
|
||
'calibration_factor.json',
|
||
'weight_0.json',
|
||
'gitea_secrets.json', # Секреты Gitea
|
||
]
|
||
|
||
self.protected_folders = [
|
||
'__pycache__',
|
||
'.git',
|
||
'backups',
|
||
'temp_updates',
|
||
'.secret',
|
||
'*.db-shm',
|
||
'*.db-wal',
|
||
]
|
||
|
||
# Файлы, которые обновляются с умным слиянием
|
||
self.merge_files = [
|
||
'config.json'
|
||
]
|
||
|
||
logger.info(
|
||
"AutoUpdater: версия %s, проверка %s, Gitea %s",
|
||
self.current_version,
|
||
"включена" if self.enabled else "выключена",
|
||
f"{self.gitea_url}/{self.gitea_owner}/{self.gitea_repo}" if self.gitea_url else "не настроен",
|
||
)
|
||
|
||
@staticmethod
|
||
def _resolve_restart_cmd() -> str:
|
||
cmd = (os.environ.get("WESP_UPDATE_RESTART_CMD") or "").strip()
|
||
if cmd:
|
||
return cmd
|
||
try:
|
||
from config import get_config_class
|
||
|
||
return str(getattr(get_config_class(), "UPDATE_RESTART_CMD", "") or "").strip()
|
||
except Exception:
|
||
return ""
|
||
|
||
def _resolve_current_version(self) -> str:
|
||
v = str(self.config.get("version") or "").strip()
|
||
if v:
|
||
return v
|
||
try:
|
||
from config import get_config_class
|
||
|
||
cls = get_config_class()
|
||
sv = str(getattr(cls, "SYNC_CLIENT_VERSION", "") or "").strip()
|
||
if sv:
|
||
return sv
|
||
except Exception as exc:
|
||
logger.debug("SYNC_CLIENT_VERSION недоступен: %s", exc)
|
||
return (os.getenv("WESP_SYNC_CLIENT_VERSION", "2.0.0") or "2.0.0").strip() or "2.0.0"
|
||
|
||
def refresh_check_cache(self, release_info: Optional[Dict]) -> None:
|
||
with self._state_lock:
|
||
self.last_check_at = datetime.now(timezone.utc)
|
||
self.pending_release = dict(release_info) if release_info else None
|
||
|
||
def get_pending_update(self) -> Optional[Dict]:
|
||
with self._state_lock:
|
||
if not self.pending_release:
|
||
return None
|
||
return dict(self.pending_release)
|
||
|
||
def clear_pending_after_install(self) -> None:
|
||
with self._state_lock:
|
||
self.pending_release = None
|
||
|
||
def _set_update_progress(
|
||
self,
|
||
stage: str,
|
||
message: str,
|
||
percent: int,
|
||
*,
|
||
detail: Optional[str] = None,
|
||
target_version: Optional[str] = None,
|
||
) -> None:
|
||
with self._state_lock:
|
||
self.update_progress = {
|
||
"stage": stage,
|
||
"message": message,
|
||
"percent": max(0, min(100, int(percent))),
|
||
"detail": detail,
|
||
"target_version": target_version,
|
||
}
|
||
|
||
def get_update_progress(self) -> Optional[Dict]:
|
||
with self._state_lock:
|
||
if not self.update_progress:
|
||
return None
|
||
return dict(self.update_progress)
|
||
|
||
def _clear_update_progress(self) -> None:
|
||
with self._state_lock:
|
||
self.update_progress = None
|
||
|
||
def _parse_repository_url(self, url: str):
|
||
"""Парсит полный URL репозитория Gitea"""
|
||
try:
|
||
parsed = urlparse(url)
|
||
# Формат: https://git.example.com/owner/repo
|
||
parts = parsed.path.strip('/').split('/')
|
||
if len(parts) >= 2:
|
||
if not self.gitea_url:
|
||
self.gitea_url = f"{parsed.scheme}://{parsed.netloc}"
|
||
if not self.gitea_owner:
|
||
self.gitea_owner = parts[0]
|
||
if not self.gitea_repo:
|
||
self.gitea_repo = parts[1]
|
||
logger.info(f"Распарсен URL репозитория: {self.gitea_url}/{self.gitea_owner}/{self.gitea_repo}")
|
||
except Exception as e:
|
||
logger.warning(f"Ошибка парсинга URL репозитория: {e}")
|
||
|
||
def load_config(self) -> Dict:
|
||
"""config.json (прочие ключи) + auto_update из recipes.db, если БД есть; иначе legacy из JSON."""
|
||
cfg: Dict = {}
|
||
try:
|
||
if os.path.exists(self.config_file):
|
||
with open(self.config_file, "r", encoding="utf-8") as f:
|
||
raw = json.load(f)
|
||
if isinstance(raw, dict):
|
||
cfg = raw
|
||
except Exception as e:
|
||
logger.error(f"Ошибка загрузки config.json: {e}")
|
||
au_sql = load_auto_update_dict(self.base_dir)
|
||
if au_sql is not None:
|
||
cfg["auto_update"] = au_sql
|
||
elif recipes_db_path(self.base_dir).is_file():
|
||
cfg["auto_update"] = {}
|
||
elif not isinstance(cfg.get("auto_update"), dict):
|
||
cfg["auto_update"] = {}
|
||
return cfg
|
||
|
||
def _load_secrets(self) -> Dict:
|
||
"""
|
||
Безопасная загрузка секретов с приоритетом:
|
||
1. Переменные окружения (наивысший приоритет)
|
||
2. Файл .secret/gitea_secrets.json
|
||
3. config.json (низший приоритет, только для публичных настроек)
|
||
"""
|
||
secrets = {}
|
||
|
||
# 1. Переменные окружения (наивысший приоритет)
|
||
env_mapping = {
|
||
'GITEA_URL': 'gitea_url',
|
||
'GITEA_OWNER': 'gitea_owner',
|
||
'GITEA_REPO': 'gitea_repo',
|
||
'GITEA_TOKEN': 'gitea_token',
|
||
'GITEA_USERNAME': 'gitea_username',
|
||
'GITEA_PASSWORD': 'gitea_password',
|
||
'GITEA_REPOSITORY_URL': 'repository_url',
|
||
}
|
||
|
||
for env_key, secret_key in env_mapping.items():
|
||
value = os.getenv(env_key, '')
|
||
if value:
|
||
secrets[secret_key] = value
|
||
logger.debug(f"Секрет загружен из переменной окружения: {secret_key}")
|
||
|
||
# 2. Файл .secret/gitea_secrets.json
|
||
secret_file = os.path.join(self.base_dir, '.secret', 'gitea_secrets.json')
|
||
if os.path.exists(secret_file):
|
||
try:
|
||
# Устанавливаем права только для чтения владельцем
|
||
if os.name != 'nt': # Unix-like системы
|
||
os.chmod(secret_file, 0o600)
|
||
|
||
with open(secret_file, 'r', encoding='utf-8') as f:
|
||
file_secrets = json.load(f)
|
||
# Объединяем, но не перезаписываем переменные окружения
|
||
skipped = 0
|
||
for key, value in file_secrets.items():
|
||
if key not in secrets and value:
|
||
if isinstance(value, str):
|
||
if not SecretEncryption.is_encrypted(value):
|
||
logger.warning(
|
||
"Пропуск %s в gitea_secrets.json: значение не зашифровано. "
|
||
"Сохраните секреты через админку.",
|
||
key,
|
||
)
|
||
skipped += 1
|
||
continue
|
||
|
||
try:
|
||
decrypted = SecretEncryption.decrypt_secret(value)
|
||
secrets[key] = decrypted
|
||
logger.debug("Секрет расшифрован и загружен: %s", key)
|
||
except Exception as e:
|
||
logger.warning(
|
||
"Пропуск %s в gitea_secrets.json (не расшифровывается): %s. "
|
||
"Задайте новые секреты в админке.",
|
||
key,
|
||
e,
|
||
)
|
||
skipped += 1
|
||
else:
|
||
secrets[key] = value
|
||
logger.debug("Значение загружено из файла: %s", key)
|
||
if skipped:
|
||
logger.info(
|
||
"gitea_secrets.json: пропущено %s пол(я/ей). "
|
||
"Админка → Обновления (Gitea) → сохраните токен или включите «Удалить секреты».",
|
||
skipped,
|
||
)
|
||
except Exception as e:
|
||
logger.warning("Ошибка загрузки секретов из файла: %s", e)
|
||
|
||
# 3. config.json уже загружен в self.update_config, но секреты там не должны быть
|
||
|
||
return secrets
|
||
|
||
def save_config(self, config: Dict):
|
||
"""Пишет data/config.json без ключа auto_update; auto_update — в recipes.db."""
|
||
try:
|
||
forbidden_keys = ["master_password", "WESP_MASTER_PASSWORD", "wesp_master_password"]
|
||
for key in forbidden_keys:
|
||
if key in config:
|
||
logger.error(f"КРИТИЧЕСКАЯ ОШИБКА: Попытка сохранить мастер-пароль в config.json!")
|
||
logger.error(" Мастер-пароль НИКОГДА не должен сохраняться в файлы!")
|
||
raise ValueError(f"Попытка сохранить мастер-пароль в конфигурацию: {key}")
|
||
|
||
cfg_out = dict(config) if isinstance(config, dict) else {}
|
||
au = cfg_out.pop("auto_update", None)
|
||
|
||
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
|
||
with open(self.config_file, "w", encoding="utf-8") as f:
|
||
json.dump(cfg_out, f, ensure_ascii=False, indent=2)
|
||
|
||
if isinstance(au, dict):
|
||
try:
|
||
upsert_auto_update_dict(self.base_dir, au)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка записи auto_update в SQLite: {e}", exc_info=True)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка сохранения конфигурации: {e}")
|
||
|
||
def _get_headers(self) -> Dict[str, str]:
|
||
"""Получение заголовков для запросов к Gitea API"""
|
||
headers = {
|
||
'Accept': 'application/json',
|
||
'User-Agent': 'WESP-AutoUpdater/1.0'
|
||
}
|
||
|
||
# Приоритет: токен > логин/пароль > без аутентификации
|
||
if self.gitea_token:
|
||
headers['Authorization'] = f'token {self.gitea_token}'
|
||
elif self.gitea_username and self.gitea_password:
|
||
import base64
|
||
credentials = f"{self.gitea_username}:{self.gitea_password}"
|
||
encoded = base64.b64encode(credentials.encode()).decode()
|
||
headers['Authorization'] = f'Basic {encoded}'
|
||
|
||
return headers
|
||
|
||
def check_for_updates(self) -> Optional[Dict]:
|
||
"""
|
||
Проверка наличия обновлений через Gitea API
|
||
Возвращает информацию о последнем релизе, если версия новее текущей
|
||
"""
|
||
if not self.enabled:
|
||
return None
|
||
|
||
if not self.gitea_url or not self.gitea_owner or not self.gitea_repo:
|
||
logger.warning("Настройки Gitea не заполнены")
|
||
self.refresh_check_cache(None)
|
||
return None
|
||
|
||
try:
|
||
# Gitea API: GET /api/v1/repos/{owner}/{repo}/releases
|
||
api_url = f"{self.gitea_url}/api/v1/repos/{self.gitea_owner}/{self.gitea_repo}/releases"
|
||
|
||
logger.info(f"Проверка обновлений: {api_url}")
|
||
response = requests.get(
|
||
api_url,
|
||
headers=self._get_headers(),
|
||
timeout=10
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
body_preview = (response.text or "")[:200]
|
||
if response.status_code == 404:
|
||
logger.error(
|
||
"Gitea API 404: %s — репозиторий не найден или нет доступа (проверьте owner/repo и токен API)",
|
||
api_url,
|
||
)
|
||
else:
|
||
logger.error("Ошибка API Gitea: %s — %s", response.status_code, body_preview)
|
||
self.refresh_check_cache(None)
|
||
return None
|
||
|
||
releases = response.json()
|
||
if not releases:
|
||
logger.info("Релизов не найдено")
|
||
self.refresh_check_cache(None)
|
||
return None
|
||
|
||
# Берем последний релиз (первый в списке)
|
||
latest_release = releases[0]
|
||
latest_version = latest_release.get('tag_name', '').lstrip('v')
|
||
|
||
logger.info(f"Последний релиз: {latest_version}, текущая версия: {self.current_version}")
|
||
|
||
# Сравнение версий
|
||
if self._compare_versions(latest_version, self.current_version) > 0:
|
||
logger.info(f"✨ Найдено обновление: {latest_version}")
|
||
info = {
|
||
'version': latest_version,
|
||
'tag_name': latest_release.get('tag_name', ''),
|
||
'name': latest_release.get('name', ''),
|
||
'body': latest_release.get('body', ''),
|
||
'published_at': latest_release.get('published_at', ''),
|
||
'assets': latest_release.get('assets', [])
|
||
}
|
||
self.refresh_check_cache(info)
|
||
return info
|
||
else:
|
||
logger.info("Установлена актуальная версия")
|
||
self.refresh_check_cache(None)
|
||
return None
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
logger.error(f"Ошибка сети при проверке обновлений: {e}")
|
||
self.refresh_check_cache(None)
|
||
return None
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при проверке обновлений: {e}", exc_info=True)
|
||
self.refresh_check_cache(None)
|
||
return None
|
||
|
||
def _compare_versions(self, v1: str, v2: str) -> int:
|
||
"""
|
||
Сравнение версий (например, "1.3.1" vs "1.3.0")
|
||
Возвращает: 1 если v1 > v2, -1 если v1 < v2, 0 если равны
|
||
"""
|
||
try:
|
||
parts1 = [int(x) for x in v1.split('.')]
|
||
parts2 = [int(x) for x in v2.split('.')]
|
||
|
||
# Дополняем нулями до одинаковой длины
|
||
max_len = max(len(parts1), len(parts2))
|
||
parts1.extend([0] * (max_len - len(parts1)))
|
||
parts2.extend([0] * (max_len - len(parts2)))
|
||
|
||
for p1, p2 in zip(parts1, parts2):
|
||
if p1 > p2:
|
||
return 1
|
||
elif p1 < p2:
|
||
return -1
|
||
return 0
|
||
except Exception:
|
||
# Fallback: строковое сравнение
|
||
return 1 if v1 > v2 else (-1 if v1 < v2 else 0)
|
||
|
||
def _fix_download_url_host(self, url: str) -> str:
|
||
"""Подменить хост browser_download_url на gitea_url из настроек (ROOT_URL ≠ API URL)."""
|
||
configured = (self.gitea_url or "").strip().rstrip("/")
|
||
if not configured or not url:
|
||
return url
|
||
cfg = urlparse(configured)
|
||
cur = urlparse(url)
|
||
if not cfg.netloc or cfg.netloc == cur.netloc:
|
||
return url
|
||
return urlunparse(
|
||
(cfg.scheme or cur.scheme, cfg.netloc, cur.path, cur.params, cur.query, cur.fragment)
|
||
)
|
||
|
||
def _asset_download_urls(self, zip_asset: Dict) -> List[str]:
|
||
"""Кандидаты URL для скачивания ZIP (API + browser, с исправленным хостом)."""
|
||
urls: List[str] = []
|
||
seen: set = set()
|
||
asset_id = zip_asset.get("id")
|
||
base = (self.gitea_url or "").rstrip("/")
|
||
if asset_id and base:
|
||
urls.append(
|
||
f"{base}/api/v1/repos/{self.gitea_owner}/{self.gitea_repo}/releases/assets/{asset_id}"
|
||
)
|
||
browser = (zip_asset.get("browser_download_url") or "").strip()
|
||
if browser:
|
||
urls.append(browser)
|
||
fixed = self._fix_download_url_host(browser)
|
||
if fixed != browser:
|
||
urls.append(fixed)
|
||
out: List[str] = []
|
||
for u in urls:
|
||
if u and u not in seen:
|
||
seen.add(u)
|
||
out.append(u)
|
||
return out
|
||
|
||
def download_update(self, release_info: Dict) -> Optional[str]:
|
||
"""
|
||
Скачивание архива обновления
|
||
Возвращает путь к скачанному файлу или None
|
||
"""
|
||
try:
|
||
# Ищем ZIP архив в assets релиза
|
||
zip_asset = None
|
||
for asset in release_info.get('assets', []):
|
||
if asset.get('name', '').endswith('.zip'):
|
||
zip_asset = asset
|
||
break
|
||
|
||
if not zip_asset:
|
||
logger.error("ZIP архив не найден в релизе")
|
||
return None
|
||
|
||
download_urls = self._asset_download_urls(zip_asset)
|
||
if not download_urls:
|
||
logger.error("Не удалось определить URL для скачивания ZIP")
|
||
return None
|
||
|
||
temp_dir = os.path.join(self.base_dir, 'temp_updates')
|
||
os.makedirs(temp_dir, exist_ok=True)
|
||
|
||
zip_path = os.path.join(temp_dir, f"update_{release_info['version']}.zip")
|
||
target_version = str(release_info.get("version") or "")
|
||
|
||
response = None
|
||
total_urls = len(download_urls)
|
||
for idx, download_url in enumerate(download_urls, start=1):
|
||
self._set_update_progress(
|
||
"download",
|
||
"Скачивание обновления…",
|
||
12 + int(3 * idx),
|
||
detail=f"Способ {idx} из {total_urls}",
|
||
target_version=target_version,
|
||
)
|
||
logger.info(f"Скачивание обновления: {download_url}")
|
||
try:
|
||
resp = requests.get(
|
||
download_url,
|
||
headers=self._get_headers(),
|
||
timeout=300,
|
||
stream=True,
|
||
allow_redirects=True,
|
||
)
|
||
except requests.exceptions.RequestException as exc:
|
||
logger.warning("Ошибка сети при скачивании %s: %s", download_url, exc)
|
||
continue
|
||
if resp.status_code == 200:
|
||
response = resp
|
||
break
|
||
logger.warning("Скачивание %s: HTTP %s", download_url, resp.status_code)
|
||
|
||
if response is None:
|
||
logger.error(
|
||
"Не удалось скачать обновление (пробовали %s URL). "
|
||
"Проверьте Gitea URL в админке и доступность ZIP в релизе.",
|
||
len(download_urls),
|
||
)
|
||
return None
|
||
|
||
total_bytes = int(response.headers.get("content-length") or 0)
|
||
downloaded = 0
|
||
with open(zip_path, "wb") as f:
|
||
for chunk in response.iter_content(chunk_size=8192):
|
||
if not chunk:
|
||
continue
|
||
f.write(chunk)
|
||
downloaded += len(chunk)
|
||
if total_bytes > 0:
|
||
frac = downloaded / total_bytes
|
||
pct = 18 + int(27 * frac)
|
||
detail = f"{downloaded // 1024} КБ из {total_bytes // 1024} КБ"
|
||
else:
|
||
pct = 18 + min(27, downloaded // (512 * 1024))
|
||
detail = f"Загружено {downloaded // 1024} КБ"
|
||
self._set_update_progress(
|
||
"download",
|
||
"Скачивание обновления…",
|
||
pct,
|
||
detail=detail,
|
||
target_version=target_version,
|
||
)
|
||
|
||
logger.info(f"Обновление скачано: {zip_path}")
|
||
return zip_path
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при скачивании обновления: {e}", exc_info=True)
|
||
return None
|
||
|
||
def create_backup(self) -> Optional[str]:
|
||
"""Создание резервной копии перед обновлением"""
|
||
try:
|
||
backups_dir = os.path.join(self.base_dir, 'backups')
|
||
os.makedirs(backups_dir, exist_ok=True)
|
||
|
||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||
backup_name = f"backup_{self.current_version}_{timestamp}"
|
||
backup_path = os.path.join(backups_dir, backup_name)
|
||
|
||
logger.info(f"Создание резервной копии: {backup_path}")
|
||
self._set_update_progress("backup", "Создание резервной копии…", 8)
|
||
|
||
# Копируем все файлы, кроме исключенных
|
||
def ignore_func(src, names):
|
||
ignored = []
|
||
for name in names:
|
||
# Исключаем папки
|
||
if name in ['__pycache__', '.git', 'backups', 'temp_updates', '.secret']:
|
||
ignored.append(name)
|
||
# Исключаем файлы БД
|
||
elif name.endswith('.db') or name.endswith('.db-shm') or name.endswith('.db-wal'):
|
||
ignored.append(name)
|
||
return ignored
|
||
|
||
shutil.copytree(
|
||
self.base_dir,
|
||
backup_path,
|
||
ignore=ignore_func
|
||
)
|
||
|
||
logger.info(f"Резервная копия создана: {backup_path}")
|
||
self._set_update_progress("backup", "Резервная копия создана", 15)
|
||
return backup_path
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка создания резервной копии: {e}", exc_info=True)
|
||
return None
|
||
|
||
def _should_protect_file(self, file_path: str) -> bool:
|
||
"""Проверка, нужно ли защищать файл от обновления"""
|
||
filename = os.path.basename(file_path)
|
||
norm_path = file_path.replace("\\", "/")
|
||
try:
|
||
rel = os.path.relpath(file_path, self.base_dir).replace("\\", "/")
|
||
except ValueError:
|
||
rel = norm_path
|
||
|
||
if rel in self.protected_files or filename in self.protected_files:
|
||
return True
|
||
|
||
for pattern in self.protected_files:
|
||
if pattern.startswith("*."):
|
||
if filename.endswith(pattern[1:]):
|
||
return True
|
||
elif pattern in (rel, norm_path):
|
||
return True
|
||
|
||
if filename.endswith((".db", ".db-shm", ".db-wal")):
|
||
return True
|
||
if rel.startswith("data/") and filename.endswith((".db", ".db-shm", ".db-wal")):
|
||
return True
|
||
|
||
for folder in self.protected_folders:
|
||
if folder.startswith("*."):
|
||
continue
|
||
parts = norm_path.split("/")
|
||
if folder in parts:
|
||
return True
|
||
|
||
return False
|
||
|
||
def _restore_from_backup(self, backup_path: Optional[str]) -> None:
|
||
if not backup_path or not os.path.isdir(backup_path):
|
||
return
|
||
logger.warning("Восстановление файлов из резервной копии: %s", backup_path)
|
||
try:
|
||
for root, dirs, files in os.walk(backup_path):
|
||
dirs[:] = [
|
||
d
|
||
for d in dirs
|
||
if d not in ("__pycache__", ".git", "backups", "temp_updates", ".secret")
|
||
]
|
||
for file in files:
|
||
src = os.path.join(root, file)
|
||
rel = os.path.relpath(src, backup_path)
|
||
dst = os.path.join(self.base_dir, rel)
|
||
if self._should_protect_file(dst):
|
||
continue
|
||
os.makedirs(os.path.dirname(dst), exist_ok=True)
|
||
shutil.copy2(src, dst)
|
||
logger.info("Восстановление из резервной копии завершено")
|
||
except Exception as exc:
|
||
logger.error("Ошибка восстановления из резервной копии: %s", exc, exc_info=True)
|
||
|
||
def _write_pending_restart(
|
||
self,
|
||
backup_path: str,
|
||
target_version: str,
|
||
previous_version: str,
|
||
) -> None:
|
||
listen = os.environ.get("WESP_LISTEN", "0.0.0.0:80")
|
||
host_port = listen.split(":")[-1] if ":" in listen else "80"
|
||
health_url = os.environ.get(
|
||
"WESP_HEALTH_URL",
|
||
f"http://127.0.0.1:{host_port}/api/health",
|
||
)
|
||
payload = {
|
||
"backup_path": backup_path,
|
||
"target_version": target_version,
|
||
"previous_version": previous_version,
|
||
"health_url": health_url,
|
||
"listen": listen,
|
||
}
|
||
path = os.path.join(self.base_dir, "data", ".pending_restart.json")
|
||
os.makedirs(os.path.dirname(path), exist_ok=True)
|
||
with open(path, "w", encoding="utf-8") as f:
|
||
json.dump(payload, f, ensure_ascii=False, indent=2)
|
||
f.write("\n")
|
||
|
||
def _install_deps_and_verify(self, backup_path: str, target_version: str) -> bool:
|
||
import importlib
|
||
from pathlib import Path
|
||
|
||
import install_deps
|
||
|
||
importlib.reload(install_deps)
|
||
|
||
from app.services.update_state_store import mark_failed
|
||
from app.services.update_verify import (
|
||
install_requirements_from_package,
|
||
run_smoke_test,
|
||
wheelhouse_covers_requirements,
|
||
)
|
||
|
||
req = install_deps.production_requirements_path(self.base_dir)
|
||
wheels = Path(self.base_dir) / "vendor" / "wheels"
|
||
ok, missing = wheelhouse_covers_requirements(req, wheels)
|
||
if not ok:
|
||
detail = ", ".join(missing[:8])
|
||
msg = "В пакете нет зависимостей"
|
||
logger.error("%s: %s", msg, detail)
|
||
self._set_update_progress("error", msg, 0, detail=detail, target_version=target_version)
|
||
mark_failed(self.base_dir, message=msg, stage="deps", detail=detail)
|
||
self._restore_from_backup(backup_path)
|
||
return False
|
||
|
||
self._set_update_progress(
|
||
"deps",
|
||
"Установка зависимостей из пакета…",
|
||
88,
|
||
target_version=target_version,
|
||
)
|
||
py = install_deps.install_python_executable(self.base_dir)
|
||
logger.info("OTA pip: %s", py)
|
||
try:
|
||
install_requirements_from_package(self.base_dir)
|
||
except install_deps.UpdateDepsError as exc:
|
||
msg = "Не удалось установить зависимости из пакета"
|
||
logger.error("%s: %s", msg, exc)
|
||
self._set_update_progress("error", msg, 0, detail=str(exc), target_version=target_version)
|
||
mark_failed(self.base_dir, message=msg, stage="deps", detail=str(exc))
|
||
self._restore_from_backup(backup_path)
|
||
return False
|
||
|
||
self._set_update_progress(
|
||
"verify",
|
||
"Проверка запуска…",
|
||
94,
|
||
target_version=target_version,
|
||
)
|
||
from app.services.update_verify import run_smoke_test
|
||
|
||
smoke_ok, smoke_msg = run_smoke_test(
|
||
self.base_dir,
|
||
python_exe=install_deps.install_python_executable(self.base_dir),
|
||
)
|
||
if not smoke_ok:
|
||
msg = "Проверка запуска не пройдена"
|
||
logger.error("%s: %s", msg, smoke_msg)
|
||
self._set_update_progress("error", msg, 0, detail=smoke_msg[:200], target_version=target_version)
|
||
mark_failed(self.base_dir, message=msg, stage="verify", detail=smoke_msg[:200])
|
||
self._restore_from_backup(backup_path)
|
||
return False
|
||
return True
|
||
|
||
def _run_restart_cmd(self) -> None:
|
||
cmd = (self.restart_cmd or "").strip()
|
||
if not cmd:
|
||
default_script = os.path.join(self.base_dir, "scripts", "post_update.sh")
|
||
if os.path.isfile(default_script):
|
||
cmd = f'bash "{default_script}"'
|
||
if not cmd:
|
||
logger.info("WESP_UPDATE_RESTART_CMD не задан — перезапуск вручную")
|
||
self._set_update_progress("restart", "Обновление завершено — перезапустите сервер вручную", 98)
|
||
return
|
||
try:
|
||
parts = shlex.split(cmd)
|
||
if not parts:
|
||
return
|
||
self._set_update_progress("restart", "Перезапуск системы…", 98)
|
||
logger.info("Перезапуск после обновления")
|
||
subprocess.Popen(parts, close_fds=True)
|
||
except Exception as exc:
|
||
logger.error("Не удалось выполнить команду перезапуска: %s", exc, exc_info=True)
|
||
self._set_update_progress("restart", "Не удалось выполнить перезапуск автоматически", 98)
|
||
|
||
def _merge_config_json(self, old_config: Dict, new_config: Dict) -> Dict:
|
||
"""
|
||
Умное слияние config.json
|
||
Сохраняет пользовательские настройки, обновляет версию
|
||
"""
|
||
merged = old_config.copy()
|
||
|
||
# Обновляем версию
|
||
if 'version' in new_config:
|
||
merged['version'] = new_config['version']
|
||
|
||
# Добавляем новые поля из обновления (если их нет)
|
||
for key, value in new_config.items():
|
||
if key not in merged:
|
||
merged[key] = value
|
||
elif isinstance(value, dict) and isinstance(merged.get(key), dict):
|
||
# Рекурсивно сливаем вложенные словари
|
||
merged[key] = self._merge_config_json(merged[key], value)
|
||
|
||
# Сохраняем пользовательские настройки (не перезаписываем)
|
||
user_keys = [
|
||
'server_url', 'client_id', 'updated_at', 'first_bootstrap_done',
|
||
'pinned_server_id', 'gitea', 'auto_update', 'sync_compression',
|
||
'version_constraints'
|
||
]
|
||
|
||
for key in user_keys:
|
||
if key in old_config and key not in ['version']: # version обновляем
|
||
merged[key] = old_config[key]
|
||
|
||
return merged
|
||
|
||
def apply_update(self, zip_path: str, release_info: Dict) -> bool:
|
||
"""
|
||
Применение обновления
|
||
Распаковывает архив и заменяет файлы с защитой данных
|
||
"""
|
||
try:
|
||
target_version = str(release_info.get("version") or "")
|
||
logger.info(f"Применение обновления {target_version}")
|
||
self._set_update_progress(
|
||
"apply",
|
||
f"Применение обновления {target_version}…",
|
||
48,
|
||
target_version=target_version,
|
||
)
|
||
|
||
# Создаем временную папку для распаковки
|
||
extract_dir = os.path.join(self.base_dir, 'temp_updates', 'extract')
|
||
if os.path.exists(extract_dir):
|
||
shutil.rmtree(extract_dir)
|
||
os.makedirs(extract_dir, exist_ok=True)
|
||
|
||
# Распаковываем архив
|
||
logger.info(f"Распаковка архива...")
|
||
self._set_update_progress(
|
||
"extract",
|
||
"Распаковка архива…",
|
||
52,
|
||
target_version=target_version,
|
||
)
|
||
with zipfile.ZipFile(zip_path, 'r') as zip_ref:
|
||
zip_ref.extractall(extract_dir)
|
||
|
||
# Находим корневую папку проекта в архиве (может быть вложена)
|
||
project_root = extract_dir
|
||
for root, dirs, files in os.walk(extract_dir):
|
||
if "run.py" in files or "proga.py" in files or "config.json" in files:
|
||
project_root = root
|
||
break
|
||
|
||
# Загружаем старый config.json для слияния
|
||
old_config = self.load_config()
|
||
new_config = None
|
||
|
||
# Копируем файлы, исключая защищенные
|
||
logger.info(f"Копирование файлов...")
|
||
copied_count = 0
|
||
skipped_count = 0
|
||
protected_count = 0
|
||
pending_copies: List[Tuple[str, str, str]] = []
|
||
|
||
for root, dirs, files in os.walk(project_root):
|
||
dirs[:] = [d for d in dirs if not any(
|
||
pattern in d for pattern in ['__pycache__', '.git', 'backups', 'temp_updates', '.secret']
|
||
)]
|
||
|
||
for file in files:
|
||
src_path = os.path.join(root, file)
|
||
rel_path = os.path.relpath(src_path, project_root)
|
||
dst_path = os.path.join(self.base_dir, rel_path)
|
||
if file == "config.json":
|
||
dst_path = self.config_file
|
||
|
||
if self._should_protect_file(dst_path):
|
||
protected_count += 1
|
||
logger.debug(f"Защищен от обновления: {rel_path}")
|
||
continue
|
||
|
||
if file == 'config.json' and os.path.exists(dst_path):
|
||
try:
|
||
with open(src_path, 'r', encoding='utf-8') as f:
|
||
new_config = json.load(f)
|
||
except Exception:
|
||
pass
|
||
continue
|
||
|
||
pending_copies.append((src_path, dst_path, rel_path))
|
||
|
||
total_files = len(pending_copies)
|
||
self._set_update_progress(
|
||
"install",
|
||
"Копирование файлов…",
|
||
55,
|
||
detail=f"0 из {total_files}" if total_files else None,
|
||
target_version=target_version,
|
||
)
|
||
|
||
for idx, (src_path, dst_path, rel_path) in enumerate(pending_copies, start=1):
|
||
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
|
||
shutil.copy2(src_path, dst_path)
|
||
copied_count += 1
|
||
if idx == total_files or idx % 25 == 0:
|
||
pct = 55 + int(33 * idx / max(total_files, 1))
|
||
self._set_update_progress(
|
||
"install",
|
||
"Копирование файлов…",
|
||
pct,
|
||
detail=f"Скопировано {idx} из {total_files}",
|
||
target_version=target_version,
|
||
)
|
||
|
||
# Применяем умное слияние config.json
|
||
if new_config:
|
||
logger.info(f"Слияние config.json...")
|
||
self._set_update_progress(
|
||
"config",
|
||
"Обновление конфигурации…",
|
||
92,
|
||
target_version=target_version,
|
||
)
|
||
merged_config = self._merge_config_json(old_config, new_config)
|
||
self.save_config(merged_config)
|
||
logger.info(f"config.json обновлен с сохранением пользовательских настроек")
|
||
|
||
logger.info(f"Обновление применено: скопировано {copied_count} файлов, пропущено {skipped_count}, защищено {protected_count}")
|
||
self._set_update_progress(
|
||
"done",
|
||
f"Обновление до версии {target_version} применено",
|
||
96,
|
||
detail=f"Скопировано {copied_count} файлов, защищено {protected_count}",
|
||
target_version=target_version,
|
||
)
|
||
|
||
# Обновляем версию в памяти
|
||
self.current_version = release_info['version']
|
||
|
||
# Очищаем временные файлы
|
||
try:
|
||
shutil.rmtree(extract_dir)
|
||
os.remove(zip_path)
|
||
except:
|
||
pass
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при применении обновления: {e}", exc_info=True)
|
||
return False
|
||
|
||
def update(self, release_info: Optional[Dict] = None) -> bool:
|
||
"""
|
||
Полный процесс обновления
|
||
"""
|
||
if self.is_updating:
|
||
logger.warning("Обновление уже выполняется")
|
||
return False
|
||
|
||
if not self.enabled:
|
||
logger.warning("Автообновление отключено")
|
||
return False
|
||
|
||
self.is_updating = True
|
||
target_version = str((release_info or {}).get("version") or "")
|
||
|
||
try:
|
||
self._set_update_progress(
|
||
"prepare",
|
||
"Подготовка к обновлению…",
|
||
3,
|
||
target_version=target_version or None,
|
||
)
|
||
if not release_info:
|
||
release_info = self.check_for_updates()
|
||
if not release_info:
|
||
logger.info("Обновлений не найдено")
|
||
return False
|
||
target_version = str(release_info.get("version") or "")
|
||
self._set_update_progress(
|
||
"prepare",
|
||
"Подготовка к обновлению…",
|
||
3,
|
||
target_version=target_version,
|
||
)
|
||
|
||
backup_path = self.create_backup()
|
||
if not backup_path:
|
||
logger.error("Не удалось создать резервную копию")
|
||
self._set_update_progress("error", "Не удалось создать резервную копию", 0, target_version=target_version)
|
||
return False
|
||
|
||
previous_version = self.current_version
|
||
try:
|
||
from app.services.update_state_store import mark_in_progress
|
||
|
||
mark_in_progress(
|
||
self.base_dir,
|
||
target_version=target_version,
|
||
previous_version=previous_version,
|
||
backup_path=backup_path,
|
||
)
|
||
except Exception as exc:
|
||
logger.debug("update_state mark_in_progress: %s", exc)
|
||
|
||
zip_path = self.download_update(release_info)
|
||
if not zip_path:
|
||
logger.error("Не удалось скачать обновление")
|
||
self._set_update_progress("error", "Не удалось скачать обновление", 0, target_version=target_version)
|
||
return False
|
||
|
||
if not self.apply_update(zip_path, release_info):
|
||
logger.error("Не удалось применить обновление")
|
||
self._set_update_progress("error", "Не удалось применить обновление", 0, target_version=target_version)
|
||
self._restore_from_backup(backup_path)
|
||
try:
|
||
from app.services.update_state_store import mark_failed
|
||
|
||
mark_failed(self.base_dir, message="Не удалось применить обновление", stage="install")
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
if not self._install_deps_and_verify(backup_path, target_version):
|
||
return False
|
||
|
||
self.clear_pending_after_install()
|
||
logger.info(f"Обновление до версии {release_info['version']} успешно применено")
|
||
self._write_pending_restart(backup_path, target_version, previous_version)
|
||
try:
|
||
from app.services.update_state_store import mark_pending_restart
|
||
|
||
mark_pending_restart(self.base_dir)
|
||
except Exception:
|
||
pass
|
||
self._set_update_progress(
|
||
"restart",
|
||
"Перезапуск системы…",
|
||
99,
|
||
target_version=target_version,
|
||
)
|
||
self._run_restart_cmd()
|
||
|
||
return True
|
||
|
||
finally:
|
||
self.is_updating = False
|
||
self._clear_update_progress()
|
||
|
||
def _update_loop(self):
|
||
"""Цикл проверки обновлений в фоновом потоке (без автоустановки)."""
|
||
logger.info("Запущен цикл проверки обновлений")
|
||
|
||
while self.is_running:
|
||
try:
|
||
if self.enabled:
|
||
release_info = self.check_for_updates()
|
||
if release_info:
|
||
logger.info(
|
||
"Доступно обновление %s — ожидает выбора пользователя",
|
||
release_info.get("version"),
|
||
)
|
||
|
||
for _ in range(self.check_interval):
|
||
if not self.is_running:
|
||
break
|
||
time.sleep(1)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка в цикле обновлений: {e}", exc_info=True)
|
||
time.sleep(60)
|
||
|
||
logger.info("Цикл проверки обновлений остановлен")
|
||
|
||
def start(self):
|
||
"""Запуск фоновой проверки обновлений"""
|
||
if not self.enabled:
|
||
logger.info("Проверка обновлений отключена в конфигурации")
|
||
return
|
||
|
||
if self.is_running:
|
||
return
|
||
|
||
if not self.gitea_url or not self.gitea_owner or not self.gitea_repo:
|
||
logger.warning("Настройки Gitea не заполнены, проверка обновлений не запущена")
|
||
return
|
||
|
||
self.is_running = True
|
||
self.update_thread = threading.Thread(
|
||
target=self._update_loop, daemon=True, name="wesp-update-checker"
|
||
)
|
||
self.update_thread.start()
|
||
logger.info("Фоновая проверка обновлений запущена")
|
||
|
||
def stop(self):
|
||
"""Остановка фоновой проверки"""
|
||
self.is_running = False
|
||
if self.update_thread:
|
||
self.update_thread.join(timeout=5)
|
||
logger.info("Фоновая проверка обновлений остановлена")
|
||
|
||
|
||
# Глобальный экземпляр
|
||
auto_updater = None
|
||
|
||
|
||
def init_auto_updater(*, reload: bool = False) -> None:
|
||
"""Инициализация / перезагрузка фоновой проверки обновлений."""
|
||
global auto_updater
|
||
|
||
try:
|
||
if reload and auto_updater is not None:
|
||
auto_updater.stop()
|
||
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
config_file = wesp_data_config_path(script_dir)
|
||
legacy_root = os.path.join(script_dir, "config.json")
|
||
if not os.path.exists(config_file) and os.path.exists(legacy_root):
|
||
try:
|
||
shutil.copy2(legacy_root, config_file)
|
||
except OSError:
|
||
pass
|
||
|
||
auto_updater = AutoUpdater()
|
||
if auto_updater.enabled and auto_updater.gitea_url and auto_updater.gitea_owner and auto_updater.gitea_repo:
|
||
auto_updater.start()
|
||
logger.info("Фоновая проверка обновлений запущена")
|
||
elif auto_updater.enabled:
|
||
logger.warning(
|
||
"Проверка включена, но Gitea не полностью настроен (url/owner/repo)"
|
||
)
|
||
else:
|
||
logger.info(
|
||
"Проверка обновлений выключена — включите «Проверять обновления» в админке и нажмите «Сохранить»"
|
||
)
|
||
except Exception as e:
|
||
logger.error("Ошибка инициализации проверки обновлений: %s", e, exc_info=True)
|
||
|
||
|
||
def stop_auto_updater():
|
||
"""Остановка фоновой проверки обновлений"""
|
||
global auto_updater
|
||
|
||
if auto_updater:
|
||
auto_updater.stop()
|
||
logger.info("Проверка обновлений остановлена")
|
||
auto_updater = None
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
if sys.platform == "win32":
|
||
import codecs
|
||
sys.stdout = codecs.getwriter("utf-8")(sys.stdout.detach())
|
||
|
||
print("Тестирование автообновления")
|
||
updater = AutoUpdater()
|
||
|
||
if updater.enabled:
|
||
release_info = updater.check_for_updates()
|
||
if release_info:
|
||
print(f"Найдено обновление: {release_info['version']}")
|
||
print(f"Название: {release_info['name']}")
|
||
print(f"Описание: {release_info['body'][:100]}...")
|
||
else:
|
||
print("Обновлений не найдено")
|
||
else:
|
||
print("Автообновление отключено")
|