ComConfigCopy/update_checker.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import logging
import requests
import threading
from packaging import version

class UpdateCheckError(Exception):
    """Исключение для ошибок проверки обновлений"""
    pass

class UpdateChecker:
    """Класс для проверки обновлений программы"""
    
    def __init__(self, current_version, repo_url):
        self.current_version = current_version
        self.repo_url = repo_url
        # Формируем базовый URL API
        self.api_url = repo_url.replace("gitea.filow.ru", "gitea.filow.ru/api/v1/repos/LowaSC/ComConfigCopy")
        self._update_available = False
        self._latest_version = None
        self._latest_release = None
        self._error = None
        self._changelog = None
    
    def get_changelog(self, callback=None):
        """
        Получение changelog из репозитория.
        :param callback: Функция обратного вызова, которая будет вызвана после получения changelog
        """
        def fetch():
            try:
                # Пытаемся получить CHANGELOG.md из репозитория
                response = requests.get(f"{self.api_url}/contents/CHANGELOG.md", timeout=10)
                response.raise_for_status()
                
                content = response.json()
                if "content" in content:
                    import base64
                    changelog_content = base64.b64decode(content["content"]).decode("utf-8")
                    self._changelog = changelog_content
                    self._error = None
                else:
                    raise UpdateCheckError("Не удалось получить содержимое CHANGELOG.md")
                    
            except requests.RequestException as e:
                error_msg = f"Ошибка получения changelog: {e}"
                logging.error(error_msg, exc_info=True)
                self._error = error_msg
                self._changelog = None
            except Exception as e:
                error_msg = f"Неизвестная ошибка при получении changelog: {e}"
                logging.error(error_msg, exc_info=True)
                self._error = error_msg
                self._changelog = None
            finally:
                if callback:
                    callback(self._changelog, self._error)
        
        # Запускаем получение в отдельном потоке
        threading.Thread(target=fetch, daemon=True).start()
    
    def check_updates(self, callback=None):
        """
        Проверка наличия обновлений.
        :param callback: Функция обратного вызова, которая будет вызвана после проверки
        """
        def check():
            try:
                response = requests.get(f"{self.api_url}/releases", timeout=10)
                response.raise_for_status()
                
                releases = response.json()
                if not releases:
                    raise UpdateCheckError("Не найдено релизов в репозитории")
                
                latest_release = releases[0]
                latest_version = latest_release.get("tag_name", "").lstrip("v")
                
                if not latest_version:
                    raise UpdateCheckError("Не удалось определить версию последнего релиза")
                
                try:
                    if version.parse(latest_version) > version.parse(self.current_version):
                        self._update_available = True
                        self._latest_version = latest_version
                        self._latest_release = latest_release
                        logging.info(f"Доступно обновление: {latest_version}")
                    else:
                        logging.info("Обновления не требуются")
                except version.InvalidVersion as e:
                    raise UpdateCheckError(f"Некорректный формат версии: {e}")
                    
                self._error = None
                    
            except requests.RequestException as e:
                error_msg = f"Ошибка сетевого подключения: {e}"
                logging.error(error_msg, exc_info=True)
                self._error = error_msg
            except UpdateCheckError as e:
                logging.error(str(e), exc_info=True)
                self._error = str(e)
            except Exception as e:
                error_msg = f"Неизвестная ошибка при проверке обновлений: {e}"
                logging.error(error_msg, exc_info=True)
                self._error = error_msg
            finally:
                if callback:
                    callback(self._update_available, self._error)
    
    @property
    def update_available(self):
        """Доступно ли обновление"""
        return self._update_available
    
    @property
    def latest_version(self):
        """Последняя доступная версия"""
        return self._latest_version
    
    @property
    def error(self):
        """Последняя ошибка при проверке обновлений"""
        return self._error
    
    @property
    def changelog(self):
        """Текущий changelog"""
        return self._changelog
    
    def get_release_notes(self):
        """Получение информации о последнем релизе"""
        if self._latest_release:
            return {
                "version": self._latest_version,
                "description": self._latest_release.get("body", ""),
                "download_url": self._latest_release.get("assets", [{}])[0].get("browser_download_url", "")
            }
        return None

    def get_releases(self, callback=None):
        """
        Получение списка релизов из репозитория.
        :param callback: Функция обратного вызова, которая будет вызвана после получения списка релизов
        """
        def fetch():
            try:
                response = requests.get(f"{self.api_url}/releases", timeout=10)
                response.raise_for_status()
                releases = response.json()
                
                if not releases:
                    raise UpdateCheckError("Не найдено релизов в репозитории")
                
                self._error = None
                if callback:
                    callback(releases, None)
                    
            except requests.RequestException as e:
                error_msg = f"Ошибка сетевого подключения: {e}"
                logging.error(error_msg, exc_info=True)
                self._error = error_msg
                if callback:
                    callback(None, error_msg)
            except Exception as e:
                error_msg = f"Ошибка при получении списка релизов: {e}"
                logging.error(error_msg, exc_info=True)
                self._error = error_msg
                if callback:
                    callback(None, error_msg)
        
        # Запускаем получение в отдельном потоке
        threading.Thread(target=fetch, daemon=True).start()