561 lines
17 KiB
Python
561 lines
17 KiB
Python
# Copyright (C) 2025 James Brotosky, Brandon Wickline
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
"""
|
|
Version checking and update notification system for Loxide.
|
|
|
|
Checks against Gitea releases at:
|
|
https://git.racooncity.org/brotoskyj/AirlockTools/releases
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
import re
|
|
import threading
|
|
from typing import Callable, Optional
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Current application version - UPDATE THIS ON EACH RELEASE
|
|
__version__ = "1.1.0"
|
|
|
|
# Gitea release API configuration
|
|
GITEA_API_BASE = "https://git.racooncity.org/api/v1"
|
|
REPO_OWNER = "brotoskyj"
|
|
REPO_NAME = "AirlockTools"
|
|
RELEASES_URL = f"{GITEA_API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/releases"
|
|
RELEASES_PAGE_URL = f"https://git.racooncity.org/{REPO_OWNER}/{REPO_NAME}/releases"
|
|
|
|
# How often to check for updates (in hours)
|
|
CHECK_INTERVAL_HOURS = 24
|
|
|
|
|
|
@dataclass
|
|
class ReleaseInfo:
|
|
"""Information about a release."""
|
|
|
|
tag_name: str
|
|
version: tuple # Parsed semantic version (major, minor, patch)
|
|
name: str
|
|
body: str # Release notes
|
|
published_at: datetime
|
|
html_url: str
|
|
download_url: Optional[str] = None # URL to download the release asset
|
|
is_prerelease: bool = False
|
|
|
|
|
|
@dataclass
|
|
class UpdateCheckResult:
|
|
"""Result of an update check."""
|
|
|
|
current_version: str
|
|
latest_version: Optional[str]
|
|
update_available: bool
|
|
release_info: Optional[ReleaseInfo]
|
|
error: Optional[str] = None
|
|
|
|
|
|
def parse_version(version_str: str) -> Optional[tuple]:
|
|
"""
|
|
Parse a version string into a comparable tuple.
|
|
Supports formats: v1.2.3, 1.2.3, v1.2, 1.2
|
|
|
|
Returns (major, minor, patch) tuple or None if parsing fails.
|
|
"""
|
|
if not version_str:
|
|
return None
|
|
|
|
# Strip 'v' prefix if present
|
|
clean = version_str.lstrip("vV").strip()
|
|
|
|
# Match semantic version pattern
|
|
match = re.match(r"^(\d+)(?:\.(\d+))?(?:\.(\d+))?", clean)
|
|
if not match:
|
|
return None
|
|
|
|
major = int(match.group(1))
|
|
minor = int(match.group(2)) if match.group(2) else 0
|
|
patch = int(match.group(3)) if match.group(3) else 0
|
|
|
|
return (major, minor, patch)
|
|
|
|
|
|
def compare_versions(v1: tuple, v2: tuple) -> int:
|
|
"""
|
|
Compare two version tuples.
|
|
|
|
Returns:
|
|
-1 if v1 < v2
|
|
0 if v1 == v2
|
|
1 if v1 > v2
|
|
"""
|
|
for a, b in zip(v1, v2):
|
|
if a < b:
|
|
return -1
|
|
if a > b:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def get_current_version() -> str:
|
|
"""Get the current application version."""
|
|
return __version__
|
|
|
|
|
|
def _parse_release_response(release_data: dict) -> Optional[ReleaseInfo]:
|
|
"""Parse a release from Gitea API response."""
|
|
try:
|
|
tag_name = release_data.get("tag_name", "")
|
|
version = parse_version(tag_name)
|
|
if not version:
|
|
logger.debug(f"Could not parse version from tag: {tag_name}")
|
|
return None
|
|
|
|
# Parse published date
|
|
published_str = release_data.get("published_at", "")
|
|
try:
|
|
published_at = datetime.fromisoformat(published_str.replace("Z", "+00:00"))
|
|
except (ValueError, AttributeError):
|
|
published_at = datetime.now(UTC)
|
|
|
|
# Get download URL from assets if available
|
|
download_url = None
|
|
assets = release_data.get("assets", [])
|
|
for asset in assets:
|
|
# Prefer .exe or .zip files
|
|
name = asset.get("name", "").lower()
|
|
if name.endswith((".exe", ".zip", ".msi")):
|
|
download_url = asset.get("browser_download_url")
|
|
break
|
|
|
|
return ReleaseInfo(
|
|
tag_name=tag_name,
|
|
version=version,
|
|
name=release_data.get("name", tag_name),
|
|
body=release_data.get("body", ""),
|
|
published_at=published_at,
|
|
html_url=release_data.get("html_url", RELEASES_PAGE_URL),
|
|
download_url=download_url,
|
|
is_prerelease=release_data.get("prerelease", False),
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to parse release data: {e}")
|
|
return None
|
|
|
|
|
|
def fetch_latest_release(
|
|
include_prerelease: bool = False, timeout: int = 10
|
|
) -> Optional[ReleaseInfo]:
|
|
"""
|
|
Fetch the latest release from Gitea.
|
|
|
|
Args:
|
|
include_prerelease: Whether to include pre-release versions
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
ReleaseInfo for the latest release, or None if fetch fails
|
|
"""
|
|
try:
|
|
response = requests.get(
|
|
RELEASES_URL,
|
|
params={"limit": 10}, # Get last 10 releases to find latest stable
|
|
timeout=timeout,
|
|
headers={"Accept": "application/json"},
|
|
)
|
|
response.raise_for_status()
|
|
|
|
releases = response.json()
|
|
if not releases:
|
|
logger.debug("No releases found")
|
|
return None
|
|
|
|
# Find the latest release (first non-prerelease if we're excluding them)
|
|
for release_data in releases:
|
|
release_info = _parse_release_response(release_data)
|
|
if release_info is None:
|
|
continue
|
|
|
|
if include_prerelease or not release_info.is_prerelease:
|
|
return release_info
|
|
|
|
# If all are prereleases and we're excluding them, return the first one anyway
|
|
# but log a warning
|
|
if releases:
|
|
logger.debug("All releases are pre-releases")
|
|
return _parse_release_response(releases[0])
|
|
|
|
return None
|
|
|
|
except requests.exceptions.Timeout:
|
|
logger.warning("Timeout fetching releases from Gitea")
|
|
return None
|
|
except requests.exceptions.RequestException as e:
|
|
logger.warning(f"Failed to fetch releases: {e}")
|
|
return None
|
|
except (json.JSONDecodeError, KeyError) as e:
|
|
logger.warning(f"Failed to parse release response: {e}")
|
|
return None
|
|
|
|
|
|
def check_for_updates(include_prerelease: bool = False) -> UpdateCheckResult:
|
|
"""
|
|
Check if a newer version is available.
|
|
|
|
Args:
|
|
include_prerelease: Whether to consider pre-release versions
|
|
|
|
Returns:
|
|
UpdateCheckResult with the check results
|
|
"""
|
|
current = get_current_version()
|
|
current_parsed = parse_version(current)
|
|
|
|
if not current_parsed:
|
|
return UpdateCheckResult(
|
|
current_version=current,
|
|
latest_version=None,
|
|
update_available=False,
|
|
release_info=None,
|
|
error="Could not parse current version",
|
|
)
|
|
|
|
release_info = fetch_latest_release(include_prerelease=include_prerelease)
|
|
|
|
if release_info is None:
|
|
return UpdateCheckResult(
|
|
current_version=current,
|
|
latest_version=None,
|
|
update_available=False,
|
|
release_info=None,
|
|
error="Could not fetch release information",
|
|
)
|
|
|
|
is_newer = compare_versions(release_info.version, current_parsed) > 0
|
|
|
|
return UpdateCheckResult(
|
|
current_version=current,
|
|
latest_version=release_info.tag_name,
|
|
update_available=is_newer,
|
|
release_info=release_info,
|
|
)
|
|
|
|
|
|
class VersionChecker:
|
|
"""
|
|
Background version checker that periodically checks for updates
|
|
and can notify the application when updates are available.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
cache_dir: Optional[Path] = None,
|
|
check_interval_hours: int = CHECK_INTERVAL_HOURS,
|
|
on_update_available: Optional[Callable[[UpdateCheckResult], None]] = None,
|
|
):
|
|
"""
|
|
Initialize the version checker.
|
|
|
|
Args:
|
|
cache_dir: Directory to store last check timestamp
|
|
check_interval_hours: Hours between automatic checks
|
|
on_update_available: Callback when update is available
|
|
"""
|
|
self.cache_dir = cache_dir
|
|
self.check_interval = timedelta(hours=check_interval_hours)
|
|
self.on_update_available = on_update_available
|
|
self._last_check: Optional[datetime] = None
|
|
self._last_result: Optional[UpdateCheckResult] = None
|
|
self._check_thread: Optional[threading.Thread] = None
|
|
self._dismissed_version: Optional[str] = None
|
|
|
|
# Load cached state
|
|
self._load_cache()
|
|
|
|
@property
|
|
def cache_file(self) -> Optional[Path]:
|
|
if self.cache_dir:
|
|
return self.cache_dir / "version_check_cache.json"
|
|
return None
|
|
|
|
def _load_cache(self) -> None:
|
|
"""Load cached check state."""
|
|
if not self.cache_file or not self.cache_file.exists():
|
|
return
|
|
|
|
try:
|
|
with open(self.cache_file, "r") as f:
|
|
data = json.load(f)
|
|
|
|
if "last_check" in data:
|
|
self._last_check = datetime.fromisoformat(data["last_check"])
|
|
# Don't load dismissed_version - dismiss is session-only
|
|
|
|
except (json.JSONDecodeError, ValueError, OSError) as e:
|
|
logger.debug(f"Could not load version check cache: {e}")
|
|
|
|
def _save_cache(self) -> None:
|
|
"""Save check state to cache."""
|
|
if not self.cache_file:
|
|
return
|
|
|
|
try:
|
|
self.cache_file.parent.mkdir(parents=True, exist_ok=True)
|
|
data = {}
|
|
if self._last_check:
|
|
data["last_check"] = self._last_check.isoformat()
|
|
# Don't save dismissed_version - dismiss is session-only
|
|
|
|
with open(self.cache_file, "w") as f:
|
|
json.dump(data, f)
|
|
|
|
except OSError as e:
|
|
logger.debug(f"Could not save version check cache: {e}")
|
|
|
|
def should_check(self) -> bool:
|
|
"""Determine if enough time has passed to check again."""
|
|
if self._last_check is None:
|
|
return True
|
|
|
|
elapsed = datetime.now(UTC) - self._last_check
|
|
return elapsed >= self.check_interval
|
|
|
|
def check_now(
|
|
self, force: bool = False, include_prerelease: bool = False
|
|
) -> UpdateCheckResult:
|
|
"""
|
|
Check for updates immediately.
|
|
|
|
Args:
|
|
force: Check even if recently checked
|
|
include_prerelease: Include pre-release versions
|
|
|
|
Returns:
|
|
UpdateCheckResult
|
|
"""
|
|
if not force and not self.should_check() and self._last_result:
|
|
return self._last_result
|
|
|
|
result = check_for_updates(include_prerelease=include_prerelease)
|
|
self._last_check = datetime.now(UTC)
|
|
self._last_result = result
|
|
self._save_cache()
|
|
|
|
# Notify if update available and not dismissed
|
|
if (
|
|
result.update_available
|
|
and self.on_update_available
|
|
and result.latest_version != self._dismissed_version
|
|
):
|
|
self.on_update_available(result)
|
|
|
|
return result
|
|
|
|
def check_async(
|
|
self, force: bool = False, include_prerelease: bool = False
|
|
) -> None:
|
|
"""
|
|
Check for updates in background thread.
|
|
|
|
Args:
|
|
force: Check even if recently checked
|
|
include_prerelease: Include pre-release versions
|
|
"""
|
|
if self._check_thread and self._check_thread.is_alive():
|
|
return # Already checking
|
|
|
|
if not force and not self.should_check():
|
|
return # Too soon to check again
|
|
|
|
def _check():
|
|
try:
|
|
self.check_now(force=True, include_prerelease=include_prerelease)
|
|
except Exception as e:
|
|
logger.debug(f"Background version check failed: {e}")
|
|
|
|
self._check_thread = threading.Thread(target=_check, daemon=True)
|
|
self._check_thread.start()
|
|
|
|
def dismiss_update(self, version: str) -> None:
|
|
"""
|
|
Dismiss update notification for a specific version.
|
|
Only lasts for the current session - will nag again on next startup.
|
|
|
|
Args:
|
|
version: Version to dismiss (e.g., "v1.2.3")
|
|
"""
|
|
# Session-only dismiss - don't save to cache
|
|
self._dismissed_version = version
|
|
|
|
def clear_dismissed(self) -> None:
|
|
"""Clear the dismissed version so user will be nagged again."""
|
|
self._dismissed_version = None
|
|
|
|
def get_last_result(self) -> Optional[UpdateCheckResult]:
|
|
"""Get the result of the last check."""
|
|
return self._last_result
|
|
|
|
|
|
# Global instance for easy access
|
|
_global_checker: Optional[VersionChecker] = None
|
|
|
|
|
|
def get_version_checker(
|
|
cache_dir: Optional[Path] = None,
|
|
on_update_available: Optional[Callable[[UpdateCheckResult], None]] = None,
|
|
) -> VersionChecker:
|
|
"""
|
|
Get or create the global version checker instance.
|
|
|
|
Args:
|
|
cache_dir: Directory for caching (only used on first call)
|
|
on_update_available: Callback for updates (only used on first call)
|
|
|
|
Returns:
|
|
The global VersionChecker instance
|
|
"""
|
|
global _global_checker
|
|
|
|
if _global_checker is None:
|
|
_global_checker = VersionChecker(
|
|
cache_dir=cache_dir,
|
|
on_update_available=on_update_available,
|
|
)
|
|
|
|
return _global_checker
|
|
|
|
|
|
def format_update_message(result: UpdateCheckResult, short: bool = False) -> str:
|
|
"""
|
|
Format a human-readable update message.
|
|
|
|
Args:
|
|
result: The update check result
|
|
short: Whether to use a short format
|
|
|
|
Returns:
|
|
Formatted message string
|
|
"""
|
|
if not result.update_available:
|
|
return f"✅ Loxide is up to date (v{result.current_version})"
|
|
|
|
if short:
|
|
return f"🆕 Update available: {result.latest_version}"
|
|
|
|
msg = f"🆕 Loxide {result.latest_version} is available! (current: v{result.current_version})"
|
|
|
|
if result.release_info:
|
|
msg += f"\n📥 Download: {result.release_info.html_url}"
|
|
|
|
# Include release notes preview if available
|
|
if result.release_info.body:
|
|
notes = result.release_info.body.strip()
|
|
# Truncate if too long
|
|
if len(notes) > 200:
|
|
notes = notes[:200] + "..."
|
|
msg += f"\n\n📋 Release Notes:\n{notes}"
|
|
|
|
return msg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Textual TUI Integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def create_update_notifier(
|
|
app, cache_dir: Optional[Path] = None, nag_on_startup: bool = True
|
|
):
|
|
"""
|
|
Create a version checker that notifies via Textual toast notifications.
|
|
|
|
This should be called after the Textual app is created.
|
|
|
|
Args:
|
|
app: The Textual App instance
|
|
cache_dir: Directory for caching check state
|
|
nag_on_startup: Always show notification on startup if update available
|
|
|
|
Returns:
|
|
The VersionChecker instance
|
|
"""
|
|
|
|
def on_update_available(result: UpdateCheckResult):
|
|
"""Callback when update is available - show toast notification."""
|
|
try:
|
|
msg = f"🆕 Update available: {result.latest_version}\nGo to Settings to download"
|
|
try:
|
|
app.notify(
|
|
msg, title="Loxide Update Available", severity="warning", timeout=15
|
|
)
|
|
except RuntimeError:
|
|
app.call_from_thread(
|
|
app.notify,
|
|
msg,
|
|
title="Loxide Update Available",
|
|
severity="warning",
|
|
timeout=15,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Could not show update notification: {e}")
|
|
|
|
checker = get_version_checker(
|
|
cache_dir=cache_dir,
|
|
on_update_available=on_update_available,
|
|
)
|
|
|
|
# Store checker on app so Loxide.on_mount can use it
|
|
if nag_on_startup:
|
|
app._version_checker = checker
|
|
app._version_nag_shown = False
|
|
|
|
return checker
|
|
|
|
|
|
def check_for_updates_startup(
|
|
cache_dir: Optional[Path] = None,
|
|
) -> Optional[UpdateCheckResult]:
|
|
"""
|
|
Check for updates during application startup.
|
|
|
|
This performs a synchronous check but respects the cache interval,
|
|
so it will only actually query the network once per CHECK_INTERVAL_HOURS.
|
|
|
|
Returns the result if an update is available, None otherwise.
|
|
|
|
Example usage:
|
|
result = check_for_updates_startup(cache_dir)
|
|
if result and result.update_available:
|
|
print(format_update_message(result))
|
|
"""
|
|
checker = get_version_checker(cache_dir=cache_dir)
|
|
|
|
# Only check if enough time has passed (uses cache)
|
|
if not checker.should_check():
|
|
result = checker.get_last_result()
|
|
if result and result.update_available:
|
|
return result
|
|
return None
|
|
|
|
result = checker.check_now(force=False)
|
|
if result.update_available:
|
|
return result
|
|
return None
|