130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
|
"""Update checker for Local Transcription - checks Gitea for new releases."""
|
||
|
|
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Optional, Callable
|
||
|
|
import threading
|
||
|
|
import requests
|
||
|
|
from packaging import version
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class ReleaseInfo:
|
||
|
|
"""Information about a release."""
|
||
|
|
version: str # Parsed version without 'v' prefix (e.g., "1.4.0")
|
||
|
|
tag_name: str # Original tag name (e.g., "v1.4.0")
|
||
|
|
release_notes: str # Release notes / body markdown
|
||
|
|
download_url: str # Browser URL for release page (html_url)
|
||
|
|
|
||
|
|
|
||
|
|
class UpdateChecker:
|
||
|
|
"""Checks for updates from a Gitea server."""
|
||
|
|
|
||
|
|
def __init__(
|
||
|
|
self,
|
||
|
|
current_version: str,
|
||
|
|
gitea_url: str = "",
|
||
|
|
owner: str = "",
|
||
|
|
repo: str = "",
|
||
|
|
timeout: int = 10
|
||
|
|
):
|
||
|
|
"""
|
||
|
|
Initialize the update checker.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
current_version: Current application version (e.g., "1.3.1")
|
||
|
|
gitea_url: Base URL of the Gitea server (e.g., "https://git.example.com")
|
||
|
|
owner: Repository owner/organization name
|
||
|
|
repo: Repository name
|
||
|
|
timeout: Request timeout in seconds
|
||
|
|
"""
|
||
|
|
self.current_version = current_version
|
||
|
|
self.gitea_url = gitea_url.rstrip('/') if gitea_url else ""
|
||
|
|
self.owner = owner
|
||
|
|
self.repo = repo
|
||
|
|
self.timeout = timeout
|
||
|
|
|
||
|
|
def is_configured(self) -> bool:
|
||
|
|
"""Check if update checking is properly configured."""
|
||
|
|
return bool(self.gitea_url and self.owner and self.repo)
|
||
|
|
|
||
|
|
def check_for_update(self) -> tuple[Optional[ReleaseInfo], Optional[str]]:
|
||
|
|
"""
|
||
|
|
Check for updates synchronously.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Tuple of (ReleaseInfo or None, error_message or None)
|
||
|
|
- If update available: (ReleaseInfo, None)
|
||
|
|
- If no update: (None, None)
|
||
|
|
- If error: (None, error_message)
|
||
|
|
"""
|
||
|
|
if not self.is_configured():
|
||
|
|
return None, "Update checking not configured"
|
||
|
|
|
||
|
|
try:
|
||
|
|
# Query Gitea API for latest release
|
||
|
|
api_url = f"{self.gitea_url}/api/v1/repos/{self.owner}/{self.repo}/releases/latest"
|
||
|
|
response = requests.get(api_url, timeout=self.timeout)
|
||
|
|
|
||
|
|
if response.status_code == 404:
|
||
|
|
return None, "No releases found"
|
||
|
|
|
||
|
|
response.raise_for_status()
|
||
|
|
data = response.json()
|
||
|
|
|
||
|
|
# Parse version from tag name (strip 'v' prefix if present)
|
||
|
|
tag_name = data.get('tag_name', '')
|
||
|
|
release_version = tag_name.lstrip('v')
|
||
|
|
|
||
|
|
if not release_version:
|
||
|
|
return None, "Invalid release tag"
|
||
|
|
|
||
|
|
# Compare versions using packaging library
|
||
|
|
try:
|
||
|
|
current_ver = version.parse(self.current_version)
|
||
|
|
release_ver = version.parse(release_version)
|
||
|
|
except version.InvalidVersion as e:
|
||
|
|
return None, f"Invalid version format: {e}"
|
||
|
|
|
||
|
|
# Check if release is newer
|
||
|
|
if release_ver > current_ver:
|
||
|
|
release_info = ReleaseInfo(
|
||
|
|
version=release_version,
|
||
|
|
tag_name=tag_name,
|
||
|
|
release_notes=data.get('body', ''),
|
||
|
|
download_url=data.get('html_url', '')
|
||
|
|
)
|
||
|
|
return release_info, None
|
||
|
|
else:
|
||
|
|
return None, None # No update available
|
||
|
|
|
||
|
|
except requests.exceptions.Timeout:
|
||
|
|
return None, "Connection timed out"
|
||
|
|
except requests.exceptions.ConnectionError:
|
||
|
|
return None, "Could not connect to server"
|
||
|
|
except requests.exceptions.RequestException as e:
|
||
|
|
return None, f"Request failed: {e}"
|
||
|
|
except Exception as e:
|
||
|
|
return None, f"Error checking for updates: {e}"
|
||
|
|
|
||
|
|
def check_for_update_async(
|
||
|
|
self,
|
||
|
|
callback: Callable[[Optional[ReleaseInfo], Optional[str]], None]
|
||
|
|
) -> threading.Thread:
|
||
|
|
"""
|
||
|
|
Check for updates asynchronously in a background thread.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
callback: Function to call with (ReleaseInfo or None, error_message or None)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
The started thread
|
||
|
|
"""
|
||
|
|
def worker():
|
||
|
|
result, error = self.check_for_update()
|
||
|
|
callback(result, error)
|
||
|
|
|
||
|
|
thread = threading.Thread(target=worker, daemon=True)
|
||
|
|
thread.start()
|
||
|
|
return thread
|