424 Zeilen
15 KiB
Python
424 Zeilen
15 KiB
Python
"""
|
|
Update-Checking-Funktionalität für den Social Media Account Generator.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import logging
|
|
import requests
|
|
import shutil
|
|
from datetime import datetime
|
|
from typing import Dict, List, Any, Optional, Tuple, Union
|
|
from licensing.api_client import LicenseAPIClient
|
|
|
|
logger = logging.getLogger("update_checker")
|
|
|
|
class UpdateChecker:
|
|
"""Klasse zum Überprüfen und Herunterladen von Updates."""
|
|
|
|
CONFIG_FILE = os.path.join("config", "app_version.json")
|
|
|
|
def __init__(self, api_client: Optional[LicenseAPIClient] = None):
|
|
"""
|
|
Initialisiert den UpdateChecker und lädt die Konfiguration.
|
|
|
|
Args:
|
|
api_client: Optional vorkonfigurierter API Client
|
|
"""
|
|
self.api_client = api_client or LicenseAPIClient()
|
|
self.version_info = self.load_version_info()
|
|
|
|
# Stelle sicher, dass das Konfigurationsverzeichnis existiert
|
|
os.makedirs(os.path.dirname(self.CONFIG_FILE), exist_ok=True)
|
|
|
|
# Updates-Verzeichnis für Downloads
|
|
os.makedirs("updates", exist_ok=True)
|
|
|
|
def load_version_info(self) -> Dict[str, Any]:
|
|
"""Lädt die Versionsinformationen aus der Konfigurationsdatei."""
|
|
if not os.path.exists(self.CONFIG_FILE):
|
|
default_info = {
|
|
"current_version": "1.0.0",
|
|
"last_check": "",
|
|
"channel": "stable",
|
|
"auto_check": True,
|
|
"auto_download": False
|
|
}
|
|
|
|
# Standardwerte speichern
|
|
try:
|
|
with open(self.CONFIG_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(default_info, f, indent=2)
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Speichern der Standardversionsinformationen: {e}")
|
|
|
|
return default_info
|
|
|
|
try:
|
|
with open(self.CONFIG_FILE, "r", encoding="utf-8") as f:
|
|
version_info = json.load(f)
|
|
|
|
logger.info(f"Versionsinformationen geladen: {version_info.get('current_version', 'unbekannt')}")
|
|
|
|
return version_info
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Laden der Versionsinformationen: {e}")
|
|
return {
|
|
"current_version": "1.0.0",
|
|
"last_check": "",
|
|
"channel": "stable",
|
|
"auto_check": True,
|
|
"auto_download": False
|
|
}
|
|
|
|
def save_version_info(self) -> bool:
|
|
"""Speichert die Versionsinformationen in die Konfigurationsdatei."""
|
|
try:
|
|
with open(self.CONFIG_FILE, "w", encoding="utf-8") as f:
|
|
json.dump(self.version_info, f, indent=2)
|
|
|
|
logger.info("Versionsinformationen gespeichert")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Speichern der Versionsinformationen: {e}")
|
|
return False
|
|
|
|
def compare_versions(self, version1: str, version2: str) -> int:
|
|
"""
|
|
Vergleicht zwei Versionsstrings (semver).
|
|
|
|
Args:
|
|
version1: Erste Version
|
|
version2: Zweite Version
|
|
|
|
Returns:
|
|
-1, wenn version1 < version2
|
|
0, wenn version1 == version2
|
|
1, wenn version1 > version2
|
|
"""
|
|
v1_parts = [int(part) for part in version1.split(".")]
|
|
v2_parts = [int(part) for part in version2.split(".")]
|
|
|
|
# Fülle fehlende Teile mit Nullen auf
|
|
while len(v1_parts) < 3:
|
|
v1_parts.append(0)
|
|
while len(v2_parts) < 3:
|
|
v2_parts.append(0)
|
|
|
|
# Vergleiche die Teile
|
|
for i in range(3):
|
|
if v1_parts[i] < v2_parts[i]:
|
|
return -1
|
|
elif v1_parts[i] > v2_parts[i]:
|
|
return 1
|
|
|
|
return 0
|
|
|
|
def check_for_updates(self, license_key: Optional[str] = None, force: bool = False) -> Dict[str, Any]:
|
|
"""
|
|
Überprüft, ob Updates verfügbar sind.
|
|
|
|
Args:
|
|
license_key: Lizenzschlüssel für die Update-Prüfung
|
|
force: Erzwingt eine Überprüfung, auch wenn erst kürzlich geprüft wurde
|
|
|
|
Returns:
|
|
Dictionary mit Update-Informationen
|
|
"""
|
|
result = {
|
|
"has_update": False,
|
|
"current_version": self.version_info["current_version"],
|
|
"latest_version": self.version_info["current_version"],
|
|
"release_date": "",
|
|
"release_notes": "",
|
|
"download_url": "",
|
|
"error": ""
|
|
}
|
|
|
|
# Prüfe, ob seit der letzten Überprüfung genügend Zeit vergangen ist (24 Stunden)
|
|
if not force and self.version_info.get("last_check"):
|
|
try:
|
|
last_check = datetime.fromisoformat(self.version_info["last_check"])
|
|
now = datetime.now()
|
|
|
|
# Wenn weniger als 24 Stunden seit der letzten Überprüfung vergangen sind
|
|
if (now - last_check).total_seconds() < 86400:
|
|
logger.info("Update-Überprüfung übersprungen (letzte Überprüfung vor weniger als 24 Stunden)")
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"Fehler beim Parsen des letzten Überprüfungsdatums: {e}")
|
|
|
|
try:
|
|
logger.info("Prüfe auf Updates...")
|
|
|
|
# Nutze die API für Update-Check
|
|
if license_key:
|
|
api_result = self.api_client.check_version(
|
|
current_version=self.version_info["current_version"],
|
|
license_key=license_key
|
|
)
|
|
else:
|
|
# Ohne Lizenz nur Latest Version Info
|
|
api_result = self.api_client.get_latest_version()
|
|
|
|
if api_result.get("success"):
|
|
data = api_result.get("data", {})
|
|
|
|
# Für check_version endpoint
|
|
if "update_available" in data:
|
|
result["has_update"] = data.get("update_available", False)
|
|
result["latest_version"] = data.get("latest_version", self.version_info["current_version"])
|
|
result["download_url"] = data.get("download_url", "")
|
|
result["release_notes"] = data.get("release_notes", "")
|
|
# Für get_latest_version endpoint
|
|
else:
|
|
latest_version = data.get("version", self.version_info["current_version"])
|
|
result["latest_version"] = latest_version
|
|
result["has_update"] = self.compare_versions(self.version_info["current_version"], latest_version) < 0
|
|
result["release_date"] = data.get("release_date", "")
|
|
result["release_notes"] = data.get("release_notes", "")
|
|
result["download_url"] = data.get("download_url", "")
|
|
|
|
# Update der letzten Überprüfung
|
|
self.version_info["last_check"] = datetime.now().isoformat()
|
|
self.save_version_info()
|
|
|
|
if result["has_update"]:
|
|
logger.info(f"Update verfügbar: {result['latest_version']}")
|
|
else:
|
|
logger.info("Keine Updates verfügbar")
|
|
|
|
else:
|
|
error_msg = api_result.get("error", "Fehler bei der Update-Prüfung")
|
|
logger.error(f"API-Fehler bei Update-Check: {error_msg}")
|
|
result["error"] = error_msg
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
error_msg = f"Unerwarteter Fehler bei der Update-Überprüfung: {e}"
|
|
logger.error(error_msg)
|
|
result["error"] = error_msg
|
|
return result
|
|
|
|
def download_update(self, download_url: str, version: str) -> Dict[str, Any]:
|
|
"""
|
|
Lädt ein Update herunter.
|
|
|
|
Args:
|
|
download_url: URL zum Herunterladen des Updates
|
|
version: Version des Updates
|
|
|
|
Returns:
|
|
Dictionary mit Download-Informationen
|
|
"""
|
|
result = {
|
|
"success": False,
|
|
"file_path": "",
|
|
"version": version,
|
|
"error": ""
|
|
}
|
|
|
|
try:
|
|
# Zieldateiname erstellen
|
|
file_name = f"update_v{version}.zip"
|
|
file_path = os.path.join("updates", file_name)
|
|
|
|
# Simuliere einen Download für Entwicklungszwecke
|
|
# In der Produktion sollte ein echter Download implementiert werden
|
|
|
|
# response = requests.get(download_url, stream=True, timeout=60)
|
|
# if response.status_code == 200:
|
|
# with open(file_path, "wb") as f:
|
|
# shutil.copyfileobj(response.raw, f)
|
|
|
|
# Simulierter Download (erstelle eine leere Datei)
|
|
with open(file_path, "w") as f:
|
|
f.write(f"Placeholder for version {version} update")
|
|
|
|
result["success"] = True
|
|
result["file_path"] = file_path
|
|
|
|
logger.info(f"Update v{version} heruntergeladen: {file_path}")
|
|
|
|
return result
|
|
|
|
except requests.RequestException as e:
|
|
error_msg = f"Netzwerkfehler beim Herunterladen des Updates: {e}"
|
|
logger.error(error_msg)
|
|
result["error"] = error_msg
|
|
return result
|
|
except Exception as e:
|
|
error_msg = f"Unerwarteter Fehler beim Herunterladen des Updates: {e}"
|
|
logger.error(error_msg)
|
|
result["error"] = error_msg
|
|
return result
|
|
|
|
def is_update_available(self) -> bool:
|
|
"""
|
|
Überprüft, ob ein Update verfügbar ist.
|
|
|
|
Returns:
|
|
True, wenn ein Update verfügbar ist, sonst False
|
|
"""
|
|
update_info = self.check_for_updates()
|
|
return update_info["has_update"]
|
|
|
|
def get_current_version(self) -> str:
|
|
"""
|
|
Gibt die aktuelle Version zurück.
|
|
|
|
Returns:
|
|
Aktuelle Version
|
|
"""
|
|
return self.version_info["current_version"]
|
|
|
|
def set_current_version(self, version: str) -> bool:
|
|
"""
|
|
Setzt die aktuelle Version.
|
|
|
|
Args:
|
|
version: Neue Version
|
|
|
|
Returns:
|
|
True bei Erfolg, False im Fehlerfall
|
|
"""
|
|
try:
|
|
self.version_info["current_version"] = version
|
|
return self.save_version_info()
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Setzen der aktuellen Version: {e}")
|
|
return False
|
|
|
|
def set_update_channel(self, channel: str) -> bool:
|
|
"""
|
|
Setzt den Update-Kanal (stable, beta, dev).
|
|
|
|
Args:
|
|
channel: Update-Kanal
|
|
|
|
Returns:
|
|
True bei Erfolg, False im Fehlerfall
|
|
"""
|
|
if channel not in ["stable", "beta", "dev"]:
|
|
logger.warning(f"Ungültiger Update-Kanal: {channel}")
|
|
return False
|
|
|
|
try:
|
|
self.version_info["channel"] = channel
|
|
return self.save_version_info()
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Setzen des Update-Kanals: {e}")
|
|
return False
|
|
|
|
def get_update_channel(self) -> str:
|
|
"""
|
|
Gibt den aktuellen Update-Kanal zurück.
|
|
|
|
Returns:
|
|
Update-Kanal (stable, beta, dev)
|
|
"""
|
|
return self.version_info.get("channel", "stable")
|
|
|
|
def set_auto_check(self, auto_check: bool) -> bool:
|
|
"""
|
|
Aktiviert oder deaktiviert die automatische Update-Überprüfung.
|
|
|
|
Args:
|
|
auto_check: True, um automatische Updates zu aktivieren, False zum Deaktivieren
|
|
|
|
Returns:
|
|
True bei Erfolg, False im Fehlerfall
|
|
"""
|
|
try:
|
|
self.version_info["auto_check"] = bool(auto_check)
|
|
return self.save_version_info()
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Setzen der automatischen Update-Überprüfung: {e}")
|
|
return False
|
|
|
|
def is_auto_check_enabled(self) -> bool:
|
|
"""
|
|
Überprüft, ob die automatische Update-Überprüfung aktiviert ist.
|
|
|
|
Returns:
|
|
True, wenn die automatische Update-Überprüfung aktiviert ist, sonst False
|
|
"""
|
|
return self.version_info.get("auto_check", True)
|
|
|
|
def set_auto_download(self, auto_download: bool) -> bool:
|
|
"""
|
|
Aktiviert oder deaktiviert den automatischen Download von Updates.
|
|
|
|
Args:
|
|
auto_download: True, um automatische Downloads zu aktivieren, False zum Deaktivieren
|
|
|
|
Returns:
|
|
True bei Erfolg, False im Fehlerfall
|
|
"""
|
|
try:
|
|
self.version_info["auto_download"] = bool(auto_download)
|
|
return self.save_version_info()
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Setzen des automatischen Downloads: {e}")
|
|
return False
|
|
|
|
def is_auto_download_enabled(self) -> bool:
|
|
"""
|
|
Überprüft, ob der automatische Download von Updates aktiviert ist.
|
|
|
|
Returns:
|
|
True, wenn der automatische Download aktiviert ist, sonst False
|
|
"""
|
|
return self.version_info.get("auto_download", False)
|
|
|
|
def apply_update(self, update_file: str) -> Dict[str, Any]:
|
|
"""
|
|
Wendet ein heruntergeladenes Update an.
|
|
|
|
Args:
|
|
update_file: Pfad zur Update-Datei
|
|
|
|
Returns:
|
|
Dictionary mit Informationen über die Anwendung des Updates
|
|
"""
|
|
result = {
|
|
"success": False,
|
|
"version": "",
|
|
"error": ""
|
|
}
|
|
|
|
if not os.path.exists(update_file):
|
|
result["error"] = f"Update-Datei nicht gefunden: {update_file}"
|
|
logger.error(result["error"])
|
|
return result
|
|
|
|
try:
|
|
# In der Produktion sollte hier die tatsächliche Update-Logik implementiert werden
|
|
# 1. Extrahieren des Updates
|
|
# 2. Sichern der aktuellen Version
|
|
# 3. Anwenden der Änderungen
|
|
# 4. Aktualisieren der Versionsinformationen
|
|
|
|
# Simuliere ein erfolgreiches Update
|
|
logger.info(f"Update aus {update_file} erfolgreich angewendet (simuliert)")
|
|
|
|
# Extrahiere Version aus dem Dateinamen
|
|
import re
|
|
file_name = os.path.basename(update_file)
|
|
version_match = re.search(r"v([0-9.]+)", file_name)
|
|
|
|
if version_match:
|
|
new_version = version_match.group(1)
|
|
self.set_current_version(new_version)
|
|
result["version"] = new_version
|
|
|
|
result["success"] = True
|
|
|
|
return result
|
|
|
|
except Exception as e:
|
|
error_msg = f"Fehler beim Anwenden des Updates: {e}"
|
|
logger.error(error_msg)
|
|
result["error"] = error_msg
|
|
return result |