Detect squash-merged branches via Forgejo API
Uses the Forgejo PR API to find branches from merged PRs, catching squash-merged branches that git --merged misses. Also warns about stale local branches that couldn't be confirmed as merged. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
85b3d92fbf
commit
8ce8d39f1c
1 changed files with 205 additions and 84 deletions
|
|
@ -1,9 +1,10 @@
|
|||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.12"
|
||||
# dependencies = ["rich>=13.0.0", "typer>=0.15.0"]
|
||||
# dependencies = ["httpx>=0.28.0", "rich>=13.0.0", "typer>=0.15.0"]
|
||||
# ///
|
||||
#MISE description="Delete branches that have been merged into main (local and remote)"
|
||||
#MISE alias="bc"
|
||||
#USAGE flag "--dry-run" help="Show what would be deleted without deleting"
|
||||
#USAGE flag "--yes" help="Skip confirmation prompt"
|
||||
#USAGE flag "--local-only" help="Only clean up local branches"
|
||||
|
|
@ -11,11 +12,15 @@
|
|||
#USAGE flag "--cutoff <cutoff>" default="30" help="Only delete branches whose HEAD commit is older than N days (default 30)"
|
||||
"""Clean up merged branches locally and on the Forgejo remote.
|
||||
|
||||
Finds branches that have been fully merged into main whose HEAD commit
|
||||
is older than a cutoff (default 30 days), then offers to delete them.
|
||||
Uses git operations directly (SSH for remote), no API token needed.
|
||||
Detects merged branches via two methods:
|
||||
1. git branch --merged (catches fast-forward merges)
|
||||
2. Forgejo API (catches squash-merged PRs)
|
||||
|
||||
Protected branches (main) are never deleted.
|
||||
For remote branches, deletes those confirmed merged by either method.
|
||||
For local branches, deletes those that match a merged remote branch or
|
||||
are detected as merged by git.
|
||||
|
||||
Warns about stale local branches that couldn't be confirmed as merged.
|
||||
|
||||
Usage:
|
||||
mise run branch-cleanup # interactive cleanup (30-day cutoff)
|
||||
|
|
@ -28,11 +33,14 @@ import subprocess
|
|||
from datetime import datetime, timezone
|
||||
from typing import Annotated
|
||||
|
||||
import httpx
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
PROTECTED_BRANCHES = {"main", "master"}
|
||||
FORGE_API = "https://forge.ops.eblu.me/api/v1"
|
||||
REPO = "eblume/blumeops"
|
||||
|
||||
|
||||
def run_git(*args: str) -> str:
|
||||
|
|
@ -58,37 +66,101 @@ def branch_head_age_days(ref: str) -> int | None:
|
|||
return None
|
||||
|
||||
|
||||
def get_merged_local_branches() -> list[str]:
|
||||
"""Get local branches that are fully merged into main."""
|
||||
output = run_git("branch", "--merged", "main")
|
||||
branches = []
|
||||
def get_git_merged_local_branches() -> set[str]:
|
||||
"""Get local branches that are fully merged into main (fast-forward)."""
|
||||
try:
|
||||
output = run_git("branch", "--merged", "main")
|
||||
except subprocess.CalledProcessError:
|
||||
return set()
|
||||
branches = set()
|
||||
for line in output.splitlines():
|
||||
name = line.strip().lstrip("* ")
|
||||
if name and name not in PROTECTED_BRANCHES:
|
||||
branches.append(name)
|
||||
return sorted(branches)
|
||||
branches.add(name)
|
||||
return branches
|
||||
|
||||
|
||||
def get_merged_remote_branches() -> list[str]:
|
||||
"""Get remote branches that are fully merged into origin/main."""
|
||||
output = run_git("branch", "-r", "--merged", "origin/main")
|
||||
branches = []
|
||||
def get_git_merged_remote_branches() -> set[str]:
|
||||
"""Get remote branches that are fully merged into origin/main (fast-forward)."""
|
||||
try:
|
||||
output = run_git("branch", "-r", "--merged", "origin/main")
|
||||
except subprocess.CalledProcessError:
|
||||
return set()
|
||||
branches = set()
|
||||
for line in output.splitlines():
|
||||
name = line.strip()
|
||||
if " -> " in name:
|
||||
continue # skip HEAD pointer
|
||||
continue
|
||||
if not name.startswith("origin/"):
|
||||
continue
|
||||
short = name.removeprefix("origin/")
|
||||
if short not in PROTECTED_BRANCHES:
|
||||
branches.append(short)
|
||||
return sorted(branches)
|
||||
branches.add(short)
|
||||
return branches
|
||||
|
||||
|
||||
def get_all_local_branches() -> set[str]:
|
||||
"""Get all local branch names."""
|
||||
output = run_git("branch")
|
||||
branches = set()
|
||||
for line in output.splitlines():
|
||||
name = line.strip().lstrip("* ")
|
||||
if name and name not in PROTECTED_BRANCHES:
|
||||
branches.add(name)
|
||||
return branches
|
||||
|
||||
|
||||
def get_all_remote_branches() -> set[str]:
|
||||
"""Get all remote branch names (short form, without origin/ prefix)."""
|
||||
output = run_git("branch", "-r")
|
||||
branches = set()
|
||||
for line in output.splitlines():
|
||||
name = line.strip()
|
||||
if " -> " in name:
|
||||
continue
|
||||
if not name.startswith("origin/"):
|
||||
continue
|
||||
short = name.removeprefix("origin/")
|
||||
if short not in PROTECTED_BRANCHES:
|
||||
branches.add(short)
|
||||
return branches
|
||||
|
||||
|
||||
def get_squash_merged_branches(console: Console) -> set[str]:
|
||||
"""Query Forgejo API for branch names from squash-merged PRs."""
|
||||
merged_branches: set[str] = set()
|
||||
page = 1
|
||||
limit = 50
|
||||
|
||||
try:
|
||||
with httpx.Client(timeout=15) as client:
|
||||
while True:
|
||||
resp = client.get(
|
||||
f"{FORGE_API}/repos/{REPO}/pulls",
|
||||
params={"state": "closed", "limit": limit, "page": page},
|
||||
)
|
||||
resp.raise_for_status()
|
||||
prs = resp.json()
|
||||
if not prs:
|
||||
break
|
||||
for pr in prs:
|
||||
if pr.get("merged"):
|
||||
ref = pr.get("head", {}).get("ref", "")
|
||||
if ref and ref not in PROTECTED_BRANCHES:
|
||||
merged_branches.add(ref)
|
||||
page += 1
|
||||
except httpx.HTTPError as e:
|
||||
console.print(f"[yellow]Warning:[/yellow] Failed to query Forgejo API: {e}")
|
||||
console.print("[dim]Falling back to git-only merge detection[/dim]")
|
||||
|
||||
return merged_branches
|
||||
|
||||
|
||||
def delete_local_branch(branch: str) -> tuple[bool, str]:
|
||||
"""Delete a local branch. Returns (success, message)."""
|
||||
try:
|
||||
run_git("branch", "-d", branch)
|
||||
# Use -D since squash-merged branches aren't git-ancestors of main
|
||||
run_git("branch", "-D", branch)
|
||||
return True, "deleted"
|
||||
except subprocess.CalledProcessError as e:
|
||||
return False, e.stderr.strip()
|
||||
|
|
@ -140,76 +212,125 @@ def main(
|
|||
console.print(f"[red]Failed to fetch:[/red] {e.stderr}")
|
||||
raise typer.Exit(1)
|
||||
|
||||
# Gather merge info from both sources
|
||||
console.print("[dim]Checking Forgejo for squash-merged PRs...[/dim]")
|
||||
api_merged = get_squash_merged_branches(console)
|
||||
git_merged_local = get_git_merged_local_branches()
|
||||
git_merged_remote = get_git_merged_remote_branches()
|
||||
|
||||
# Union of all confirmed-merged branch names
|
||||
all_confirmed_merged = api_merged | git_merged_local | git_merged_remote
|
||||
|
||||
do_local = not remote_only
|
||||
do_remote = not local_only
|
||||
|
||||
local_merged = get_merged_local_branches() if do_local else []
|
||||
remote_merged = get_merged_remote_branches() if do_remote else []
|
||||
# Remote: delete if confirmed merged by either method
|
||||
all_remote = get_all_remote_branches() if do_remote else set()
|
||||
remote_to_delete = sorted(all_remote & all_confirmed_merged)
|
||||
|
||||
# Compute ages and filter by cutoff
|
||||
all_names = sorted(set(local_merged) | set(remote_merged))
|
||||
# Local: delete if confirmed merged (git --merged OR matched a merged PR)
|
||||
all_local = get_all_local_branches() if do_local else set()
|
||||
local_to_delete = sorted(all_local & all_confirmed_merged)
|
||||
|
||||
# Compute ages for all candidates
|
||||
all_candidates = sorted(set(local_to_delete) | set(remote_to_delete))
|
||||
branch_ages: dict[str, int | None] = {}
|
||||
|
||||
for name in all_names:
|
||||
# Prefer remote ref for age (more authoritative), fall back to local
|
||||
for name in all_candidates:
|
||||
age = None
|
||||
if name in remote_merged:
|
||||
if name in all_remote:
|
||||
age = branch_head_age_days(f"origin/{name}")
|
||||
if age is None and name in local_merged:
|
||||
if age is None and name in all_local:
|
||||
age = branch_head_age_days(name)
|
||||
branch_ages[name] = age
|
||||
|
||||
# Apply cutoff filter
|
||||
local_branches = [
|
||||
b for b in local_merged
|
||||
if (branch_ages.get(b) is not None and branch_ages[b] >= cutoff) # type: ignore[operator]
|
||||
local_to_delete = [
|
||||
b for b in local_to_delete
|
||||
if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator]
|
||||
]
|
||||
remote_branches = [
|
||||
b for b in remote_merged
|
||||
if (branch_ages.get(b) is not None and branch_ages[b] >= cutoff) # type: ignore[operator]
|
||||
remote_to_delete = [
|
||||
b for b in remote_to_delete
|
||||
if branch_ages.get(b) is not None and branch_ages[b] >= cutoff # type: ignore[operator]
|
||||
]
|
||||
|
||||
skipped = [
|
||||
b for b in all_names
|
||||
if b not in local_branches and b not in remote_branches
|
||||
]
|
||||
filtered_candidates = sorted(set(local_to_delete) | set(remote_to_delete))
|
||||
|
||||
if not local_branches and not remote_branches:
|
||||
console.print("[green]No merged branches older than "
|
||||
f"{cutoff} days to clean up.[/green]")
|
||||
if skipped:
|
||||
console.print(f"[dim]({len(skipped)} merged branches skipped — "
|
||||
f"newer than {cutoff} days)[/dim]")
|
||||
raise typer.Exit(0)
|
||||
skipped_count = len(all_candidates) - len(filtered_candidates)
|
||||
|
||||
# Build summary table
|
||||
table = Table(title=f"Merged branches to delete (older than {cutoff} days)")
|
||||
table.add_column("Branch")
|
||||
table.add_column("Age (days)", justify="right")
|
||||
table.add_column("Local")
|
||||
table.add_column("Remote")
|
||||
if filtered_candidates:
|
||||
# Build summary table
|
||||
table = Table(title=f"Merged branches to delete (older than {cutoff} days)")
|
||||
table.add_column("Branch")
|
||||
table.add_column("Age (days)", justify="right")
|
||||
table.add_column("Method")
|
||||
table.add_column("Local")
|
||||
table.add_column("Remote")
|
||||
|
||||
candidates = sorted(set(local_branches) | set(remote_branches))
|
||||
for branch in candidates:
|
||||
has_local = branch in local_branches
|
||||
has_remote = branch in remote_branches
|
||||
age = branch_ages.get(branch)
|
||||
age_str = str(age) if age is not None else "?"
|
||||
table.add_row(
|
||||
branch,
|
||||
age_str,
|
||||
"[yellow]delete[/yellow]" if has_local else "[dim]-[/dim]",
|
||||
"[yellow]delete[/yellow]" if has_remote else "[dim]-[/dim]",
|
||||
for branch in filtered_candidates:
|
||||
has_local = branch in local_to_delete
|
||||
has_remote = branch in remote_to_delete
|
||||
age = branch_ages.get(branch)
|
||||
age_str = str(age) if age is not None else "?"
|
||||
|
||||
methods = []
|
||||
if branch in git_merged_local or branch in git_merged_remote:
|
||||
methods.append("git")
|
||||
if branch in api_merged:
|
||||
methods.append("api")
|
||||
method_str = "+".join(methods)
|
||||
|
||||
table.add_row(
|
||||
branch,
|
||||
age_str,
|
||||
method_str,
|
||||
"[yellow]delete[/yellow]" if has_local else "[dim]-[/dim]",
|
||||
"[yellow]delete[/yellow]" if has_remote else "[dim]-[/dim]",
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
console.print(
|
||||
f"\n[bold]{len(local_to_delete)}[/bold] local, "
|
||||
f"[bold]{len(remote_to_delete)}[/bold] remote branches to delete"
|
||||
)
|
||||
if skipped_count:
|
||||
console.print(f"[dim]({skipped_count} merged branches skipped — "
|
||||
f"newer than {cutoff} days)[/dim]")
|
||||
else:
|
||||
console.print(f"[green]No merged branches older than "
|
||||
f"{cutoff} days to clean up.[/green]")
|
||||
if skipped_count:
|
||||
console.print(f"[dim]({skipped_count} merged branches skipped — "
|
||||
f"newer than {cutoff} days)[/dim]")
|
||||
|
||||
console.print(table)
|
||||
console.print(
|
||||
f"\n[bold]{len(local_branches)}[/bold] local, "
|
||||
f"[bold]{len(remote_branches)}[/bold] remote branches to delete"
|
||||
)
|
||||
if skipped:
|
||||
console.print(f"[dim]({len(skipped)} merged branches skipped — "
|
||||
f"newer than {cutoff} days)[/dim]")
|
||||
# Warn about stale unmerged local branches
|
||||
if do_local:
|
||||
unmerged_local = all_local - all_confirmed_merged
|
||||
stale_unmerged = []
|
||||
for name in sorted(unmerged_local):
|
||||
age = branch_head_age_days(name)
|
||||
if age is not None and age >= cutoff:
|
||||
stale_unmerged.append((name, age))
|
||||
|
||||
if stale_unmerged:
|
||||
console.print()
|
||||
warn_table = Table(
|
||||
title=f"[yellow]Warning:[/yellow] Stale unmerged local branches "
|
||||
f"(older than {cutoff} days)",
|
||||
title_style="",
|
||||
)
|
||||
warn_table.add_column("Branch")
|
||||
warn_table.add_column("Age (days)", justify="right")
|
||||
for name, age in stale_unmerged:
|
||||
warn_table.add_row(f"[yellow]{name}[/yellow]", str(age))
|
||||
console.print(warn_table)
|
||||
console.print(
|
||||
f"[dim]These {len(stale_unmerged)} branches have no merged PR on "
|
||||
f"Forgejo and are not git-ancestors of main.\n"
|
||||
f"They may contain work-in-progress — inspect manually before deleting.[/dim]"
|
||||
)
|
||||
|
||||
if not filtered_candidates:
|
||||
raise typer.Exit(0)
|
||||
|
||||
if dry_run:
|
||||
console.print("\n[dim]Dry run — no branches were deleted.[/dim]")
|
||||
|
|
@ -220,26 +341,26 @@ def main(
|
|||
console.print("[dim]Aborted.[/dim]")
|
||||
raise typer.Exit(0)
|
||||
|
||||
# Delete local branches
|
||||
if local_branches:
|
||||
console.print("\n[bold]Deleting local branches...[/bold]")
|
||||
for branch in local_branches:
|
||||
ok, msg = delete_local_branch(branch)
|
||||
if ok:
|
||||
console.print(f" [green]✓[/green] {branch}")
|
||||
else:
|
||||
console.print(f" [red]✗[/red] {branch}: {msg}")
|
||||
|
||||
# Delete remote branches
|
||||
if remote_branches:
|
||||
# Delete remote branches first (source of truth)
|
||||
if remote_to_delete:
|
||||
console.print("\n[bold]Deleting remote branches...[/bold]")
|
||||
for branch in remote_branches:
|
||||
for branch in remote_to_delete:
|
||||
ok, msg = delete_remote_branch(branch)
|
||||
if ok:
|
||||
console.print(f" [green]✓[/green] origin/{branch}")
|
||||
else:
|
||||
console.print(f" [red]✗[/red] origin/{branch}: {msg}")
|
||||
|
||||
# Delete local branches
|
||||
if local_to_delete:
|
||||
console.print("\n[bold]Deleting local branches...[/bold]")
|
||||
for branch in local_to_delete:
|
||||
ok, msg = delete_local_branch(branch)
|
||||
if ok:
|
||||
console.print(f" [green]✓[/green] {branch}")
|
||||
else:
|
||||
console.print(f" [red]✗[/red] {branch}: {msg}")
|
||||
|
||||
console.print("\n[green]Done.[/green]")
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue