HoDA_Radio/app/core/update_manager.py
2025-09-21 20:13:54 +02:00

196 lines
8.6 KiB
Python

import requests
from packaging import version
from PyQt6.QtWidgets import QApplication
from PyQt6.QtWidgets import QFileDialog, QDialog, QVBoxLayout, QTextEdit, QPushButton, QHBoxLayout, QLabel
from app.core.alert_manager import AlertManager
from app.core.settings_manager import SettingsManager
from app.core.language_manager import LanguageManager
from app.ui.widgets.loading_bar import LoadingBar
import os
import sys
import subprocess
from typing import List, Dict
class UpdateManager:
def __init__(self, settings_manager: SettingsManager, language_manager: LanguageManager, alert_manager: AlertManager) -> None:
self.settings_manager = settings_manager
self.language_manager = language_manager
self.alert_manager = alert_manager
self.repo_url = self.settings_manager.get_config("git_repo")
self.app_name = self.settings_manager.get_config("app_name").replace(" ","_")
self.app_os = self.settings_manager.get_config("app_os")
self.arch = self.settings_manager.get_config("architecture")
def get_releases_with_asset(self) -> List[Dict]:
"""
Retourne la liste des releases (dict) qui contiennent un asset correspondant
à l'OS/architecture attendus. Chaque dict contient: tag_name, download_url, body.
"""
releases_list: List[Dict] = []
try:
if "gitea" in self.repo_url:
# Gitea: construire URL API (essai basique)
owner_repo = self.repo_url.rstrip("/").split("/")[-2:]
api_base = self.repo_url.replace("/" + owner_repo[0] + "/" + owner_repo[1], "/api/v1/repos/" + owner_repo[0] + "/" + owner_repo[1])
api_url = api_base + "/releases"
resp = requests.get(api_url)
resp.raise_for_status()
releases = resp.json()
else:
owner_repo = self.repo_url.rstrip("/").split("/")[-2:]
api_url = f"https://api.github.com/repos/{owner_repo[0]}/{owner_repo[1]}/releases"
resp = requests.get(api_url)
resp.raise_for_status()
releases = resp.json()
expected_filename_frag = f"{self.app_name}-{self.app_os}-{self.arch}"
for release in releases:
tag = release.get("tag_name") or release.get("name")
body = release.get("body", "") or ""
for asset in release.get("assets", []):
name = asset.get("name", "")
if expected_filename_frag in name:
downloads = asset.get("browser_download_url") or asset.get("url")
releases_list.append({
"tag_name": tag,
"download_url": downloads,
"body": body
})
break
except Exception:
# En cas d'erreur, retourner liste vide (on ne lève pas pour ne pas bloquer l'app)
return []
return releases_list
def check_for_update(self, parent=None) -> bool:
current_version = self.settings_manager.get_config("app_version")
releases = self.get_releases_with_asset()
release = releases[0] if releases else None
if release and version.parse(release["tag_name"]) > version.parse(current_version):
choice = self.show_update_dialog(releases, current_version, parent)
if choice:
folder = QFileDialog.getExistingDirectory(parent, self.language_manager.get_text("choose_update_folder"))
if folder:
return self.download(release["download_url"], release["tag_name"], folder, parent)
return False
def download(self, download_url, version, folder, parent=None):
try:
filename = os.path.basename(download_url).replace(".", "-" +version + ".")
local_path = os.path.join(folder, filename)
resp = requests.get(download_url, stream=True)
total = int(resp.headers.get('content-length', 0))
# Crée une boîte de dialogue avec la barre de chargement
dialog = QDialog(parent)
dialog.setWindowTitle(self.language_manager.get_text("update"))
layout = QVBoxLayout(dialog)
loading_bar = LoadingBar(self.language_manager.get_text("downloading_update"), dialog)
layout.addWidget(loading_bar)
dialog.setModal(True)
# Variable pour tracker si le téléchargement a été annulé
download_cancelled = False
def on_dialog_rejected():
nonlocal download_cancelled
download_cancelled = True
dialog.rejected.connect(on_dialog_rejected)
dialog.show()
downloaded = 0
with open(local_path, "wb") as f:
for chunk in resp.iter_content(chunk_size=8192):
QApplication.processEvents()
if download_cancelled:
f.truncate(0)
f.close()
break
if chunk:
f.write(chunk)
downloaded += len(chunk)
percent = int(downloaded * 100 / total) if total else 0
loading_bar.set_progress(percent)
dialog.close()
if download_cancelled:
os.remove(local_path, dir_fd=None)
self.alert_manager.show_info(self.language_manager.get_text("update_aborted"), parent=parent)
return False
msg = self.language_manager.get_text("update_downloaded").replace("{local_path}", local_path)
self.alert_manager.show_success(msg, parent=parent)
if sys.platform.startswith("win"):
os.startfile(local_path)
else:
subprocess.Popen(["chmod", "+x", local_path])
subprocess.Popen([local_path])
return True
except Exception as e:
self.alert_manager.show_error("update_download_error", parent=parent)
return False
def show_update_dialog(self, releases: List[Dict], current_version: str, parent=None) -> bool:
"""
Affiche une boîte de dialogue avec options Mettre à jour et Détails via l'alert_manager
"""
latest_release = releases[0]
message = self.language_manager.get_text("update_found").replace("{latest_tag}", latest_release["tag_name"])
choice = self.alert_manager.show_choice_with_details(
message,
parent=parent,
details_callback=lambda: self.show_details_dialog(releases, current_version, parent)
)
return choice
def show_details_dialog(self, releases: List[Dict], current_version: str, parent=None) -> None:
"""
Affiche tous les changelogs des versions supérieures à la version actuelle
"""
dialog = QDialog(parent)
dialog.setWindowTitle(self.language_manager.get_text("update_details"))
dialog.setModal(True)
dialog.resize(600, 500)
layout = QVBoxLayout(dialog)
# Zone de texte pour afficher les changelogs
text_edit = QTextEdit()
text_edit.setReadOnly(True)
# Filtrer et trier les releases supérieures à la version actuelle
newer_releases = []
for release in releases:
if version.parse(release["tag_name"]) > version.parse(current_version):
newer_releases.append(release)
# Trier par version décroissante (plus récente en premier)
newer_releases.sort(key=lambda x: version.parse(x["tag_name"]), reverse=True)
# Construire le texte des changelogs
changelog_text = ""
for release in newer_releases:
changelog_text += f"## {self.language_manager.get_text('version')} {release['tag_name']} :\n\n"
body = release['body'].replace('\n','\n### ')
changelog_text += f"### {body}"
changelog_text += "\n\n"
if release != newer_releases[-1]:
changelog_text += "---\n\n"
text_edit.setAcceptRichText(True)
text_edit.setMarkdown(changelog_text)
layout.addWidget(text_edit)
# Bouton Fermer
close_button = QPushButton(self.language_manager.get_text("close"))
close_button.clicked.connect(dialog.close)
layout.addWidget(close_button)
dialog.exec()