|
| 1 | +"""Git-based update checker with throttling and conflict detection.""" |
| 2 | + |
| 3 | +import subprocess |
| 4 | +import time |
| 5 | +from dataclasses import dataclass |
| 6 | +from pathlib import Path |
| 7 | + |
| 8 | +# Throttle: check at most once per this many seconds |
| 9 | +CHECK_INTERVAL_SECONDS = 12 * 60 * 60 # 12 hours |
| 10 | + |
| 11 | +# Git command timeout |
| 12 | +GIT_TIMEOUT_SECONDS = 10 |
| 13 | + |
| 14 | + |
| 15 | +@dataclass |
| 16 | +class UpdateStatus: |
| 17 | + """Result of an update check.""" |
| 18 | + |
| 19 | + has_update: bool = False |
| 20 | + local_ref: str = "" |
| 21 | + remote_ref: str = "" |
| 22 | + behind_count: int = 0 |
| 23 | + ahead_count: int = 0 |
| 24 | + changed_files: list[str] | None = None # files changed on remote |
| 25 | + conflict_files: list[str] | None = None # locally modified files that remote also changed |
| 26 | + error: str | None = None |
| 27 | + |
| 28 | + |
| 29 | +def _get_repo_root() -> Path | None: |
| 30 | + """Find the git repo root from the package's installed location.""" |
| 31 | + # Walk up from this file to find .git |
| 32 | + current = Path(__file__).resolve().parent |
| 33 | + for _ in range(10): |
| 34 | + if (current / ".git").exists(): |
| 35 | + return current |
| 36 | + parent = current.parent |
| 37 | + if parent == current: |
| 38 | + break |
| 39 | + current = parent |
| 40 | + return None |
| 41 | + |
| 42 | + |
| 43 | +def _run_git(repo: Path, *args: str, timeout: int = GIT_TIMEOUT_SECONDS) -> str | None: |
| 44 | + """Run a git command, return stdout or None on failure.""" |
| 45 | + try: |
| 46 | + result = subprocess.run( |
| 47 | + ["git", "-C", str(repo), *args], |
| 48 | + capture_output=True, |
| 49 | + text=True, |
| 50 | + timeout=timeout, |
| 51 | + ) |
| 52 | + if result.returncode == 0: |
| 53 | + return result.stdout.strip() |
| 54 | + return None |
| 55 | + except (subprocess.TimeoutExpired, FileNotFoundError, OSError): |
| 56 | + return None |
| 57 | + |
| 58 | + |
| 59 | +def _get_stamp_path(repo: Path) -> Path: |
| 60 | + """Path to the last-check timestamp file.""" |
| 61 | + return repo / ".git" / "axp_update_check" |
| 62 | + |
| 63 | + |
| 64 | +def _should_check(repo: Path) -> bool: |
| 65 | + """Return True if enough time has passed since the last check.""" |
| 66 | + stamp = _get_stamp_path(repo) |
| 67 | + if not stamp.exists(): |
| 68 | + return True |
| 69 | + try: |
| 70 | + last = float(stamp.read_text().strip()) |
| 71 | + return (time.time() - last) >= CHECK_INTERVAL_SECONDS |
| 72 | + except (ValueError, OSError): |
| 73 | + return True |
| 74 | + |
| 75 | + |
| 76 | +def _touch_stamp(repo: Path) -> None: |
| 77 | + """Record the current time as last-checked.""" |
| 78 | + try: |
| 79 | + _get_stamp_path(repo).write_text(str(time.time())) |
| 80 | + except OSError: |
| 81 | + pass |
| 82 | + |
| 83 | + |
| 84 | +def _get_tracking_branch(repo: Path) -> str | None: |
| 85 | + """Get the remote tracking branch for the current branch (e.g. 'origin/main').""" |
| 86 | + branch = _run_git(repo, "rev-parse", "--abbrev-ref", "HEAD") |
| 87 | + if not branch: |
| 88 | + return None |
| 89 | + upstream = _run_git(repo, "rev-parse", "--abbrev-ref", f"{branch}@{{upstream}}") |
| 90 | + return upstream # e.g. "origin/main" |
| 91 | + |
| 92 | + |
| 93 | +def check_for_updates(repo: Path | None = None, force: bool = False) -> UpdateStatus | None: |
| 94 | + """Check if the remote has new commits. |
| 95 | +
|
| 96 | + Returns UpdateStatus if a check was performed, None if skipped (throttled or not a repo). |
| 97 | + """ |
| 98 | + if repo is None: |
| 99 | + repo = _get_repo_root() |
| 100 | + if repo is None: |
| 101 | + return None |
| 102 | + |
| 103 | + if not force and not _should_check(repo): |
| 104 | + return None |
| 105 | + |
| 106 | + # Find tracking branch |
| 107 | + upstream = _get_tracking_branch(repo) |
| 108 | + if not upstream: |
| 109 | + _touch_stamp(repo) |
| 110 | + return None |
| 111 | + |
| 112 | + remote_name = upstream.split("/")[0] if "/" in upstream else "origin" |
| 113 | + |
| 114 | + # Fetch from remote (lightweight, no merge) |
| 115 | + fetch_result = _run_git(repo, "fetch", remote_name, "--quiet") |
| 116 | + if fetch_result is None: |
| 117 | + # Network failure — silently skip |
| 118 | + _touch_stamp(repo) |
| 119 | + return UpdateStatus(error="fetch failed (network issue?)") |
| 120 | + |
| 121 | + _touch_stamp(repo) |
| 122 | + |
| 123 | + # Compare local HEAD vs upstream |
| 124 | + local_ref = _run_git(repo, "rev-parse", "HEAD") or "" |
| 125 | + remote_ref = _run_git(repo, "rev-parse", upstream) or "" |
| 126 | + |
| 127 | + if local_ref == remote_ref: |
| 128 | + return UpdateStatus(local_ref=local_ref, remote_ref=remote_ref) |
| 129 | + |
| 130 | + # Count ahead/behind |
| 131 | + rev_list = _run_git(repo, "rev-list", "--left-right", "--count", f"HEAD...{upstream}") |
| 132 | + ahead, behind = 0, 0 |
| 133 | + if rev_list: |
| 134 | + parts = rev_list.split() |
| 135 | + if len(parts) == 2: |
| 136 | + ahead, behind = int(parts[0]), int(parts[1]) |
| 137 | + |
| 138 | + if behind == 0: |
| 139 | + # Local is ahead or in sync — no update needed |
| 140 | + return UpdateStatus( |
| 141 | + local_ref=local_ref, |
| 142 | + remote_ref=remote_ref, |
| 143 | + ahead_count=ahead, |
| 144 | + ) |
| 145 | + |
| 146 | + # There are updates to pull — find which files changed |
| 147 | + changed_raw = _run_git(repo, "diff", "--name-only", f"HEAD...{upstream}") |
| 148 | + changed_files = changed_raw.splitlines() if changed_raw else [] |
| 149 | + |
| 150 | + # Detect potential conflicts: locally modified files that also changed on remote |
| 151 | + local_modified_raw = _run_git(repo, "diff", "--name-only") |
| 152 | + local_staged_raw = _run_git(repo, "diff", "--name-only", "--cached") |
| 153 | + |
| 154 | + local_dirty: set[str] = set() |
| 155 | + if local_modified_raw: |
| 156 | + local_dirty.update(local_modified_raw.splitlines()) |
| 157 | + if local_staged_raw: |
| 158 | + local_dirty.update(local_staged_raw.splitlines()) |
| 159 | + |
| 160 | + # Also check untracked files that overlap with remote changes |
| 161 | + # (not common but possible if remote adds a file the user also created) |
| 162 | + untracked_raw = _run_git(repo, "ls-files", "--others", "--exclude-standard") |
| 163 | + if untracked_raw: |
| 164 | + local_dirty.update(untracked_raw.splitlines()) |
| 165 | + |
| 166 | + conflict_files = sorted(local_dirty & set(changed_files)) |
| 167 | + |
| 168 | + return UpdateStatus( |
| 169 | + has_update=True, |
| 170 | + local_ref=local_ref, |
| 171 | + remote_ref=remote_ref, |
| 172 | + behind_count=behind, |
| 173 | + ahead_count=ahead, |
| 174 | + changed_files=changed_files, |
| 175 | + conflict_files=conflict_files if conflict_files else None, |
| 176 | + ) |
| 177 | + |
| 178 | + |
| 179 | +def pull_updates(repo: Path | None = None) -> tuple[bool, str]: |
| 180 | + """Run git pull. Returns (success, message).""" |
| 181 | + if repo is None: |
| 182 | + repo = _get_repo_root() |
| 183 | + if repo is None: |
| 184 | + return False, "Not a git repository" |
| 185 | + |
| 186 | + result = _run_git(repo, "pull", "--ff-only", timeout=30) |
| 187 | + if result is not None: |
| 188 | + return True, result |
| 189 | + |
| 190 | + # --ff-only failed, try normal pull |
| 191 | + result = _run_git(repo, "pull", timeout=30) |
| 192 | + if result is not None: |
| 193 | + return True, result |
| 194 | + |
| 195 | + return False, "git pull failed — you may need to resolve conflicts manually" |
0 commit comments