|
5 | 5 | import subprocess |
6 | 6 | import sys |
7 | 7 | import tempfile |
| 8 | +import time |
8 | 9 | from dataclasses import dataclass |
9 | 10 | from pathlib import Path |
10 | 11 | from typing import Any |
|
20 | 21 | f"https://raw.githubusercontent.com/{REPO_OWNER}/{REPO_NAME}/{{tag}}/install.sh" |
21 | 22 | ) |
22 | 23 | REQUEST_TIMEOUT_SECONDS = 10 |
| 24 | +AUTO_UPDATE_CHECK_TIMEOUT_SECONDS = 2 |
| 25 | +AUTO_UPDATE_CHECK_INTERVAL_SECONDS = 24 * 60 * 60 |
| 26 | +AUTO_UPDATE_CHECK_DISABLE_ENV = "ATLASSIAN_DISABLE_UPDATE_CHECK" |
23 | 27 |
|
24 | 28 |
|
25 | 29 | class UpdateError(RuntimeError): |
@@ -155,6 +159,126 @@ def get_update_info(current_version: str) -> UpdateInfo: |
155 | 159 | ) |
156 | 160 |
|
157 | 161 |
|
| 162 | +def auto_update_check_state_path( |
| 163 | + *, |
| 164 | + environ: dict[str, str] | None = None, |
| 165 | + home: Path | None = None, |
| 166 | +) -> Path: |
| 167 | + env = os.environ if environ is None else environ |
| 168 | + cache_home = env.get("XDG_CACHE_HOME") |
| 169 | + base_dir = Path(cache_home).expanduser() if cache_home else (home or Path.home()) / ".cache" |
| 170 | + return base_dir / "atlassian-cli" / "update-check.json" |
| 171 | + |
| 172 | + |
| 173 | +def _env_flag_enabled(value: str | None) -> bool: |
| 174 | + return value is not None and value.strip().lower() in {"1", "true", "yes", "on"} |
| 175 | + |
| 176 | + |
| 177 | +def _read_update_check_state(state_path: Path) -> dict[str, Any]: |
| 178 | + try: |
| 179 | + payload = json.loads(state_path.read_text(encoding="utf-8")) |
| 180 | + except (OSError, json.JSONDecodeError): |
| 181 | + return {} |
| 182 | + return payload if isinstance(payload, dict) else {} |
| 183 | + |
| 184 | + |
| 185 | +def _write_update_check_state(state_path: Path, state: dict[str, Any]) -> None: |
| 186 | + try: |
| 187 | + state_path.parent.mkdir(parents=True, exist_ok=True) |
| 188 | + state_path.write_text(json.dumps(state, sort_keys=True), encoding="utf-8") |
| 189 | + except OSError: |
| 190 | + return |
| 191 | + |
| 192 | + |
| 193 | +def _record_update_check_error( |
| 194 | + state_path: Path | None, |
| 195 | + *, |
| 196 | + checked_at: int, |
| 197 | + error: Exception, |
| 198 | + extra_state: dict[str, Any] | None = None, |
| 199 | +) -> None: |
| 200 | + if state_path is None: |
| 201 | + return |
| 202 | + state = { |
| 203 | + "last_checked_at": checked_at, |
| 204 | + "last_error": str(error), |
| 205 | + } |
| 206 | + if extra_state: |
| 207 | + state.update(extra_state) |
| 208 | + _write_update_check_state(state_path, state) |
| 209 | + |
| 210 | + |
| 211 | +def _recently_checked( |
| 212 | + state: dict[str, Any], |
| 213 | + *, |
| 214 | + now: int, |
| 215 | + interval_seconds: int, |
| 216 | +) -> bool: |
| 217 | + if "last_checked_at" not in state: |
| 218 | + return False |
| 219 | + try: |
| 220 | + last_checked_at = int(state["last_checked_at"]) |
| 221 | + except (TypeError, ValueError): |
| 222 | + return False |
| 223 | + if now < last_checked_at: |
| 224 | + return False |
| 225 | + return now - last_checked_at < interval_seconds |
| 226 | + |
| 227 | + |
| 228 | +def format_update_notice(current_version: str, latest: ReleaseInfo) -> str: |
| 229 | + lines = [ |
| 230 | + f"atlassian-cli {current_version} can be updated to {latest.tag}.", |
| 231 | + "Run: atlassian update install", |
| 232 | + ] |
| 233 | + if latest.url: |
| 234 | + lines.append(f"Release: {latest.url}") |
| 235 | + return "\n".join(lines) |
| 236 | + |
| 237 | + |
| 238 | +def check_for_update_notice( |
| 239 | + current_version: str, |
| 240 | + *, |
| 241 | + state_path: Path | None = None, |
| 242 | + now: int | None = None, |
| 243 | + environ: dict[str, str] | None = None, |
| 244 | + timeout: int = AUTO_UPDATE_CHECK_TIMEOUT_SECONDS, |
| 245 | + interval_seconds: int = AUTO_UPDATE_CHECK_INTERVAL_SECONDS, |
| 246 | +) -> str | None: |
| 247 | + env = os.environ if environ is None else environ |
| 248 | + if _env_flag_enabled(env.get(AUTO_UPDATE_CHECK_DISABLE_ENV)): |
| 249 | + return None |
| 250 | + |
| 251 | + checked_at = int(time.time() if now is None else now) |
| 252 | + path: Path | None = None |
| 253 | + release_state: dict[str, Any] | None = None |
| 254 | + try: |
| 255 | + path = state_path or auto_update_check_state_path(environ=env) |
| 256 | + previous_state = _read_update_check_state(path) |
| 257 | + if _recently_checked(previous_state, now=checked_at, interval_seconds=interval_seconds): |
| 258 | + return None |
| 259 | + |
| 260 | + latest = fetch_latest_release(timeout=timeout) |
| 261 | + release_state = { |
| 262 | + "last_checked_at": checked_at, |
| 263 | + "latest_tag": latest.tag, |
| 264 | + "latest_version": latest.version, |
| 265 | + "release_url": latest.url, |
| 266 | + } |
| 267 | + _write_update_check_state(path, release_state) |
| 268 | + update_available = is_newer_version(latest.version, current_version) |
| 269 | + except Exception as exc: |
| 270 | + _record_update_check_error( |
| 271 | + path, |
| 272 | + checked_at=checked_at, |
| 273 | + error=exc, |
| 274 | + extra_state=release_state, |
| 275 | + ) |
| 276 | + return None |
| 277 | + if not update_available: |
| 278 | + return None |
| 279 | + return format_update_notice(current_version, latest) |
| 280 | + |
| 281 | + |
158 | 282 | def default_install_dir( |
159 | 283 | *, |
160 | 284 | environ: dict[str, str] | None = None, |
|
0 commit comments