ComConfigCopy/update_checker.py

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

import logging
import requests
import threading
import re
from packaging import version
import xml.etree.ElementTree as ET
import html

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

class ReleaseType:
    """Типы релизов"""
    STABLE = "stable"
    PRERELEASE = "prerelease"

class UpdateChecker:
    """Класс для проверки обновлений программы"""
    
    def __init__(self, current_version, repo_url, include_prereleases=False):
        self.current_version = current_version
        self.repo_url = repo_url
        self.include_prereleases = include_prereleases
        self.rss_url = f"{repo_url}/releases.rss"
        self.release_info = None

    def _clean_html(self, html_text):
        """Очищает HTML-разметку и форматирует текст"""
        if not html_text:
            return ""
        text = re.sub(r'<[^>]+>', '', html_text)
        text = html.unescape(text)
        text = re.sub(r'\n\s*\n', '\n\n', text)
        return '\n'.join(line.strip() for line in text.splitlines()).strip()

    def _parse_release_info(self, item):
        """Извлекает информацию о релизе из RSS item"""
        title = item.find('title').text if item.find('title') is not None else ''
        link = item.find('link').text if item.find('link') is not None else ''
        description = item.find('description').text if item.find('description') is not None else ''
        content = item.find('{http://purl.org/rss/1.0/modules/content/}encoded')
        content_text = content.text if content is not None else ''
        
        # Извлекаем версию и проверяем тип релиза из тега
        version_match = re.search(r'/releases/tag/(?:pre-)?v?(\d+\.\d+(?:\.\d+)?)', link)
        if not version_match:
            return None
            
        version_str = version_match.group(1)
        # Проверяем наличие префикса pre- в теге
        is_prerelease = 'pre-' in link.lower()
        
        # Форматируем название релиза
        formatted_title = title
        if title == version_str or not title.strip():
            # Если заголовок пустой или совпадает с версией, создаем стандартное название
            release_type = "Пре-релиз" if is_prerelease else "Версия"
            formatted_title = f"{release_type} {version_str}"
        elif not re.search(version_str, title):
            # Если версия не указана в заголовке, добавляем её
            formatted_title = f"{title} ({version_str})"
            
        # Форматируем описание
        formatted_description = self._clean_html(content_text or description)
        if not formatted_description.strip():
            formatted_description = "Нет описания"
            
        # Добавляем метку типа релиза в начало описания
        release_type_label = "[Пре-релиз] " if is_prerelease else ""
        formatted_description = f"{release_type_label}{formatted_description}"
        
        return {
            'title': formatted_title,
            'link': link,
            'description': formatted_description,
            'version': version_str,
            'type': ReleaseType.PRERELEASE if is_prerelease else ReleaseType.STABLE
        }

    def check_updates(self, callback=None):
        """Проверяет наличие обновлений в асинхронном режиме"""
        def check_worker():
            try:
                logging.info(f"Текущая версия программы: {self.current_version}")
                logging.info(f"Проверка пре-релизов: {self.include_prereleases}")
                logging.info(f"Запрос RSS ленты: {self.rss_url}")
                
                response = requests.get(self.rss_url, timeout=10)
                response.raise_for_status()
                
                root = ET.fromstring(response.content)
                items = root.findall('.//item')
                if not items:
                    raise UpdateCheckError("Релизы не найдены")
                
                logging.info(f"Найдено {len(items)} релизов")
                
                latest_version = None
                latest_info = None
                
                for item in items:
                    release_info = self._parse_release_info(item)
                    if not release_info:
                        continue
                        
                    is_prerelease = release_info['type'] == ReleaseType.PRERELEASE
                    logging.info(
                        f"Проверка релиза: {release_info['title']}, "
                        f"версия: {release_info['version']}, "
                        f"тип: {'пре-релиз' if is_prerelease else 'стабильная'}"
                    )
                    
                    # Пропускаем пре-релизы если они не включены
                    if is_prerelease and not self.include_prereleases:
                        logging.info(f"Пропуск пре-релиза: {release_info['version']}")
                        continue
                    
                    # Сравниваем версии
                    try:
                        current_ver = version.parse(latest_version or "0.0.0")
                        new_ver = version.parse(release_info['version'].split('-')[0])  # Убираем суффикс для сравнения
                        
                        if new_ver > current_ver:
                            latest_version = release_info['version']
                            latest_info = release_info
                            logging.info(f"Новая версия: {latest_version}")
                            
                    except version.InvalidVersion as e:
                        logging.warning(f"Некорректный формат версии {release_info['version']}: {e}")
                        continue
                
                if not latest_info:
                    raise UpdateCheckError("Не найдены подходящие версии")
                
                self.release_info = latest_info
                
                # Сравниваем с текущей версией
                current_ver = version.parse(self.current_version)
                latest_ver = version.parse(latest_version.split('-')[0])
                update_available = latest_ver > current_ver
                
                logging.info(f"Сравнение версий: текущая {current_ver} <-> последняя {latest_ver}")
                logging.info(f"Доступно обновление: {update_available}")
                
                if callback:
                    callback(update_available, None)
                    
            except UpdateCheckError as e:
                logging.error(str(e))
                if callback:
                    callback(False, str(e))
            except Exception as e:
                logging.error(f"Ошибка при проверке обновлений: {e}", exc_info=True)
                if callback:
                    callback(False, str(e))

        threading.Thread(target=check_worker, daemon=True).start()
    
    def get_release_notes(self):
        """Возвращает информацию о последнем релизе"""
        return self.release_info