#!/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