diff --git a/README.md b/README.md index 9e1f44b..46ae740 100644 --- a/README.md +++ b/README.md @@ -8,32 +8,32 @@ A collection of CLI tools I've written for myself that have been useful. Yes, th | Tool | Description | Language | Version | Released | |------|-------------|----------|---------|----------| -| [`asm`](asm/README.md) | AWS Secrets Manager CLI.
Get, create, update, delete, and search secrets.
Supports key/value pairs, files, TLS certs, and binary data. | Python | v1.0.0 | 2026-02-18 | +| [`asm`](asm/README.md) | AWS Secrets Manager CLI (get, create, update, delete, search secrets).
Supports key/value pairs, files, TLS certs, and binary data. | Python | v1.0.0 | 2026-02-18 | | [`aws-secret-replication-check`](aws-secret-replication-check/README.md) | Check Secrets Manager replication status.
Lists all secrets in a region with their cross-region replication state. | Python | v1.0.0 | 2026-02-18 | | [`brew-python`](brew-python/) | Manage Homebrew Python symlinks.
Lists available versions, shows current status, switches between installed versions. | Bash | v1.0.0 | 2026-02-18 | -| [`ec2-search`](ec2-search/) | Audit EC2 instances across AWS accounts.
Displays Profile, Instance ID, Type, AZ, State, and tags.
Supports multiple AWS profiles and custom tag filtering. | Python | v1.0.0 | 2026-02-18 | -| [`ec2-state`](ec2-state/README.md) | View and manage EC2 instance state.
Stop, start, or terminate instances by ID.
Auto-discovers region across all AWS regions. | Python | v1.0.0 | 2026-02-18 | +| [`ec2-search`](ec2-search/) | Search and display EC2 instances across regions and profiles.
Displays Profile, Instance ID, Type, AZ, State, and tags.
Supports multiple AWS profiles and custom tag filtering. | Python | v1.0.0 | 2026-02-18 | +| [`ec2-state`](ec2-state/README.md) | Manage EC2 instance state (get status, stop, start, terminate).
Auto-discovers region across all AWS regions. | Python | v1.0.0 | 2026-02-18 | | [`fcd`](fcd/) | Fuzzy directory finder (interactive cd).
Searches directories by name with interactive selection.
Configurable exclusions, depth control, and k8s context switching. | Bash | v1.0.0 | 2026-02-18 | | [`gh-cleanup-runners`](gh-cleanup-runners/) | Clean up GitHub Actions self-hosted runners.
Identifies and removes offline runners at org or repo level. | Bash | v1.0.0 | 2026-02-18 | | [`gh-runner-audit`](gh-runner-audit/README.md) | Audit GitHub Actions runner usage across an org.
Lists repos with workflow status, runner groups, and runner counts. | Python | v1.0.0 | 2026-02-18 | -| [`gh-search`](gh-search/) | Search GitHub for code and repositories.
Displays matching repos with option to clone and open in VS Code. | Python | v1.0.0 | 2026-02-18 | -| [`git-audit`](git-audit/README.md) | Audit GitHub PRs and direct commits.
Reports filtered by date, file patterns, PR titles, and ticket numbers.
Outputs as list or CSV. | Python | v1.0.0 | 2026-02-18 | -| [`git-cleanup`](git-cleanup/) | Manage GitHub Actions runs and git branches.
Delete all workflow runs for a branch or delete local branches except main/master. | Python | v1.0.0 | 2026-02-18 | +| [`gh-search`](gh-search/) | Search repos in GitHub or a specific organization for a string.
Displays matching repos with option to clone and open in VS Code. | Python | v1.0.0 | 2026-02-18 | +| [`git-audit`](git-audit/README.md) | Generate audit report of GitHub PRs and direct commits with filtering.
Reports filtered by date, file patterns, PR titles, and ticket numbers.
Outputs as list or CSV. | Python | v1.0.0 | 2026-02-18 | +| [`git-cleanup`](git-cleanup/) | Manage branches and GitHub Actions runs.
Delete merged branches, clear workflow runs, and bulk cleanup. | Python | v1.0.0 | 2026-02-18 | | [`git-copy-branch`](git-copy-branch/) | Copy file changes between git branches.
Copies modified/added/deleted files from origin to destination branch.
Auto-commits deleted files. | Python | v1.0.0 | 2026-02-18 | | [`git-force-quit-job`](git-force-quit-job/README.md) | Force-cancel a stuck GitHub Actions run.
Parses the run URL and issues a force-cancel via `gh api`. | Bash | v1.0.0 | 2026-02-18 | | [`git-update-branches`](git-update-branches/) | Batch update git repositories.
Iterates through a directory of repos and updates the default branch.
Preserves state (stash, current branch) and restores after update. | Bash | v1.0.1 | 2026-02-18 | -| [`github-delete-pr-comments`](github-delete-pr-comments/README.md) | Delete all comments on a pull request.
Removes comments by PR URL; reads credentials from `~/.git-credentials`. | Python | v1.0.0 | 2026-02-18 | +| [`github-delete-pr-comments`](github-delete-pr-comments/README.md) | Delete all comments on a GitHub pull request.
Removes comments by PR URL; reads credentials from `~/.git-credentials`. | Python | v1.0.0 | 2026-02-18 | | [`k8s-label-search`](k8s-label-search/) | Search Kubernetes resources by labels and annotations.
Match by key, value, or key=value across multiple clusters.
Supports JSON and matrix output formats. | Python | v1.0.0 | 2026-02-18 | | [`ktail`](ktail/README.md) | Tail Kubernetes logs with filtering and multi-pod support.
Streams logs from all matching pods simultaneously.
Supports colored output, JSON format, and include/exclude regex. | Bash | v1.0.0 | 2026-02-18 | | [`ktx`](ktx/README.md) | Fuzzy search and switch Kubernetes contexts.
Numbered menu when multiple contexts match; optionally switches namespace. | Bash | v1.0.0 | 2026-02-18 | | [`merge-yaml`](merge-yaml/README.md) | Deep-merge two YAML files.
Second file's values override the first.
Optionally annotates changed lines with a comment. | Python | v1.0.0 | 2026-02-18 | | [`open-github-repo`](open-github-repo/README.md) | Search and open a GitHub repository in browser.
Opens current repo without arguments; supports org-scoped search. | Python | v1.0.0 | 2026-02-18 | | [`sort-manifests`](sort-manifests/README.md) | Sort Kubernetes manifests by kind and name.
Sorts multi-document YAML and writes output with a `-sorted` suffix. | Python | v1.0.0 | 2026-02-18 | -| [`sort-yaml-key`](sort-yaml-key/README.md) | Sort a YAML list by a specified key.
Reorders each block so the sort key appears first. | Python | v1.0.0 | 2026-02-18 | -| [`testpod`](testpod/README.md) | Deploy a netshoot debug pod for network troubleshooting.
Creates a `nicolaka/netshoot` pod with optional interactive shell.
Supports easy cleanup with `-d` flag. | Python | v1.0.0 | 2026-02-18 | +| [`sort-yaml-key`](sort-yaml-key/README.md) | Sort a YAML file by a specified key.
Reorders each block so the sort key appears first. | Python | v1.0.0 | 2026-02-18 | +| [`testpod`](testpod/README.md) | Deploy or delete a netshoot debug pod in Kubernetes.
Creates a `nicolaka/netshoot` pod with optional interactive shell.
Supports easy cleanup with `-d` flag. | Python | v1.0.0 | 2026-02-18 | | [`tf-sort`](tf-sort/) | Sort Terraform resource definitions alphabetically.
Organizes Terraform files for better readability and git diffs.
Preserves comments and formatting. | Bash | v1.0.0 | 2026-02-18 | | [`tfdel`](tfdel/README.md) | Delete `.terraform` directories and lock files.
Recursively removes Terraform caches and reports disk space reclaimed. | Python | v1.0.0 | 2026-02-18 | -| [`tfplan-all`](tfplan-all/README.md) | Run `terraform plan` across all Terraform directories.
Parallel init+plan with filtering, output saving, and graceful cancellation. | Python | v1.0.0 | 2026-02-18 | +| [`tfplan-all`](tfplan-all/README.md) | Run terraform init+plan on all subdirectories containing .tf files.
Parallel execution with filtering, output saving, and graceful cancellation. | Python | v1.0.0 | 2026-02-18 | ## Installation diff --git a/ec2-search/ec2-search b/ec2-search/ec2-search index b038236..73f2e42 100755 --- a/ec2-search/ec2-search +++ b/ec2-search/ec2-search @@ -1,54 +1,33 @@ #!/usr/bin/env python3 -"""Search and list EC2 instances across AWS regions and profiles.""" - -from __future__ import annotations import argparse -import concurrent.futures +import subprocess import csv import json -import subprocess import sys -from typing import Any - from tabulate import tabulate +import concurrent.futures -SUBPROCESS_TIMEOUT = 60 - - -def get_regions(profile: str) -> list[str]: - """Get list of all regions accessible by the profile. - - Args: - profile: AWS profile name - - Returns: - List of region names - Raises: - subprocess.CalledProcessError: If AWS CLI fails - subprocess.TimeoutExpired: If command times out +def get_regions(profile): + """ + Gets a list of all regions in which the given profile has access. """ output = subprocess.check_output( ["aws", "ec2", "describe-regions", "--profile", profile], - timeout=SUBPROCESS_TIMEOUT, + timeout=60, ) regions = [region["RegionName"] for region in json.loads(output)["Regions"]] return regions -def get_instances(region: str, profile: str) -> list[dict[str, Any]]: - """Get all EC2 instances in a region. - - Args: - region: AWS region name - profile: AWS profile name - - Returns: - List of instance dictionaries +def get_instances(region, profile): + """ + Gets a list of all EC2 instances from each availability zone of a given region. + Returns a list of dictionaries, where each dictionary contains information about an EC2 instance. """ - # Get availability zones for the region - zones_output = subprocess.check_output( + # Get the list of availability zones for the region + zones = subprocess.check_output( [ "aws", "ec2", @@ -58,12 +37,12 @@ def get_instances(region: str, profile: str) -> list[dict[str, Any]]: "--profile", profile, ], - timeout=SUBPROCESS_TIMEOUT, + timeout=60, ) - zones = [zone["ZoneName"] for zone in json.loads(zones_output)["AvailabilityZones"]] + zones = [zone["ZoneName"] for zone in json.loads(zones)["AvailabilityZones"]] - # Get instances from each zone - instances: list[dict[str, Any]] = [] + # Get information about each EC2 instance in each availability zone + instances = [] for zone in zones: output = subprocess.check_output( [ @@ -79,98 +58,51 @@ def get_instances(region: str, profile: str) -> list[dict[str, Any]]: "--query", "Reservations[].Instances[]", ], - timeout=SUBPROCESS_TIMEOUT, + timeout=120, ) - instances.extend(json.loads(output)) + instances += json.loads(output) return instances -def get_instances_by_region( - region: str, profile: str -) -> list[tuple[dict[str, Any], str]]: - """Get instances for a region with profile info. - - Args: - region: AWS region name - profile: AWS profile name - - Returns: - List of (instance, profile) tuples - """ +def get_instances_by_region(region, profile): print(f"Getting instances in {profile}/{region}...") region_instances = get_instances(region, profile) - return [(instance, profile) for instance in region_instances] + return region_instances -def search_instances_by_profile( - profile: str, region: str | None -) -> list[tuple[dict[str, Any], str]]: - """Search instances for a profile across regions. +def search_instances_by_profile(profile, args): + regions = [args.region] if args.region else get_regions(profile) - Args: - profile: AWS profile name - region: Optional specific region, or None for all regions - - Returns: - List of (instance, profile) tuples - """ - try: - regions = [region] if region else get_regions(profile) - except subprocess.CalledProcessError as e: - print(f"Error getting regions for profile '{profile}': {e}", file=sys.stderr) - return [] - except subprocess.TimeoutExpired: - print(f"Timeout getting regions for profile '{profile}'", file=sys.stderr) - return [] - - instances: list[tuple[dict[str, Any], str]] = [] + # Get information about the instances in each region + instances = [] with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: - futures = { - executor.submit(get_instances_by_region, r, profile): r for r in regions + future_to_region = { + executor.submit(get_instances_by_region, region, profile): region + for region in regions } - for future in concurrent.futures.as_completed(futures): - region_name = futures[future] + for future in concurrent.futures.as_completed(future_to_region): + region_name = future_to_region[future] try: region_instances = future.result() - instances.extend(region_instances) - except subprocess.CalledProcessError as e: - print(f"Error in {profile}/{region_name}: {e}", file=sys.stderr) - except subprocess.TimeoutExpired: - print(f"Timeout in {profile}/{region_name}", file=sys.stderr) + instances += [(instance, profile) for instance in region_instances] except Exception as e: - print( - f"Unexpected error in {profile}/{region_name}: {e}", file=sys.stderr - ) + print(f"Error in {profile}/{region_name}: {e}", file=sys.stderr) return instances -def generate_data( - instance_tuples: list[tuple[dict[str, Any], str]], - show_all_tags: bool = False, - keys: list[str] | None = None, -) -> tuple[list[str], list[list[str]]]: - """Generate table data from instances. - - Args: - instance_tuples: List of (instance, profile) tuples - show_all_tags: Show all tags instead of just Name - keys: Filter to tags containing these keys - - Returns: - Tuple of (headers, rows) +def generate_data(instance_tuples, show_all_tags=False, keys=None): + """ + Generates a list of rows with information about EC2 instances. """ if keys is None: keys = [] - headers = ["Profile", "Instance ID", "Instance Type", "Availability Zone", "State"] - - # Collect tag keys - tag_keys: set[str] = set() - for instance, _ in instance_tuples: + tag_keys = set() + for instance, profile in instance_tuples: for tag in instance.get("Tags", []): if not keys: tag_keys.add(tag["Key"]) @@ -178,15 +110,12 @@ def generate_data( for key in keys: if key in tag["Key"]: tag_keys.add(tag["Key"]) - if show_all_tags: - headers.extend(sorted(tag_keys)) + headers += sorted(tag_keys) else: headers.append("Name") - headers.extend(sorted(tag_keys - {"Name"})) - - # Generate rows - rows: list[list[str]] = [] + headers += sorted(tag_keys - set(["Name"])) + rows = [] for instance, profile in instance_tuples: row = [ profile, @@ -195,45 +124,46 @@ def generate_data( instance["Placement"]["AvailabilityZone"], instance["State"]["Name"], ] - tag_values = {tag["Key"]: tag["Value"] for tag in instance.get("Tags", [])} - if show_all_tags: tag_row = [tag_values.get(header, "") for header in headers[5:]] else: name_tag = tag_values.get("Name", "") row.append(name_tag) tag_row = [tag_values.get(header, "") for header in headers[6:]] - - row.extend(tag_row) + row += tag_row rows.append(row) - return headers, rows -def save_instances_csv( - headers: list[str], rows: list[list[str]], filename: str -) -> None: - """Save instance data to CSV file. - - Args: - headers: Column headers - rows: Data rows - filename: Output filename +def save_instances_csv(headers, rows, filename): + """ + Saves a CSV file with information about EC2 instances. """ with open(filename, "w", newline="") as csvfile: writer = csv.writer(csvfile) writer.writerow(headers) - writer.writerows(rows) - print(f"Saved to {filename}") + for row in rows: + writer.writerow(row) -def main() -> int: - """Main entry point.""" +def main(): + """ + Gets a list of all EC2 instances from each availability zone of each region, and includes their tags and running status. + Outputs the data as a table, and saves it as a CSV file if a filename is provided. + + Command-line arguments: + -r, --region AWS region to search. If not provided, searches all regions. + -p, --profile AWS profile to use. Required. + -c, --csv CSV filename to save the data. If not provided, does not save to CSV. + --all-tags Show all tags of an instance, instead of just the Name tag. + -k, --keys Comma-separated list of keys to search in tags. Only show tags that contain one of the specified keys. + """ + # Parse command-line arguments parser = argparse.ArgumentParser( description="Get information about EC2 instances.", formatter_class=argparse.RawDescriptionHelpFormatter, - epilog="""Examples: + epilog="""Example usage: ec2-search -p primary,secondary ec2-search -r us-west-2 -p primary,secondary ec2-search -p primary,secondary -c ec2-instances.csv @@ -243,68 +173,67 @@ def main() -> int: parser.add_argument( "-r", "--region", - help="AWS region (default: all regions)", + type=str, + help="AWS region to search. If not provided, searches all regions.", ) parser.add_argument( "-p", "--profile", + type=str, + help="Comma-separated list of AWS profiles to use. Required.", required=True, - help="Comma-separated AWS profiles", ) parser.add_argument( "-c", "--csv", - help="Save to CSV file", + type=str, + help="CSV filename to save the data. If not provided, does not save to CSV.", ) parser.add_argument( "--all-tags", action="store_true", - help="Show all tags (not just Name)", + help="Show all tags of an instance, instead of just the Name tag.", ) parser.add_argument( "-k", "--keys", - help="Comma-separated tag keys to filter", + type=str, + help="Comma-separated list of keys to search in tags. Only show tags that contain one of the specified keys.", ) - args = parser.parse_args() + # Get a list of regions to search profiles = args.profile.split(",") - all_instances: list[tuple[dict[str, Any], str]] = [] + all_instances = [] - # Search profiles in parallel - with concurrent.futures.ThreadPoolExecutor(max_workers=len(profiles)) as executor: - futures = { - executor.submit(search_instances_by_profile, p, args.region): p - for p in profiles + # Run searches for each profile in parallel + with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: + future_to_profile = { + executor.submit(search_instances_by_profile, profile, args): profile + for profile in profiles } - for future in concurrent.futures.as_completed(futures): - profile_name = futures[future] + for future in concurrent.futures.as_completed(future_to_profile): + profile_name = future_to_profile[future] try: profile_instances = future.result() all_instances.extend(profile_instances) except Exception as e: print(f"Error searching profile '{profile_name}': {e}", file=sys.stderr) - if not all_instances: - print("No instances found.") - return 0 - - # Generate and display data - keys = args.keys.split(",") if args.keys else None - headers, rows = generate_data(all_instances, args.all_tags, keys) + # Generate the data + headers, rows = generate_data( + all_instances, args.all_tags, args.keys.split(",") if args.keys else [] + ) + # Print the instances table print(tabulate(rows, headers=headers)) - print(f"\nTotal instances: {len(rows)}") - # Save CSV if requested + # Save the instances CSV file if filename is provided if args.csv: - filename = args.csv if args.csv.endswith(".csv") else f"{args.csv}.csv" + filename = args.csv if args.csv.endswith(".csv") else f"{args.csv}-ec2.csv" save_instances_csv(headers, rows, filename) - return 0 - if __name__ == "__main__": - sys.exit(main()) + main() diff --git a/ec2-state/ec2-state b/ec2-state/ec2-state index 06e09a6..c1fdf5b 100755 --- a/ec2-state/ec2-state +++ b/ec2-state/ec2-state @@ -7,21 +7,19 @@ from concurrent.futures import ThreadPoolExecutor, as_completed def get_instance_region(profile, region, instance_id): - cmd = [ - "aws", - "ec2", - "describe-instances", - "--instance-ids", - instance_id, - "--region", - region, - ] - if profile: - cmd.extend(["--profile", profile]) try: - output = subprocess.check_output( - cmd, stderr=subprocess.PIPE, universal_newlines=True - ) + cmd = [ + "aws", + "ec2", + "describe-instances", + "--instance-ids", + instance_id, + "--region", + region, + ] + if profile: + cmd.extend(["--profile", profile]) + output = subprocess.check_output(cmd, stderr=subprocess.PIPE, text=True) response = json.loads(output) return response except subprocess.CalledProcessError as e: @@ -44,8 +42,7 @@ def get_instance_regions(profile, instance_ids): if profile: cmd.extend(["--profile", profile]) regions = [ - x["RegionName"] - for x in json.loads(subprocess.check_output(cmd, universal_newlines=True)) + x["RegionName"] for x in json.loads(subprocess.check_output(cmd, text=True)) ] instance_responses = [] @@ -132,13 +129,14 @@ def process_instance_data(response, profile, args): # Refresh instance state refreshed_response = get_instance_region(profile, region, instance_id) - refreshed_instance = refreshed_response["Reservations"][0]["Instances"][ - 0 - ] - refreshed_state = refreshed_instance["State"]["Name"] - print( - f"{instance_id} in ({availability_zone}) has a NEW state of [{refreshed_state}]" - ) + if refreshed_response: + refreshed_instance = refreshed_response["Reservations"][0][ + "Instances" + ][0] + refreshed_state = refreshed_instance["State"]["Name"] + print( + f"{instance_id} in ({availability_zone}) has a NEW state of [{refreshed_state}]" + ) print("-" * 60) @@ -148,23 +146,11 @@ def main(): description="Manage EC2 instances", formatter_class=argparse.RawTextHelpFormatter, epilog="""Examples: -# Get the running state and availability zone of EC2 instances -ec2-state -p my-profile -i i-1234567890abcdef0,i-0987654321fedcba0 - -# Use a specific AWS profile -ec2-state -p my-profile -i i-1234567890abcdef0 - -# Stop instances -ec2-state -s -p my-profile -i i-1234567890abcdef0,i-0987654321fedcba0 - -# Start instances -ec2-state -r -p my-profile -i i-1234567890abcdef0,i-0987654321fedcba0 - -# Terminate instances -ec2-state -t -p my-profile -i i-1234567890abcdef0,i-0987654321fedcba0 - -# Specify region explicitly -ec2-state -p my-profile -i i-1234567890abcdef0 --region us-west-2 + ec2-state -p my-profile -i i-1234567890abcdef0,i-0987654321fedcba0 + ec2-state -s -p my-profile -i i-1234567890abcdef0 + ec2-state -r -p my-profile -i i-1234567890abcdef0 + ec2-state -t -p my-profile -i i-1234567890abcdef0 + ec2-state -p my-profile -i i-1234567890abcdef0 --region us-west-2 """, ) parser.add_argument("-p", "--profile", required=True, help="AWS profile to use") @@ -180,30 +166,25 @@ ec2-state -p my-profile -i i-1234567890abcdef0 --region us-west-2 args = parser.parse_args() - # Set AWS profile profile = args.profile - - # Set instance IDs instance_ids = args.instances.split(",") - # Check if a region was specified if args.region: - # Use the specified region and don't get instance regions instance_responses = [ - get_instance_region(profile, args.region, instance_id) - for instance_id in instance_ids + r + for r in ( + get_instance_region(profile, args.region, instance_id) + for instance_id in instance_ids + ) + if r is not None ] else: - # Get instance regions and perform actions concurrently print("Now finding regions for your instances...", end="", flush=True) instance_responses = get_instance_regions(profile, instance_ids) print(" ...done") print("-" * 60) - # Process instance data for response in instance_responses: - if response is None: - continue process_instance_data(response, profile, args) diff --git a/gh-search/gh-search b/gh-search/gh-search index fa1b3b6..209274b 100755 --- a/gh-search/gh-search +++ b/gh-search/gh-search @@ -1,141 +1,180 @@ #!/usr/bin/env python3 -"""Search GitHub repositories for code patterns.""" - -from __future__ import annotations - import argparse import os import subprocess import sys import urllib.parse from pathlib import Path -from typing import Final import requests -DEFAULT_CLONE_DIR: Final[str] = "~/git" -DEFAULT_GITHUB_API_BASE_URL: Final[str] = "https://api.github.com" -REQUEST_TIMEOUT: Final[int] = 30 +DEFAULT_CLONE_DIR = "~/git" +DEFAULT_GITHUB_API_BASE_URL = "https://api.github.com" + +def get_token_from_env(): + return os.getenv("GITHUB_TOKEN") -def get_access_token(token: str | None = None, token_file: str | None = None) -> str: - """Get GitHub access token from argument, file, or default location. - Args: - token: Direct token string - token_file: Path to file containing token +def get_token_from_gitconfig(): + try: + result = subprocess.run( + ["git", "config", "--get", "github.token"], capture_output=True, text=True + ) + if result.returncode == 0: + return result.stdout.strip() + except Exception: + pass + return None + - Returns: - GitHub access token +def get_token_from_gh_cli(): + try: + result = subprocess.run(["gh", "auth", "token"], capture_output=True, text=True) + if result.returncode == 0: + return result.stdout.strip() + except Exception: + pass + return None - Raises: - FileNotFoundError: If token file doesn't exist - ValueError: If token file is empty - """ + +def get_token_from_1password(): + try: + result = subprocess.run( + ["op", "read", "op://Personal/GitHub/token"], capture_output=True, text=True + ) + if result.returncode == 0: + return result.stdout.strip() + except Exception: + pass + return None + + +def get_access_token(token=None, token_file=None): if token: return token + env_token = get_token_from_env() + if env_token: + return env_token + + gh_token = get_token_from_gh_cli() + if gh_token: + return gh_token + + git_token = get_token_from_gitconfig() + if git_token: + return git_token + + op_token = get_token_from_1password() + if op_token: + return op_token + if token_file: - path = Path(os.path.expanduser(token_file)) + token_file = Path(os.path.expanduser(token_file)) else: - path = Path(os.path.expanduser("~/.ssh/.ghe")) + token_file = Path(os.path.expanduser("~/.github_token")) + + if not token_file.is_file(): + print( + "Error: no GitHub token found. Checked: GITHUB_TOKEN env, gh CLI, " + "git config, 1Password, and token file.", + file=sys.stderr, + ) + sys.exit(1) + + token = token_file.read_text().strip() + if not token: + print(f"Error: token file '{token_file}' is empty.", file=sys.stderr) + sys.exit(1) + + return token + + +def search_repositories( + query, + access_token, + organization=None, + github_api_base_url=DEFAULT_GITHUB_API_BASE_URL, +): + headers = {"Authorization": f"token {access_token}"} + org_query = f"org:{organization}" if organization else "" + search_query = f"{org_query} {query}".strip() + + url = f"{github_api_base_url}/search/repositories?q={urllib.parse.quote(search_query)}" + response = requests.get(url, headers=headers, timeout=30) + + if response.status_code != 200: + print(f"Error: {response.status_code}", file=sys.stderr) + return [] - if not path.is_file(): - raise FileNotFoundError(f"Access token file not found: {path}") + data = response.json() + repos = [] - content = path.read_text().strip() - if not content: - raise ValueError(f"Access token file is empty: {path}") + for repo in data.get("items", []): + repos.append( + { + "name": repo["full_name"], + "description": repo["description"], + "url": repo["html_url"], + "stars": repo["stargazers_count"], + } + ) - return content + return repos def search_repos( - query: str, - access_token: str, - organization: str | None = None, - github_api_base_url: str = DEFAULT_GITHUB_API_BASE_URL, -) -> list[tuple[str, str, str]]: - """Search GitHub for code matching query. - - Args: - query: Search string - access_token: GitHub API token - organization: Optional org to limit search - github_api_base_url: API base URL - - Returns: - List of (full_name, html_url, clone_url) tuples - """ - headers = {"Authorization": f"Bearer {access_token}"} - - # Build query with proper URL encoding - search_query = f"org:{organization} {query}" if organization else query - encoded_query = urllib.parse.quote(search_query) - url = f"{github_api_base_url}/search/code?q={encoded_query}" - - try: - response = requests.get(url, headers=headers, timeout=REQUEST_TIMEOUT) - response.raise_for_status() - except requests.exceptions.Timeout: - print(f"Error: Request timed out after {REQUEST_TIMEOUT}s", file=sys.stderr) - return [] - except requests.exceptions.HTTPError as e: - print(f"Error: GitHub API returned {e.response.status_code}", file=sys.stderr) - if e.response.status_code == 401: - print(" Check your access token is valid", file=sys.stderr) - elif e.response.status_code == 403: - print(" Rate limit may be exceeded", file=sys.stderr) - return [] - except requests.exceptions.RequestException as e: - print(f"Error: Request failed: {e}", file=sys.stderr) + query, + access_token, + organization=None, + github_api_base_url=DEFAULT_GITHUB_API_BASE_URL, +): + headers = {"Authorization": f"token {access_token}"} + org_query = f"org:{organization}" if organization else "" + search_query = f"{org_query} {query}".strip() + url = f"{github_api_base_url}/search/code?q={urllib.parse.quote(search_query)}" + response = requests.get(url, headers=headers, timeout=30) + + if response.status_code != 200: + print(f"Error: {response.status_code}", file=sys.stderr) return [] data = response.json() - repos: set[tuple[str, str, str]] = set() + repos = set() for item in data.get("items", []): repo = item["repository"] clone_url = repo["html_url"] + ".git" repos.add((repo["full_name"], repo["html_url"], clone_url)) - return sorted(repos) - + return sorted(list(repos)) -def clone_repos( - repos: list[tuple[str, str, str]], clone_directory: str | Path -) -> list[tuple[str, str, str]]: - """Clone repositories to local directory. - Args: - repos: List of (full_name, html_url, clone_url) tuples - clone_directory: Target directory for clones +def clone_repos(repos, clone_directory): + clone_directory = Path(os.path.expanduser(clone_directory)).resolve() + clone_directory.mkdir(parents=True, exist_ok=True) - Returns: - List of (full_name, html_url, local_path) tuples - """ - clone_dir = Path(os.path.expanduser(str(clone_directory))).resolve() - clone_dir.mkdir(parents=True, exist_ok=True) - - print(f"Cloning repositories to {clone_dir}...") + print(f"Cloning repositories to {clone_directory}...") print("-" * 80) - all_repos: list[tuple[str, str, str]] = [] + all_repos = [] for repo_name, repo_url, clone_url in repos: - # Remove organization prefix from path - repo_path = clone_dir / repo_name.split("/")[-1] - + # Remove organization name from repository path + repo_name_parts = repo_name.split("/") + repo_path = clone_directory / Path(*repo_name_parts[1:]) if not repo_path.exists(): print(f"Cloning {repo_name}...") - try: - subprocess.run( - ["git", "clone", "-q", clone_url, str(repo_path)], - check=True, - capture_output=True, - ) - except subprocess.CalledProcessError as e: - print(f" Failed to clone: {e.stderr.decode()}", file=sys.stderr) + result = subprocess.run( + ["git", "clone", "-q", clone_url, str(repo_path)], + capture_output=True, + text=True, + ) + if result.returncode != 0: + print(f"Error: failed to clone {repo_name}", file=sys.stderr) + if result.stderr: + print(result.stderr, file=sys.stderr) continue else: print(f"Repo {repo_name} already exists. Skipping clone.") @@ -145,33 +184,25 @@ def clone_repos( return all_repos -def open_in_vscode( - repos: list[tuple[str, str, str]], new_instance: bool = False -) -> None: - """Open repositories in VS Code. - - Args: - repos: List of (name, url, local_path) tuples - new_instance: If True, open in new window - """ - paths = [repo_path for _, _, repo_path in repos] +def open_in_vscode(repos, new_instance=False): + cmd = ["code"] if new_instance: - args = ["code", "-n"] + paths + cmd.extend(["-n"] + [str(repo_path) for _, _, repo_path in repos]) else: - args = ["code", "--add"] + paths + cmd.extend(["--add"] + [str(repo_path) for _, _, repo_path in repos]) try: - subprocess.run(args, check=True) - print(f"Opened {len(repos)} repositories in VS Code.\n") - except subprocess.CalledProcessError: - print("Failed to open VS Code", file=sys.stderr) + subprocess.run(cmd, check=True) + mode = "new instance" if new_instance else "current instance" + print(f"Opened {len(repos)} repositories in {mode} of VS Code.\n") except FileNotFoundError: - print("VS Code 'code' command not found in PATH", file=sys.stderr) + print("Error: 'code' command not found. Is VS Code installed?", file=sys.stderr) + except subprocess.CalledProcessError as e: + print(f"Error: VS Code exited with code {e.returncode}", file=sys.stderr) -def main() -> int: - """Main entry point.""" +def main(): parser = argparse.ArgumentParser( prog="gh-search", description="Search repos in GitHub or a specific organization for a specific string.", @@ -181,52 +212,92 @@ def main() -> int: gh-search "runs-on: [ docker ]" # Search a specific organization - gh-search "runs-on: [ docker ]" -o myorg + gh-search "runs-on: [ docker ]" -o peng - # Clone and open in VS Code + # Clone returned repositories to the default directory and open them in Visual Studio Code gh-search "runs-on: [ docker ]" -c -e - # Clone to custom directory + # Clone returned repositories to a custom directory and open them in the current instance of Visual Studio Code gh-search "runs-on: [ docker ]" -c -e --dir ~/myrepos + + # Use a custom token and a custom directory for cloning + gh-search "runs-on: [ docker ]" -t YOUR_GITHUB_TOKEN -d ~/custom_directory """, ) - parser.add_argument("query", help="Search query string (exact match)") parser.add_argument( - "-c", "--clone", action="store_true", help="Clone repositories to directory" + "query", help="The search query string to search for in the repos. Exact match." + ) + parser.add_argument( + "-c", + "--clone", + help="Clone repositories (if not already cloned) to the specified directory.", + action="store_true", ) parser.add_argument( "-d", "--dir", + help="Directory to clone repositories into. Default is ~/git.", default=DEFAULT_CLONE_DIR, - help="Clone directory (default: ~/git)", ) parser.add_argument( - "-e", "--edit", action="store_true", help="Open in VS Code (requires -c)" + "-e", + "--edit", + help="Open the cloned repositories in Visual Studio Code. \nRequires -c.", + action="store_true", ) - parser.add_argument("-o", "--org", help="Limit search to organization") - parser.add_argument("-t", "--token", help="GitHub access token") - parser.add_argument("-tf", "--token-file", help="Path to token file") parser.add_argument( - "--new", action="store_true", help="Open in new VS Code window (requires -e)" + "-o", + "--org", + help="The organization to search in. If not specified, search all of GitHub.", + default=None, ) parser.add_argument( - "-u", "--url", default=DEFAULT_GITHUB_API_BASE_URL, help="GitHub API base URL" + "-r", "--repo", help="Search for repositories by name", action="store_true" + ) + parser.add_argument( + "-t", + "--token", + help="GitHub access token. If not specified, the script will use the token from the default token file (~/.github_token).", + default=None, + ) + parser.add_argument( + "-tf", + "--token-file", + help="Path to the GitHub access token file. If not specified, the script will use the default token file (~/.github_token).", + default=None, + ) + parser.add_argument( + "--new", + help="Open the cloned repositories in a new instance of Visual Studio Code instead of adding them to the existing instance. \nRequires -e.", + action="store_true", + ) + parser.add_argument( + "-u", + "--url", + help="The base URL for the GitHub API. Default is " + + DEFAULT_GITHUB_API_BASE_URL, + default=DEFAULT_GITHUB_API_BASE_URL, ) args = parser.parse_args() - # Validate flags if args.edit and not args.clone: - parser.error("--edit requires --clone") - - # Get token - try: - access_token = get_access_token(token=args.token, token_file=args.token_file) - except (FileNotFoundError, ValueError) as e: - print(f"Error: {e}", file=sys.stderr) - return 1 + parser.error("The --edit flag requires the --clone flag to be specified.") + + access_token = get_access_token(token=args.token, token_file=args.token_file) + + if args.repo: + repos = search_repositories(args.query, access_token, args.org, args.url) + print("\nMatching repositories:\n" + "-" * 80) + for repo in repos: + print(f"\nName: {repo['name']}") + print(f"URL: {repo['url']}") + print(f"Stars: {repo['stars']}") + if repo["description"]: + print(f"Description: {repo['description']}") + print(f"\nFound {len(repos)} repositories\n") + return - # Search repos = search_repos( query=args.query, access_token=access_token, @@ -234,27 +305,24 @@ def main() -> int: github_api_base_url=args.url, ) - if not repos: - print("No repositories found.") - return 0 - - # Clone if requested if args.clone: - repos = clone_repos(repos, args.dir) + clone_directory = Path(os.path.expanduser(args.dir)) + repos = clone_repos(repos, clone_directory) print("-" * 80) - # Open in VS Code if requested if args.edit: - open_in_vscode(repos, new_instance=args.new) + if args.new: + open_in_vscode(repos, new_instance=True) + else: + open_in_vscode(repos) - # Print results print("Returned Repos:\n") + for repo_name, repo_url, _ in repos: print(f"{repo_name}: {repo_url}") - print(f"\nNumber of repos found: {len(repos)}\n") - return 0 + print(f"\nNumber of repos found: {len(repos)}\n") if __name__ == "__main__": - sys.exit(main()) + main() diff --git a/git-audit/git-audit b/git-audit/git-audit index 0e7900b..03518fa 100755 --- a/git-audit/git-audit +++ b/git-audit/git-audit @@ -1,4 +1,5 @@ #!/usr/bin/env python3 +"""Generate audit report of GitHub PRs and direct commits with filtering.""" import argparse import csv @@ -6,6 +7,7 @@ import json import os import re import subprocess +import sys from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta from io import StringIO @@ -48,7 +50,7 @@ def process_items_concurrently(prs, args, default_branch, max_workers=5): if result: processed_items.append(result) except Exception as e: - print(f"Error processing item {pr_or_commit}: {e}") + print(f"Error processing item {pr_or_commit}: {e}", file=sys.stderr) return processed_items @@ -179,8 +181,8 @@ def process_pr_title(title): return process_commit_message(title) -def extract_ticket(text, ticket_pattern): - """Extract ticket number from branch name or commit message.""" +def extract_ticket(text, ticket_pattern=r"[A-Z]+-\d+"): + """Extract ticket reference from text using the given pattern.""" match = re.search(ticket_pattern, text) return match.group(0) if match else "missing" @@ -238,6 +240,7 @@ def construct_result_dict( "PR Number": pr_number, "Branch": branch_name, "Ticket": ticket, + "RFC": "", "Notes": "", } @@ -325,25 +328,23 @@ def parse_arguments(): "This script fetches and processes pull requests (PRs) and direct commits to the default branch " "within a specified date range. It allows filtering based on file patterns, title patterns, and " "description patterns. The output can be formatted as a list or CSV file, and includes additional " - "metadata such as issue tickets and application names." + "metadata such as JIRA tickets and application names." ), formatter_class=argparse.RawDescriptionHelpFormatter, epilog=( "Examples:\n" - " To find all PRs merged between 2021-01-01 and 2021-12-31 with filenames containing 'api':\n" - " git-audit -o csv --start-date 2021-01-01 --end-date 2021-12-31 --file-pattern 'api'\n\n" + " To find all PRs merged between 2021-01-01 and 2021-12-31 with filenames containing 'argocd':\n" + " git-audit run -o csv --start-date 2021-01-01 --end-date 2021-12-31 --file-pattern 'argocd'\n\n" " To process specific SHAs and include an application name:\n" - " git-audit -sha abc123 def456 --app MyApp\n\n" - " To use a custom ticket pattern:\n" - " git-audit --ticket-pattern 'PROJ-\\d+'\n\n" + " git-audit run -sha abc123 def456 --app MyApp\n\n" "Defaults:\n" " - Date Range: Last 6 months\n" " - Output Format: List\n" - " - Ticket Pattern: [A-Z]+-\\d+ (standard JIRA/issue format)\n" " - Default Branch: Automatically detected from the repository\n" ), ) + parser.add_argument("run", help="Required to execute the script") parser.add_argument( "-o", "--output", @@ -396,7 +397,7 @@ def parse_arguments(): "--ticket-pattern", metavar="PATTERN", default=r"[A-Z]+-\d+", - help="Regex pattern to extract ticket numbers from branch names (default: [A-Z]+-\\d+)", + help="Regex pattern to extract ticket references from branch names (default: [A-Z]+-\\d+)", ) return parser.parse_args() @@ -418,8 +419,6 @@ def get_default_branch(): def fetch_prs_and_commits(args, default_branch): """Fetch PRs and direct commits based on input method.""" - start_date = f"{args.start_date}T00:00:00Z" - end_date = f"{args.end_date}T23:59:59Z" prs = [] if args.sha: for sha in args.sha: @@ -438,7 +437,12 @@ def fetch_prs_and_commits(args, default_branch): else: prs.append((None, sha)) else: + # Convert dates to GitHub API format + start_date = f"{args.start_date}T00:00:00Z" + end_date = f"{args.end_date}T23:59:59Z" prs = [(pr, None) for pr in get_pr_data(start_date, end_date)] + + if not args.sha and not args.file: direct_commits = get_direct_main_commits(start_date, end_date) for commit in direct_commits: prs.append((None, commit["sha"])) @@ -458,6 +462,7 @@ def output_results(processed_items, output_format): "PR Number", "Branch", "Ticket", + "RFC", "Notes", ] diff --git a/git-cleanup/git-cleanup b/git-cleanup/git-cleanup index e4aeddb..f654d96 100755 --- a/git-cleanup/git-cleanup +++ b/git-cleanup/git-cleanup @@ -1,8 +1,11 @@ #!/usr/bin/env python3 +import functools import json +import os import subprocess -from typing import Optional +import sys +from contextlib import contextmanager import typer @@ -13,6 +16,63 @@ app = typer.Typer( ) +def _requesting_help() -> bool: + return bool({"-h", "--help"} & set(sys.argv)) + + +@app.callback() +def main( + path: str | None = typer.Option( + None, + "--path", + "-C", + "-p", + help="Run in the specified directory (like git -C)", + ), +): + """Git utility tools for managing branches and GitHub Actions runs""" + if _requesting_help(): + return + + if path: + target = os.path.expanduser(path) + if not os.path.isdir(target): + typer.echo(f"Error: '{target}' is not a directory", err=True) + raise typer.Exit(1) + os.chdir(target) + + # Validate we're in a git repo + git_check = subprocess.run( + ["git", "rev-parse", "--git-dir"], + capture_output=True, + text=True, + ) + if git_check.returncode != 0: + typer.echo( + f"Skipping '{os.getcwd()}': not a git repository", + err=True, + ) + raise typer.Exit(0) + + +@functools.lru_cache(maxsize=1) +def _repo_name() -> str: + """Return the basename of the current git repo root""" + result = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + ) + if result.returncode == 0: + return os.path.basename(result.stdout.strip()) + return os.path.basename(os.getcwd()) + + +def log(message: str, err: bool = False) -> None: + """Print a message prefixed with the repo name""" + typer.echo(f"[{_repo_name()}] {message}", err=err) + + def run_command(cmd: list[str], check: bool = True) -> subprocess.CompletedProcess: """ Execute a shell command and return the result @@ -24,26 +84,107 @@ def run_command(cmd: list[str], check: bool = True) -> subprocess.CompletedProce Returns: CompletedProcess instance with command output """ - return subprocess.run(cmd, check=check, capture_output=True, text=True, timeout=60) + return subprocess.run( + cmd, + check=check, + capture_output=True, + text=True, + timeout=60, + ) def confirm_action(message: str) -> bool: """ - Ask user for confirmation before proceeding + Ask user for confirmation before proceeding. + Reads from /dev/tty so prompts work even when stdin is piped. + """ + try: + tty = open("/dev/tty") + except OSError: + log("No TTY for confirmation -- use -f to skip prompts", err=True) + return False + try: + sys.stderr.write(f"[{_repo_name()}] {message} (y/N): ") + sys.stderr.flush() + response = tty.readline().strip().lower() + return response in ["y", "yes"] + finally: + tty.close() - Args: - message: Confirmation message to display - Returns: - bool: True if user confirms, False otherwise +def get_default_branch() -> str: """ - response = input(f"{message} (y/N): ").lower() - return response in ["y", "yes"] + Get the default branch configured for origin + Returns: + str: Default branch name + """ + result = run_command(["git", "remote", "show", "origin"]) + for line in result.stdout.splitlines(): + if "HEAD branch" in line: + return line.split()[-1] + + typer.echo("Error: Could not determine default branch", err=True) + raise typer.Exit(1) + + +@contextmanager +def preserve_branch(): + """Save and restore the current branch and stash dirty work""" + original = run_command(["git", "rev-parse", "--abbrev-ref", "HEAD"]).stdout.strip() + + # Only stash tracked changes — untracked-only won't create a stash entry + status = run_command(["git", "status", "--porcelain"]).stdout.strip() + has_tracked_changes = any( + not line.startswith("??") for line in status.splitlines() if line.strip() + ) + stashed = False + if has_tracked_changes: + run_command(["git", "stash", "push", "-m", "git-cleanup: auto-stash"]) + stashed = True -@app.command() + try: + yield original + finally: + # Fully defensive — never mask the original exception + try: + current = subprocess.run( + ["git", "rev-parse", "--abbrev-ref", "HEAD"], + capture_output=True, + text=True, + ) + if current.returncode == 0 and current.stdout.strip() != original: + subprocess.run( + ["git", "checkout", original], + capture_output=True, + text=True, + ) + except Exception: + pass + if stashed: + pop = subprocess.run( + ["git", "stash", "pop"], + capture_output=True, + text=True, + ) + if pop.returncode != 0: + log(f"Warning: stash pop failed: {pop.stderr.strip()}", err=True) + log("Recover with: git stash list && git stash apply", err=True) + + +@app.command( + epilog=( + "Examples:\n\n" + " # Delete runs for current branch, keep 10 most recent\n" + " git-cleanup clear-branch-runs\n\n" + " # Delete all runs for a specific branch\n" + " git-cleanup clear-branch-runs my-feature -k 0 -f\n\n" + " # Target a different repo\n" + " git-cleanup -C ~/git/other-repo clear-branch-runs" + ), +) def clear_branch_runs( - branch: Optional[str] = typer.Argument(None, help="Branch name to clear runs from"), + branch: str | None = typer.Argument(None, help="Branch name to clear runs from"), force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation prompt"), limit: int = typer.Option( 1000, "--limit", "-l", help="Maximum number of runs to fetch (default: 1000)" @@ -55,40 +196,41 @@ def clear_branch_runs( """Clear all GitHub Actions runs for a specific branch with proper pagination""" try: if keep_latest < 0: - typer.echo("Error: --keep must be non-negative", err=True) + log("Error: --keep must be non-negative", err=True) raise typer.Exit(1) # Get current branch if none specified - if not branch: + branch_name = branch + if not branch_name: result = run_command(["git", "rev-parse", "--abbrev-ref", "HEAD"]) - branch = result.stdout.strip() + branch_name = result.stdout.strip() # Get run IDs with manual pagination using --created filter - typer.echo(f"Fetching workflow runs for branch '{branch}'...") + log(f"Fetching workflow runs for branch '{branch_name}'...") all_runs = [] - seen_ids = set() # Track IDs to handle timestamp collisions + seen_ids: set[int] = set() page_size = 100 # GitHub CLI default page size total_fetched = 0 - created_filter = None # Start with no filter + created_filter: str | None = None while total_fetched < limit: # Calculate how many to fetch in this batch to_fetch = min(page_size, limit - total_fetched) # Build command with optional created filter for pagination - cmd = [ + cmd: list[str] = [ "gh", "run", "list", "--branch", - branch, + branch_name, "--json", "databaseId,status,createdAt", "--limit", str(to_fetch), ] - if created_filter: + if created_filter is not None: cmd.extend(["--created", created_filter]) # Fetch a batch of runs @@ -129,24 +271,24 @@ def clear_branch_runs( sorted_runs = sorted(deduped_runs.values(), key=lambda x: x["createdAt"]) if not sorted_runs: - typer.echo(f"No GitHub Actions runs found for branch '{branch}'") + log(f"No GitHub Actions runs found for branch '{branch_name}'") return if keep_latest: if keep_latest >= len(sorted_runs): - typer.echo("No runs to delete after applying keep filter") + log("No runs to delete after applying keep filter") return - typer.echo(f"Keeping the most recent {keep_latest} runs") + log(f"Keeping the most recent {keep_latest} runs") runs_to_delete = sorted_runs[:-keep_latest] else: runs_to_delete = sorted_runs run_ids = [str(run["databaseId"]) for run in runs_to_delete] - typer.echo(f"Found {len(run_ids)} runs to delete for branch '{branch}'") + log(f"Found {len(run_ids)} runs to delete for branch '{branch_name}'") if not force and not confirm_action("Do you want to proceed with deletion?"): - typer.echo("Operation cancelled") + log("Operation cancelled") return # Delete runs with progress indication @@ -161,80 +303,286 @@ def clear_branch_runs( successful_deletions += 1 else: failed_deletions.append(run_id) - typer.echo( - f"\nWarning: Failed to delete run {run_id}: {result.stderr}", + log( + f"Failed to delete run {run_id}: {result.stderr}", err=True, ) except Exception as e: failed_deletions.append(run_id) - typer.echo( - f"\nWarning: Error when deleting run {run_id}: {str(e)}", + log( + f"Error deleting run {run_id}: {str(e)}", err=True, ) if failed_deletions: - typer.echo( - f"Successfully deleted {successful_deletions}/{len(run_ids)} runs. {len(failed_deletions)} deletions failed." + log( + f"Deleted {successful_deletions}/{len(run_ids)} runs. {len(failed_deletions)} failed." ) else: - typer.echo(f"Successfully deleted all {successful_deletions} runs.") + log(f"Deleted all {successful_deletions} runs.") except subprocess.CalledProcessError as e: - typer.echo(f"Error: {e.stderr}", err=True) + log(f"Error: {e.stderr}", err=True) raise typer.Exit(1) -@app.command() +@app.command( + "delete-merged", + epilog=( + "Examples:\n\n" + " # Preview merged branches that would be deleted\n" + " git-cleanup delete-merged -d\n\n" + " # Delete merged branches against a specific base\n" + " git-cleanup delete-merged -b develop -f\n\n" + " # Across all repos (dry run)\n" + " ls -d ~/git/*/ | xargs -I{} git-cleanup -C {} delete-merged -d\n\n" + " # Across all repos (force, no prompt)\n" + " ls -d ~/git/*/ | xargs -I{} git-cleanup -C {} delete-merged -f\n\n" + " # Across all repos (interactive prompt)\n" + " ls -d ~/git/*/ | xargs -I{} git-cleanup -C {} delete-merged" + ), +) +def delete_merged_local_branches( + base_branch: str | None = typer.Option( + None, + "--base", + "-b", + help=( + "Branch used to determine merge status " "(defaults to origin HEAD branch)" + ), + ), + force: bool = typer.Option( + False, + "--force", + "-f", + help="Skip confirmation prompt", + ), + dry_run: bool = typer.Option( + False, + "--dry-run", + "-d", + help="Show merged branches that would be deleted without deleting them", + ), +): + """Delete local branches already merged into the base branch""" + try: + base_branch_name = base_branch or get_default_branch() + + # Pre-check: abort if working tree has conflicts or unusual index state + status_check = run_command(["git", "status", "--porcelain"], check=False) + if status_check.returncode == 0: + for line in status_check.stdout.splitlines(): + code = line[:2] + if code in ("AA", "UU", "AU", "UA", "DD", "DU", "UD"): + log( + f"Error: working tree has unresolved conflicts ({code}: {line[3:].strip()})", + err=True, + ) + log( + "Resolve conflicts or run 'git reset' before retrying", err=True + ) + raise typer.Exit(1) + + with preserve_branch(): + current_branch_name = run_command( + ["git", "rev-parse", "--abbrev-ref", "HEAD"] + ).stdout.strip() + + if current_branch_name != base_branch_name: + run_command(["git", "checkout", base_branch_name]) + + run_command(["git", "pull", "--prune"]) + + protected_branches = {"main", "master", base_branch_name} + + # git branch --merged is used only to choose the deletion flag + # (-d safe vs -D force). It is NOT used as a merge signal because + # it lists any branch whose tip is reachable from base, including + # branches that were never actually merged (just created from base + # or pushed without a PR). + git_merged_result = run_command( + [ + "git", + "branch", + "--merged", + base_branch_name, + "--format", + "%(refname:short)", + ] + ) + git_merged = set( + b.strip() for b in git_merged_result.stdout.splitlines() if b.strip() + ) + + confirmed_merged: set[str] = set() + + # Phase 1: Squash/rebase merge detection via git cherry + all_branches_result = run_command( + ["git", "branch", "--format", "%(refname:short)"] + ) + all_local = [ + b.strip() + for b in all_branches_result.stdout.splitlines() + if b.strip() and b.strip() not in protected_branches + ] + + for branch in all_local: + try: + cherry = run_command( + ["git", "cherry", base_branch_name, branch], check=False + ) + if cherry.returncode == 0: + lines = cherry.stdout.splitlines() + unmerged = [line for line in lines if line.startswith("+")] + # Only mark merged if cherry produced output (has - lines). + # Empty output means no unique commits, not "all merged". + if lines and not unmerged: + confirmed_merged.add(branch) + except Exception: + pass + + # Phase 2: GitHub API — the authoritative source for merged PRs. + # Catches ff-merges and squash-merges where cherry fails. + unchecked = [b for b in all_local if b not in confirmed_merged] + if unchecked: + try: + gh_result = run_command( + [ + "gh", + "pr", + "list", + "--state", + "merged", + "--limit", + "200", + "--json", + "headRefName", + "--jq", + ".[].headRefName", + ], + check=False, + ) + if gh_result.returncode == 0: + gh_merged = set( + b.strip() + for b in gh_result.stdout.splitlines() + if b.strip() + ) + for branch in unchecked: + if branch in gh_merged: + confirmed_merged.add(branch) + except Exception: + pass + + branches_to_delete = [ + branch + for branch in confirmed_merged + if branch not in protected_branches + ] + + if not branches_to_delete: + log(f"No merged branches found (base: {base_branch_name})") + return + + log( + f"Found {len(branches_to_delete)} merged " + f"branch(es) (base: {base_branch_name}):" + ) + for branch in branches_to_delete: + log(f" - {branch}") + + if dry_run: + log("Dry run complete") + return + + if not force and not confirm_action( + "Do you want to proceed with deletion?" + ): + log("Operation cancelled") + return + + for branch in branches_to_delete: + flag = "-d" if branch in git_merged else "-D" + run_command(["git", "branch", flag, branch]) + + log(f"Deleted {len(branches_to_delete)} merged branch(es)") + + except subprocess.CalledProcessError as e: + log(f"Error: {e.stderr}", err=True) + raise typer.Exit(1) + + +@app.command( + epilog=( + "Examples:\n\n" + " # Preview which branches would be deleted\n" + " git-cleanup delete-local-branches -d\n\n" + " # Force delete all local branches\n" + " git-cleanup delete-local-branches -f\n\n" + " # Across all repos (force, no prompt)\n" + " ls -d ~/git/*/ | xargs -I{} git-cleanup -C {} delete-local-branches -f\n\n" + " # Across all repos (interactive prompt)\n" + " ls -d ~/git/*/ | xargs -I{} git-cleanup -C {} delete-local-branches" + ), +) def delete_local_branches( force: bool = typer.Option(False, "--force", "-f", help="Skip confirmation prompt"), + dry_run: bool = typer.Option( + False, + "--dry-run", + "-d", + help="Show local branches that would be deleted without deleting them", + ), ): """Delete all local branches except main/master""" try: - # Get default branch - result = run_command(["git", "remote", "show", "origin"]) - default_branch = None - for line in result.stdout.splitlines(): - if "HEAD branch" in line: - default_branch = line.split()[-1] - break + default_branch = get_default_branch() + + with preserve_branch(): + # Switch to default branch and update + run_command(["git", "checkout", default_branch]) + run_command(["git", "pull", "--prune"]) + + # Get list of branches to delete + result = run_command(["git", "branch"]) + branches = [b.strip() for b in result.stdout.splitlines()] + branches_to_delete = [ + b + for b in branches + if b + not in [ + "master", + "main", + default_branch, + f"* {default_branch}", + ] + ] - if not default_branch: - typer.echo("Error: Could not determine default branch", err=True) - raise typer.Exit(1) + if not branches_to_delete: + log("No branches to delete") + return - # Switch to default branch and update - run_command(["git", "checkout", default_branch]) - run_command(["git", "pull", "--prune"]) - - # Get list of branches to delete - result = run_command(["git", "branch"]) - branches = [b.strip() for b in result.stdout.splitlines()] - branches_to_delete = [ - b - for b in branches - if b not in ["master", "main", default_branch, f"* {default_branch}"] - ] - - if not branches_to_delete: - typer.echo("No branches to delete") - return + log(f"Found {len(branches_to_delete)} branch(es) to delete:") + for branch in branches_to_delete: + log(f" - {branch}") - # Show planned actions - typer.echo(f"Found {len(branches_to_delete)} branches to delete:") - for branch in branches_to_delete: - typer.echo(f" - {branch}") + if dry_run: + log("Dry run complete") + return - if not force and not confirm_action("Do you want to proceed with deletion?"): - typer.echo("Operation cancelled") - return + if not force and not confirm_action( + "Do you want to proceed with deletion?" + ): + log("Operation cancelled") + return - # Delete branches - for branch in branches_to_delete: - run_command(["git", "branch", "-D", branch.strip()]) - typer.echo(f"Successfully deleted {len(branches_to_delete)} branches") + # Delete branches + for branch in branches_to_delete: + run_command(["git", "branch", "-D", branch.strip()]) + log(f"Deleted {len(branches_to_delete)} branch(es)") except subprocess.CalledProcessError as e: - typer.echo(f"Error: {e.stderr}", err=True) + log(f"Error: {e.stderr}", err=True) raise typer.Exit(1) diff --git a/sort-yaml-key/sort-yaml-key b/sort-yaml-key/sort-yaml-key index 135ae3d..0f3386a 100755 --- a/sort-yaml-key/sort-yaml-key +++ b/sort-yaml-key/sort-yaml-key @@ -2,10 +2,26 @@ import argparse import sys +from collections import OrderedDict import yaml +def represent_ordereddict(dumper, data): + value = [] + + for item_key, item_value in data.items(): + node_key = dumper.represent_data(item_key) + node_value = dumper.represent_data(item_value) + + value.append((node_key, node_value)) + + return yaml.nodes.MappingNode("tag:yaml.org,2002:map", value) + + +yaml.add_representer(OrderedDict, represent_ordereddict, Dumper=yaml.SafeDumper) + + def main(): parser = argparse.ArgumentParser( description="Sort a YAML file by a specified key.", @@ -27,29 +43,49 @@ def main(): try: with open(args.filename, "r") as file: - data = yaml.safe_load(file) + loaded = yaml.safe_load(file) except FileNotFoundError: print(f"Error: file not found: {args.filename}", file=sys.stderr) sys.exit(1) - - if not isinstance(data, list): - print( - f"Error: expected a YAML list, got {type(data).__name__}", file=sys.stderr - ) + except yaml.YAMLError as e: + print(f"Error: invalid YAML in {args.filename}: {e}", file=sys.stderr) sys.exit(1) + # Extract the list to sort - handle both root list and dict with list values + if isinstance(loaded, dict): + data = None + for value in loaded.values(): + if isinstance(value, list): + data = value + break + if data is None: + print("Error: no list found in YAML structure", file=sys.stderr) + sys.exit(1) + else: + if not isinstance(loaded, list): + print( + f"Error: expected a YAML list, got {type(loaded).__name__}", + file=sys.stderr, + ) + sys.exit(1) + data = loaded + try: data.sort(key=lambda x: x[args.key]) except KeyError: print(f"Error: key '{args.key}' not found in all items", file=sys.stderr) sys.exit(1) - # False sorts before True, so != pushes the sort key to index 0 in each mapping - data = [ - dict(sorted(item.items(), key=lambda kv: kv[0] != args.key)) for item in data - ] + # Reorder each block so that the specified key is first + for i in range(len(data)): + data[i] = OrderedDict( + sorted(data[i].items(), key=lambda item: item[0] != args.key) + ) - print(yaml.safe_dump(data, default_flow_style=False, sort_keys=False)) + sorted_yaml = yaml.dump( + loaded, Dumper=yaml.SafeDumper, default_flow_style=False, sort_keys=False + ) + print(sorted_yaml) if __name__ == "__main__": diff --git a/testpod/testpod b/testpod/testpod index 6ca6122..0492fa3 100755 --- a/testpod/testpod +++ b/testpod/testpod @@ -2,6 +2,7 @@ import argparse import subprocess +import sys # Create the argument parser parser = argparse.ArgumentParser( @@ -38,7 +39,7 @@ if args.delete: try: subprocess.run(kubectl_command, check=True) except subprocess.CalledProcessError as e: - print(f"Error deleting pod: {e}") + print(f"Error deleting pod: {e}", file=sys.stderr) else: kubectl_command.extend(["apply", "-f", "-"]) pod_manifest = f""" @@ -57,6 +58,12 @@ spec: try: process = subprocess.Popen(kubectl_command, stdin=subprocess.PIPE) process.communicate(input=pod_manifest.encode()) + if process.returncode != 0: + print( + f"Error deploying pod: kubectl exited with code {process.returncode}", + file=sys.stderr, + ) + sys.exit(1) if args.shell: shell_command = ["kubectl"] if args.context: @@ -65,5 +72,5 @@ spec: shell_command.extend(["--namespace", args.namespace]) shell_command.extend(["exec", "-it", args.name, "--", "bash"]) subprocess.run(shell_command) - except subprocess.CalledProcessError as e: - print(f"Error deploying pod: {e}") + except Exception as e: + print(f"Error deploying pod: {e}", file=sys.stderr) diff --git a/tfplan-all/tfplan-all b/tfplan-all/tfplan-all index 58857a7..f48b01e 100755 --- a/tfplan-all/tfplan-all +++ b/tfplan-all/tfplan-all @@ -15,6 +15,7 @@ Plans are saved as path-with-dashes.tfplan.txt in the output directory. """ import argparse +import fnmatch import os import re import shutil @@ -217,7 +218,10 @@ def main() -> None: output_dir = Path(args.output_dir).resolve() if args.output_dir else target_dir if not target_dir.exists(): - print(f"{RED}Error: Target directory does not exist: {target_dir}{RESET}") + print( + f"{RED}Error: Target directory does not exist: {target_dir}{RESET}", + file=sys.stderr, + ) sys.exit(1) # Ensure output directory exists @@ -236,12 +240,12 @@ def main() -> None: if args.filter_pattern: pattern = args.filter_pattern # Convert glob-style wildcards to regex - if "*" in pattern and not pattern.startswith("^"): - pattern = pattern.replace("*", ".*") + if any(c in pattern for c in "*?[") and not pattern.startswith("^"): + pattern = fnmatch.translate(pattern) try: regex = re.compile(pattern) except re.error as e: - print(f"{RED}Error: Invalid filter pattern: {e}{RESET}") + print(f"{RED}Error: Invalid filter pattern: {e}{RESET}", file=sys.stderr) sys.exit(1) tf_dirs = [d for d in tf_dirs if regex.search(str(d.relative_to(target_dir)))] @@ -265,7 +269,7 @@ def main() -> None: sys.exit(0) if not shutil.which("terraform"): - print(f"{RED}Error: terraform not found in PATH{RESET}") + print(f"{RED}Error: terraform not found in PATH{RESET}", file=sys.stderr) sys.exit(1) if not args.quiet: