stale_branches.py
python
sha256:94ef169c149a452bff7c604ded8b280b19bd477c2dabcb56972780b0b784c7aa
Merge 'fix/assignee-sigil-inline' into 'dev' — proposal: As…
Human
2 days ago
| 1 | #!/usr/bin/env python3 |
| 2 | """Branch staleness report across local and configured remotes. |
| 3 | |
| 4 | Usage: |
| 5 | python3 scripts/stale_branches.py [--repo PATH] [--base BRANCH] [--age-days N] [--remote NAME ...] |
| 6 | |
| 7 | Examples: |
| 8 | python3 scripts/stale_branches.py |
| 9 | python3 scripts/stale_branches.py --repo ~/ecosystem/muse --age-days 14 |
| 10 | python3 scripts/stale_branches.py --remote local staging |
| 11 | """ |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import argparse |
| 15 | import json |
| 16 | import subprocess |
| 17 | import sys |
| 18 | from datetime import datetime, timezone |
| 19 | from pathlib import Path |
| 20 | |
| 21 | |
| 22 | # ── muse wrappers ───────────────────────────────────────────────────────────── |
| 23 | |
| 24 | |
| 25 | def _muse(repo: str, *args: str) -> dict: |
| 26 | cmd = ["muse", "-C", repo, *args, "--json"] |
| 27 | result = subprocess.run(cmd, capture_output=True, text=True) |
| 28 | if result.returncode != 0: |
| 29 | raise RuntimeError(f"muse {' '.join(args)} failed:\n{result.stderr.strip()}") |
| 30 | return json.loads(result.stdout) |
| 31 | |
| 32 | |
| 33 | def _rev_count(repo: str, range_expr: str) -> int: |
| 34 | try: |
| 35 | data = _muse(repo, "rev-list", "--count", range_expr) |
| 36 | return int(data.get("count", 0)) |
| 37 | except Exception: |
| 38 | return -1 |
| 39 | |
| 40 | |
| 41 | # ── staleness logic ──────────────────────────────────────────────────────────── |
| 42 | |
| 43 | |
| 44 | def _age_days(committed_at: str | None) -> int | None: |
| 45 | if not committed_at: |
| 46 | return None |
| 47 | try: |
| 48 | ts = datetime.fromisoformat(committed_at.replace("Z", "+00:00")) |
| 49 | return (datetime.now(tz=timezone.utc) - ts).days |
| 50 | except ValueError: |
| 51 | return None |
| 52 | |
| 53 | |
| 54 | def _age_label(days: int | None) -> str: |
| 55 | if days is None: |
| 56 | return "never" |
| 57 | if days == 0: |
| 58 | return "today" |
| 59 | if days == 1: |
| 60 | return "1 day ago" |
| 61 | return f"{days}d ago" |
| 62 | |
| 63 | |
| 64 | # ── remote branch discovery ──────────────────────────────────────────────────── |
| 65 | |
| 66 | |
| 67 | def _remote_branches(repo: str, remotes: list[str]) -> dict[str, dict[str, str]]: |
| 68 | """Return {remote_name: {branch_name: commit_id}} for each reachable remote.""" |
| 69 | result: dict[str, dict[str, str]] = {} |
| 70 | for remote in remotes: |
| 71 | try: |
| 72 | data = _muse(repo, "ls-remote", remote) |
| 73 | result[remote] = data.get("branches", {}) |
| 74 | except Exception as exc: |
| 75 | print(f" [warn] could not reach remote '{remote}': {exc}", file=sys.stderr) |
| 76 | return result |
| 77 | |
| 78 | |
| 79 | def _remote_only_branches( |
| 80 | local_names: set[str], |
| 81 | remote_data: dict[str, dict[str, str]], |
| 82 | ) -> list[tuple[str, str, str]]: |
| 83 | """Branches that exist on a remote but not locally.""" |
| 84 | seen: set[tuple[str, str]] = set() |
| 85 | out: list[tuple[str, str, str]] = [] |
| 86 | for remote, branches in remote_data.items(): |
| 87 | for name, commit_id in branches.items(): |
| 88 | if name not in local_names and (remote, name) not in seen: |
| 89 | seen.add((remote, name)) |
| 90 | out.append((remote, name, commit_id)) |
| 91 | return sorted(out) |
| 92 | |
| 93 | |
| 94 | # ── report ───────────────────────────────────────────────────────────────────── |
| 95 | |
| 96 | |
| 97 | def _col(text: str, width: int) -> str: |
| 98 | return text[:width].ljust(width) |
| 99 | |
| 100 | |
| 101 | def run(repo: str, base: str, age_days_threshold: int, remotes: list[str]) -> None: |
| 102 | repo = str(Path(repo).expanduser().resolve()) |
| 103 | |
| 104 | print(f"\nBranch staleness report") |
| 105 | print(f" repo : {repo}") |
| 106 | print(f" base : {base}") |
| 107 | print(f" stale: last commit > {age_days_threshold} days ago\n") |
| 108 | |
| 109 | # Local branches |
| 110 | branches: list[dict] = _muse(repo, "branch") |
| 111 | if isinstance(branches, dict): |
| 112 | # some versions wrap in a dict — normalise |
| 113 | branches = branches.get("branches", []) |
| 114 | |
| 115 | local_names = {b["name"] for b in branches} |
| 116 | skip = {base, "main"} |
| 117 | |
| 118 | merged: list[tuple] = [] |
| 119 | active: list[tuple] = [] |
| 120 | stale: list[tuple] = [] |
| 121 | |
| 122 | for b in branches: |
| 123 | name = b["name"] |
| 124 | if name in skip: |
| 125 | continue |
| 126 | |
| 127 | commit_id = b.get("commit_id") |
| 128 | committed = b.get("committed_at") |
| 129 | resumable = b.get("resumable", False) |
| 130 | intent = b.get("intent") or "" |
| 131 | days = _age_days(committed) |
| 132 | age = _age_label(days) |
| 133 | |
| 134 | if not commit_id: |
| 135 | ahead, behind = 0, 0 |
| 136 | else: |
| 137 | ahead = _rev_count(repo, f"{base}..{name}") |
| 138 | behind = _rev_count(repo, f"{name}..{base}") |
| 139 | |
| 140 | row = (name, age, days, ahead, behind, resumable, intent) |
| 141 | |
| 142 | if ahead == 0: |
| 143 | merged.append(row) |
| 144 | elif days is not None and days > age_days_threshold: |
| 145 | stale.append(row) |
| 146 | else: |
| 147 | active.append(row) |
| 148 | |
| 149 | # Remote-only branches |
| 150 | remote_data = _remote_branches(repo, remotes) |
| 151 | remote_only = _remote_only_branches(local_names, remote_data) |
| 152 | |
| 153 | # ── print sections ──────────────────────────────────────────────────────── |
| 154 | |
| 155 | hdr = f" {'BRANCH':<40} {'LAST':<12} {'AHEAD':>5} {'BEHIND':>6} FLAGS" |
| 156 | |
| 157 | if merged: |
| 158 | print(f"MERGED INTO {base} — safe to delete ({len(merged)})") |
| 159 | print(hdr) |
| 160 | for name, age, _, ahead, behind, resumable, intent in sorted(merged, key=lambda r: r[1]): |
| 161 | flags = "resumable" if resumable else "" |
| 162 | print(f" {_col(name,40)} {_col(age,12)} {ahead:>5} {behind:>6} {flags}") |
| 163 | print() |
| 164 | |
| 165 | if active: |
| 166 | print(f"ACTIVE ({len(active)})") |
| 167 | print(hdr) |
| 168 | for name, age, _, ahead, behind, resumable, intent in sorted(active, key=lambda r: r[1]): |
| 169 | flags = "resumable" if resumable else "" |
| 170 | hint = f" # {intent}" if intent else "" |
| 171 | print(f" {_col(name,40)} {_col(age,12)} {ahead:>5} {behind:>6} {flags}{hint}") |
| 172 | print() |
| 173 | |
| 174 | if stale: |
| 175 | print(f"STALE — not merged, last commit > {age_days_threshold}d ago ({len(stale)})") |
| 176 | print(hdr) |
| 177 | for name, age, _, ahead, behind, resumable, intent in sorted(stale, key=lambda r: r[1]): |
| 178 | flags = "resumable" if resumable else "" |
| 179 | print(f" {_col(name,40)} {_col(age,12)} {ahead:>5} {behind:>6} {flags}") |
| 180 | print() |
| 181 | |
| 182 | if remote_only: |
| 183 | print(f"REMOTE-ONLY branches ({len(remote_only)})") |
| 184 | print(f" {'REMOTE':<10} {'BRANCH':<40} COMMIT") |
| 185 | for remote, name, commit_id in remote_only: |
| 186 | print(f" {_col(remote,10)} {_col(name,40)} {commit_id[:20]}…") |
| 187 | print() |
| 188 | |
| 189 | if not (merged or active or stale or remote_only): |
| 190 | print("No non-base branches found.") |
| 191 | |
| 192 | |
| 193 | # ── entry point ──────────────────────────────────────────────────────────────── |
| 194 | |
| 195 | |
| 196 | def main() -> None: |
| 197 | parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
| 198 | parser.add_argument("--repo", default=".", help="Path to muse repo (default: cwd)") |
| 199 | parser.add_argument("--base", default="dev", help="Integration branch to compare against (default: dev)") |
| 200 | parser.add_argument("--age-days", type=int, default=30, metavar="N", help="Days threshold for 'stale' (default: 30)") |
| 201 | parser.add_argument("--remote", nargs="*", default=["local", "staging"], metavar="NAME", |
| 202 | help="Remotes to inspect for remote-only branches (default: local staging)") |
| 203 | args = parser.parse_args() |
| 204 | |
| 205 | try: |
| 206 | run( |
| 207 | repo=args.repo, |
| 208 | base=args.base, |
| 209 | age_days_threshold=args.age_days, |
| 210 | remotes=args.remote, |
| 211 | ) |
| 212 | except KeyboardInterrupt: |
| 213 | pass |
| 214 | except Exception as exc: |
| 215 | print(f"error: {exc}", file=sys.stderr) |
| 216 | sys.exit(1) |
| 217 | |
| 218 | |
| 219 | if __name__ == "__main__": |
| 220 | main() |
File History
1 commit
sha256:94ef169c149a452bff7c604ded8b280b19bd477c2dabcb56972780b0b784c7aa
Merge 'fix/assignee-sigil-inline' into 'dev' — proposal: As…
Human
2 days ago