"""Update checker for DifferentialLab.
Checks weekly if a newer version is available in the repository.
If so, shows a dialog (when enabled via env) and can perform git pull
without overwriting user data (input/, output/, .env, etc.).
"""
from __future__ import annotations
import re
import subprocess
import time
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
from config import get_env_from_schema, get_project_root
from utils.logger import get_logger
logger = get_logger(__name__)
_LAST_CHECK_FILE = ".last_update_check"
_UPDATE_CHECK_TIMEOUT = 10
_VERSION_RE = re.compile(r"^(\d+(?:\.\d+)*)")
_PYPROJECT_VERSION_RE = re.compile(r'version\s*=\s*["\']([^"\']+)["\']')
def _get_last_check_path() -> Path:
"""Return path to the file storing last update check timestamp.
Returns:
Path to ``.last_update_check`` in project root.
"""
root = get_project_root()
return root / _LAST_CHECK_FILE
[docs]
def should_run_check() -> bool:
"""Return True if we should run the update check (once per week).
Returns:
True if enough time has passed since last check, or no previous check.
"""
if not get_env_from_schema("CHECK_UPDATES"):
return False
if get_env_from_schema("CHECK_UPDATES_FORCE"):
return True
path = _get_last_check_path()
if not path.exists():
return True
try:
mtime = path.stat().st_mtime
elapsed_days = (time.time() - mtime) / (24 * 3600)
days_between: int = get_env_from_schema("UPDATE_CHECK_INTERVAL_DAYS")
return elapsed_days >= days_between
except OSError as exc:
logger.debug("Could not stat last-check file, assuming check needed: %s", exc)
return True
[docs]
def record_check_done() -> None:
"""Record that an update check was performed (touch the file).
Updates the modification time of ``.last_update_check`` so the next
check is deferred by UPDATE_CHECK_INTERVAL_DAYS.
"""
path = _get_last_check_path()
try:
path.touch()
except OSError as exc:
logger.debug("Could not touch last-check file: %s", exc)
def _parse_version(version_str: str) -> tuple[int, ...]:
"""Parse a version string like '1.0.0' or '1.2.3.dev1' into a comparable tuple.
Args:
version_str: Version string from pyproject.toml.
Returns:
Tuple of integers for comparison (e.g. (0, 2, 0)).
"""
match = _VERSION_RE.match(str(version_str).strip())
if not match:
return (0,)
return tuple(int(x) for x in match.group(1).split("."))
def _fetch_latest_version(version_url: str | None = None) -> str | None:
"""Fetch the latest version from the remote pyproject.toml.
Args:
version_url: URL to pyproject.toml. If None, uses env UPDATE_CHECK_URL
or default.
Returns:
Version string (e.g. '0.4.1') or None if fetch failed.
"""
url = version_url or get_env_from_schema("UPDATE_CHECK_URL")
try:
req = Request(url, headers={"User-Agent": "DifferentialLab-UpdateChecker/1.0"})
with urlopen(req, timeout=_UPDATE_CHECK_TIMEOUT) as resp:
content = resp.read().decode("utf-8", errors="replace")
except (URLError, HTTPError, OSError, ValueError) as e:
logger.debug("Update check: could not fetch version from %s: %s", url, e)
return None
match = _PYPROJECT_VERSION_RE.search(content)
if match:
return match.group(1).strip()
return None
[docs]
def is_update_available(current_version: str) -> str | None:
"""Check if a newer version is available.
Args:
current_version: Current application version.
Returns:
The latest version string if newer, else None.
"""
latest = _fetch_latest_version()
if not latest:
return None
current_tuple = _parse_version(current_version)
latest_tuple = _parse_version(latest)
if latest_tuple > current_tuple:
return latest
return None