From 5cdbde0713d9239c08dd742106cfa731a9d9415a Mon Sep 17 00:00:00 2001 From: Shawn Koonin Date: Wed, 18 Mar 2026 18:25:28 -0700 Subject: [PATCH 1/5] Sync tool fixes from sk-tools Updated: asm, ec2-search, ec2-state, gh-search, git-audit, git-cleanup, sort-yaml-key, testpod, tfplan-all, README. Key fixes: ec2-state shell injection + ThreadPoolExecutor cleanup, git-cleanup squash merge detection, sort-yaml-key dict data loss, stderr for errors, type hint modernization. Reverted: github-delete-pr-comments (different codebase, not a direct sync). --- README.md | 20 +- asm/asm | 2 +- ec2-search/ec2-search | 265 +++++++------------ ec2-state/ec2-state | 68 ++--- gh-search/gh-search | 351 +++++++++++++++----------- git-audit/git-audit | 31 ++- git-cleanup/git-cleanup | 490 ++++++++++++++++++++++++++++++------ sort-yaml-key/sort-yaml-key | 117 +++++---- testpod/testpod | 5 +- tfplan-all/tfplan-all | 26 +- 10 files changed, 854 insertions(+), 521 deletions(-) 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/asm/asm b/asm/asm index 6a9b489..96c1593 100755 --- a/asm/asm +++ b/asm/asm @@ -163,7 +163,7 @@ Examples: asm update my-secret -v newkey=newval Update secret asm update my-secret --force Force update unchanged secret asm delete my-secret Delete immediately - asm delete my-secret -R 7 Delete with 7-day recovery + asm delete my-secret -R 7 Delete with 7-day recovery asm search List all secrets asm list List all secrets (alias) asm search "prod.*database" Search by regex pattern diff --git a/ec2-search/ec2-search b/ec2-search/ec2-search index b038236..8b03545 100755 --- a/ec2-search/ec2-search +++ b/ec2-search/ec2-search @@ -1,54 +1,32 @@ #!/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, + ["aws", "ec2", "describe-regions", "--profile", profile] ) 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", @@ -57,13 +35,12 @@ def get_instances(region: str, profile: str) -> list[dict[str, Any]]: region, "--profile", profile, - ], - timeout=SUBPROCESS_TIMEOUT, + ] ) - 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( [ @@ -78,99 +55,48 @@ def get_instances(region: str, profile: str) -> list[dict[str, Any]]: f"Name=availability-zone,Values={zone}", "--query", "Reservations[].Instances[]", - ], - timeout=SUBPROCESS_TIMEOUT, + ] ) - 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. - Args: - profile: AWS profile name - region: Optional specific region, or None for all regions +def search_instances_by_profile(profile, args): + regions = [args.region] if args.region else get_regions(profile) - 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 - } + futures = [ + executor.submit(get_instances_by_region, region, profile) + for region in regions + ] for future in concurrent.futures.as_completed(futures): - region_name = futures[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"An error occurred: {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=[]): + """ + 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 +104,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,116 +118,122 @@ 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", "") + name_tag = next( + ( + tag["Value"] + for tag in instance.get("Tags", []) + if tag["Key"] == "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: - ec2-search -p primary,secondary - ec2-search -r us-west-2 -p primary,secondary - ec2-search -p primary,secondary -c ec2-instances.csv - ec2-search -p primary,secondary --all-tags - ec2-search -p primary,secondary --keys Environment,Project""", + epilog="""Example usage: + ./get_ec2_instances.py -p primary,secondary + ./get_ec2_instances.py -r us-west-2 -p primary,secondary + ./get_ec2_instances.py -p primary,secondary -c ec2-instances.csv + ./get_ec2_instances.py -p primary,secondary --all-tags + ./get_ec2_instances.py -p primary,secondary --keys Environment,Project""", ) 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: + futures = [ + executor.submit(search_instances_by_profile, profile, args) + for profile in profiles + ] for future in concurrent.futures.as_completed(futures): - profile_name = futures[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 + print(f"An error occurred: {e}", file=sys.stderr) - # 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..98502bb 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 = [] @@ -148,23 +145,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 +165,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..1a5ca51 100755 --- a/gh-search/gh-search +++ b/gh-search/gh-search @@ -1,142 +1,169 @@ #!/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_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 + +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 -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_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 - Returns: - GitHub access token - Raises: - FileNotFoundError: If token file doesn't exist - ValueError: If token file is empty - """ +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 path.is_file(): - raise FileNotFoundError(f"Access token file not found: {path}") + try: + if not token_file.is_file(): + raise Exception("No token found in any location") + + with token_file.open() as f: + return f.read().strip() + except Exception: + raise + + +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) + + if response.status_code != 200: + print(f"Error: {response.status_code}", file=sys.stderr) + return [] - content = path.read_text().strip() - if not content: - raise ValueError(f"Access token file is empty: {path}") + data = response.json() + repos = [] - return content + for repo in data.get("items", []): + repos.append( + { + "name": repo["full_name"], + "description": repo["description"], + "url": repo["html_url"], + "stars": repo["stargazers_count"], + } + ) + 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) +def search_repos( + 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 "" + url = f"{github_api_base_url}/search/code?q={org_query}+{query}" + response = requests.get(url, headers=headers) + + 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]] = [] + cloned_repos = [] + 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) - continue + subprocess.run( + ["git", "clone", "-q", clone_url, str(repo_path)], check=True + ) + cloned_repos.append(str(repo_path)) else: print(f"Repo {repo_name} already exists. Skipping clone.") @@ -145,33 +172,20 @@ 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): + args = ["code"] if new_instance: - args = ["code", "-n"] + paths + args.extend(["-n"] + [str(repo_path) for _, _, repo_path in repos]) + print(f"Opened {len(repos)} repositories in a new instance of VS Code.\n") else: - args = ["code", "--add"] + paths + args.extend(["--add"] + [str(repo_path) for _, _, repo_path in repos]) + print(f"Opened {len(repos)} repositories in a new instance of VS Code.\n") - 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) - except FileNotFoundError: - print("VS Code 'code' command not found in PATH", file=sys.stderr) + subprocess.run(args) -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 +195,76 @@ 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="The organization to search in. If not specified, search all of GitHub.", + default=None, + ) + parser.add_argument( + "-r", "--repo", help="Search for repositories by name", 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)" + "-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( - "-u", "--url", default=DEFAULT_GITHUB_API_BASE_URL, help="GitHub API base URL" + "-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 - - # Search + access_token = get_access_token(token=args.token, token_file=args.token_file) repos = search_repos( query=args.query, access_token=access_token, @@ -234,27 +272,38 @@ def main() -> int: github_api_base_url=args.url, ) - if not repos: - print("No repositories found.") - return 0 + 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 - # 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 not args.clone: + parser.error("The --edit flag requires the --clone flag to be specified.") + 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..a853360 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,106 @@ 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) + + +@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 +195,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 +270,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 +302,285 @@ 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: + log(f"Error: {e.stderr}", err=True) + raise typer.Exit(1) + + +@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} + + # Track which phase detected each branch: -d is safe, -D for squash/rebase + safe_merged: set[str] = set() # Phase 1: git sees as merged, -d works + force_merged: set[str] = set() # Phase 2/3: squash/rebase, needs -D + + # Phase 1: Traditional merge detection (fast, catches ff/merge commits) + result = run_command( + [ + "git", + "branch", + "--merged", + base_branch_name, + "--format", + "%(refname:short)", + ] + ) + safe_merged = set( + b.strip() for b in result.stdout.splitlines() if b.strip() + ) + + # Phase 2: Squash/rebase merge detection via git cherry + all_branches_result = run_command( + ["git", "branch", "--format", "%(refname:short)"] + ) + remaining = [ + b.strip() + for b in all_branches_result.stdout.splitlines() + if b.strip() + and b.strip() not in safe_merged + and b.strip() not in protected_branches + ] + + for branch in remaining: + try: + cherry = run_command( + ["git", "cherry", base_branch_name, branch], check=False + ) + if cherry.returncode == 0: + unmerged = [ + line + for line in cherry.stdout.splitlines() + if line.startswith("+") + ] + if not unmerged: + force_merged.add(branch) + except Exception: + pass + + # Phase 3: GitHub API merge detection (catches squash merges where cherry fails) + still_remaining = [b for b in remaining if b not in force_merged] + if still_remaining: + 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 still_remaining: + if branch in gh_merged: + force_merged.add(branch) + except Exception: + pass + + all_merged = safe_merged | force_merged + branches_to_delete = [ + branch for branch in all_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 safe_merged else "-D" + run_command(["git", "branch", flag, branch]) + + log(f"Deleted {len(branches_to_delete)} merged 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) -@app.command() +@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..76d851a 100755 --- a/sort-yaml-key/sort-yaml-key +++ b/sort-yaml-key/sort-yaml-key @@ -1,56 +1,75 @@ #!/usr/bin/env python3 import argparse -import sys +from collections import OrderedDict import yaml +# Create the parser +parser = argparse.ArgumentParser( + description="Sort a YAML file by a specified key.", + epilog="Ensure that the key exists in every item of the YAML file.", +) -def main(): - parser = argparse.ArgumentParser( - description="Sort a YAML file by a specified key.", - epilog="Ensure that the key exists in every item of the YAML file.", - ) - parser.add_argument( - "-f", - "--filename", - metavar="filename", - type=str, - required=True, - help="the path to the YAML file", - ) - parser.add_argument( - "-k", "--key", metavar="key", type=str, required=True, help="the key to sort by" - ) - - args = parser.parse_args() - - try: - with open(args.filename, "r") as file: - data = 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 - ) - sys.exit(1) - - 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 - ] - - print(yaml.safe_dump(data, default_flow_style=False, sort_keys=False)) - - -if __name__ == "__main__": - main() +# Add the arguments +parser.add_argument( + "-f", + "--filename", + metavar="filename", + type=str, + required=True, + help="the path to the YAML file", +) +parser.add_argument( + "-k", "--key", metavar="key", type=str, required=True, help="the key to sort by" +) + +# Parse the arguments +args = parser.parse_args() + +# Load the YAML data from the file +with open(args.filename, "r") as file: + loaded = yaml.safe_load(file) + +# Extract the list to sort - handle both root list and dict with list values +if isinstance(loaded, dict): + # If root is a dict, find the first list value + data = None + for value in loaded.values(): + if isinstance(value, list): + data = value + break + if data is None: + raise ValueError("No list found in YAML structure") +else: + data = loaded + +# Sort the data by the specified key +data.sort(key=lambda x: x[args.key]) + +# 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)) + +# Create a custom representer for OrderedDict + + +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) + + +# Add the custom representer to the Dumper class +yaml.add_representer(OrderedDict, represent_ordereddict) + +# Dump the sorted data +sorted_yaml = yaml.dump(loaded, default_flow_style=False) + +print(sorted_yaml) diff --git a/testpod/testpod b/testpod/testpod index 6ca6122..41958b8 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""" @@ -66,4 +67,4 @@ spec: shell_command.extend(["exec", "-it", args.name, "--", "bash"]) subprocess.run(shell_command) except subprocess.CalledProcessError as e: - print(f"Error deploying pod: {e}") + print(f"Error deploying pod: {e}", file=sys.stderr) diff --git a/tfplan-all/tfplan-all b/tfplan-all/tfplan-all index 58857a7..13539e9 100755 --- a/tfplan-all/tfplan-all +++ b/tfplan-all/tfplan-all @@ -1,20 +1,21 @@ #!/usr/bin/env python3 """ -tfplan-all - Run terraform init+plan on all subdirectories containing .tf files. +tfplan_all - Run terraform init+plan on all subdirectories containing .tf files. Usage: - tfplan-all [target_dir] [filter_pattern] [options] + tfplan_all [target_dir] [filter_pattern] [options] Examples: - tfplan-all # All terraform dirs in current directory - tfplan-all /path/to/infra # All terraform dirs in /path/to/infra - tfplan-all . '*network*' # Only dirs matching *network* - tfplan-all . 'prod' # Only dirs containing 'prod' + tfplan_all # All terraform dirs in current directory + tfplan_all /path/to/infra # All terraform dirs in /path/to/infra + tfplan_all . '*network*' # Only dirs matching *network* + tfplan_all . 'prod' # Only dirs containing 'prod' 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: From 3140648c7a3ffb6a0258734acbce0c6b270ffbfa Mon Sep 17 00:00:00 2001 From: Shawn Koonin Date: Mon, 23 Mar 2026 15:08:22 -0700 Subject: [PATCH 2/5] Fix PR review issues: error handling, timeouts, URL encoding sort-yaml-key: add sort_keys=False, error handling, main guard gh-search: URL-encode search query, add timeouts, fix clone/vscode error handling ec2-search: fix mutable default arg, epilog script name, add subprocess timeouts --- ec2-search/ec2-search | 23 ++++--- gh-search/gh-search | 55 +++++++++++------ sort-yaml-key/sort-yaml-key | 119 ++++++++++++++++++++---------------- 3 files changed, 118 insertions(+), 79 deletions(-) diff --git a/ec2-search/ec2-search b/ec2-search/ec2-search index 8b03545..2a6d9f7 100755 --- a/ec2-search/ec2-search +++ b/ec2-search/ec2-search @@ -14,7 +14,8 @@ 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] + ["aws", "ec2", "describe-regions", "--profile", profile], + timeout=60, ) regions = [region["RegionName"] for region in json.loads(output)["Regions"]] return regions @@ -35,7 +36,8 @@ def get_instances(region, profile): region, "--profile", profile, - ] + ], + timeout=60, ) zones = [zone["ZoneName"] for zone in json.loads(zones)["AvailabilityZones"]] @@ -55,7 +57,8 @@ def get_instances(region, profile): f"Name=availability-zone,Values={zone}", "--query", "Reservations[].Instances[]", - ] + ], + timeout=120, ) instances += json.loads(output) @@ -90,10 +93,12 @@ def search_instances_by_profile(profile, args): return instances -def generate_data(instance_tuples, show_all_tags=False, keys=[]): +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"] tag_keys = set() for instance, profile in instance_tuples: @@ -165,11 +170,11 @@ def main(): description="Get information about EC2 instances.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog="""Example usage: - ./get_ec2_instances.py -p primary,secondary - ./get_ec2_instances.py -r us-west-2 -p primary,secondary - ./get_ec2_instances.py -p primary,secondary -c ec2-instances.csv - ./get_ec2_instances.py -p primary,secondary --all-tags - ./get_ec2_instances.py -p primary,secondary --keys Environment,Project""", + ec2-search -p primary,secondary + ec2-search -r us-west-2 -p primary,secondary + ec2-search -p primary,secondary -c ec2-instances.csv + ec2-search -p primary,secondary --all-tags + ec2-search -p primary,secondary --keys Environment,Project""", ) parser.add_argument( "-r", diff --git a/gh-search/gh-search b/gh-search/gh-search index 1a5ca51..756f1a5 100755 --- a/gh-search/gh-search +++ b/gh-search/gh-search @@ -75,14 +75,20 @@ def get_access_token(token=None, token_file=None): else: token_file = Path(os.path.expanduser("~/.github_token")) - try: - if not token_file.is_file(): - raise Exception("No token found in any location") + 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) - with token_file.open() as f: - return f.read().strip() - except Exception: - raise + token = token_file.open().read().strip() + if not token: + print(f"Error: token file '{token_file}' is empty.", file=sys.stderr) + sys.exit(1) + + return token def search_repositories( @@ -96,7 +102,7 @@ def search_repositories( 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) + response = requests.get(url, headers=headers, timeout=30) if response.status_code != 200: print(f"Error: {response.status_code}", file=sys.stderr) @@ -126,8 +132,9 @@ def search_repos( ): headers = {"Authorization": f"token {access_token}"} org_query = f"org:{organization}" if organization else "" - url = f"{github_api_base_url}/search/code?q={org_query}+{query}" - response = requests.get(url, headers=headers) + 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) @@ -160,9 +167,16 @@ def clone_repos(repos, clone_directory): repo_path = clone_directory / Path(*repo_name_parts[1:]) if not repo_path.exists(): print(f"Cloning {repo_name}...") - subprocess.run( - ["git", "clone", "-q", clone_url, str(repo_path)], check=True + 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 cloned_repos.append(str(repo_path)) else: print(f"Repo {repo_name} already exists. Skipping clone.") @@ -173,16 +187,21 @@ def clone_repos(repos, clone_directory): def open_in_vscode(repos, new_instance=False): - args = ["code"] + cmd = ["code"] if new_instance: - args.extend(["-n"] + [str(repo_path) for _, _, repo_path in repos]) - print(f"Opened {len(repos)} repositories in a new instance of VS Code.\n") + cmd.extend(["-n"] + [str(repo_path) for _, _, repo_path in repos]) else: - args.extend(["--add"] + [str(repo_path) for _, _, repo_path in repos]) - print(f"Opened {len(repos)} repositories in a new instance of VS Code.\n") + cmd.extend(["--add"] + [str(repo_path) for _, _, repo_path in repos]) - subprocess.run(args) + try: + 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("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(): diff --git a/sort-yaml-key/sort-yaml-key b/sort-yaml-key/sort-yaml-key index 76d851a..5dcbbcc 100755 --- a/sort-yaml-key/sort-yaml-key +++ b/sort-yaml-key/sort-yaml-key @@ -1,58 +1,11 @@ #!/usr/bin/env python3 import argparse +import sys from collections import OrderedDict import yaml -# Create the parser -parser = argparse.ArgumentParser( - description="Sort a YAML file by a specified key.", - epilog="Ensure that the key exists in every item of the YAML file.", -) - -# Add the arguments -parser.add_argument( - "-f", - "--filename", - metavar="filename", - type=str, - required=True, - help="the path to the YAML file", -) -parser.add_argument( - "-k", "--key", metavar="key", type=str, required=True, help="the key to sort by" -) - -# Parse the arguments -args = parser.parse_args() - -# Load the YAML data from the file -with open(args.filename, "r") as file: - loaded = yaml.safe_load(file) - -# Extract the list to sort - handle both root list and dict with list values -if isinstance(loaded, dict): - # If root is a dict, find the first list value - data = None - for value in loaded.values(): - if isinstance(value, list): - data = value - break - if data is None: - raise ValueError("No list found in YAML structure") -else: - data = loaded - -# Sort the data by the specified key -data.sort(key=lambda x: x[args.key]) - -# 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)) - -# Create a custom representer for OrderedDict - def represent_ordereddict(dumper, data): value = [] @@ -66,10 +19,72 @@ def represent_ordereddict(dumper, data): return yaml.nodes.MappingNode("tag:yaml.org,2002:map", value) -# Add the custom representer to the Dumper class yaml.add_representer(OrderedDict, represent_ordereddict) -# Dump the sorted data -sorted_yaml = yaml.dump(loaded, default_flow_style=False) -print(sorted_yaml) +def main(): + parser = argparse.ArgumentParser( + description="Sort a YAML file by a specified key.", + epilog="Ensure that the key exists in every item of the YAML file.", + ) + parser.add_argument( + "-f", + "--filename", + metavar="filename", + type=str, + required=True, + help="the path to the YAML file", + ) + parser.add_argument( + "-k", "--key", metavar="key", type=str, required=True, help="the key to sort by" + ) + + args = parser.parse_args() + + try: + with open(args.filename, "r") as file: + loaded = yaml.safe_load(file) + except FileNotFoundError: + print(f"Error: file not found: {args.filename}", file=sys.stderr) + sys.exit(1) + 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) + + # 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) + ) + + sorted_yaml = yaml.dump(loaded, default_flow_style=False, sort_keys=False) + print(sorted_yaml) + + +if __name__ == "__main__": + main() From 96664b7f0d125511351379f78e1098f9c419e0d0 Mon Sep 17 00:00:00 2001 From: Shawn Koonin Date: Tue, 31 Mar 2026 11:02:13 -0700 Subject: [PATCH 3/5] Fix delete-merged false positives on branches with no commits Branches created from base with no work were flagged as merged by git branch --merged and git cherry (empty output). Partition --merged candidates by remote tracking ref: branches with upstream go to safe_merged, branches without require GH API PR verification. --- git-cleanup/git-cleanup | 55 ++++++++++++++++++++++++++++++++--------- 1 file changed, 43 insertions(+), 12 deletions(-) diff --git a/git-cleanup/git-cleanup b/git-cleanup/git-cleanup index a853360..b02b332 100755 --- a/git-cleanup/git-cleanup +++ b/git-cleanup/git-cleanup @@ -409,9 +409,35 @@ def delete_merged_local_branches( "%(refname:short)", ] ) - safe_merged = set( - b.strip() for b in result.stdout.splitlines() if b.strip() + git_merged = set(b.strip() for b in result.stdout.splitlines() if b.strip()) + + # Partition --merged candidates: branches with a remote tracking ref + # are confirmed (were pushed, so had real work). Branches without a + # remote ref may be newly created and never pushed — require GH API + # verification in Phase 3 before deletion. + upstream_result = run_command( + [ + "git", + "for-each-ref", + "--format", + "%(refname:short) %(upstream)", + "refs/heads/", + ] ) + has_upstream = set() + for line in upstream_result.stdout.splitlines(): + parts = line.strip().split(None, 1) + if len(parts) == 2 and parts[1]: + has_upstream.add(parts[0]) + + needs_gh_verification: list[str] = [] + for branch in git_merged: + if branch in protected_branches: + continue + if branch in has_upstream: + safe_merged.add(branch) + else: + needs_gh_verification.append(branch) # Phase 2: Squash/rebase merge detection via git cherry all_branches_result = run_command( @@ -423,6 +449,7 @@ def delete_merged_local_branches( if b.strip() and b.strip() not in safe_merged and b.strip() not in protected_branches + and b.strip() not in git_merged ] for branch in remaining: @@ -431,19 +458,23 @@ def delete_merged_local_branches( ["git", "cherry", base_branch_name, branch], check=False ) if cherry.returncode == 0: - unmerged = [ - line - for line in cherry.stdout.splitlines() - if line.startswith("+") - ] - if not unmerged: + 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: force_merged.add(branch) except Exception: pass - # Phase 3: GitHub API merge detection (catches squash merges where cherry fails) - still_remaining = [b for b in remaining if b not in force_merged] - if still_remaining: + # Phase 3: GitHub API merge detection + # Catches squash merges where cherry fails, and verifies --merged + # branches that lack a remote tracking ref (may be ff-merged or + # newly created with no work). + gh_check_branches = [ + b for b in remaining if b not in force_merged + ] + needs_gh_verification + if gh_check_branches: try: gh_result = run_command( [ @@ -467,7 +498,7 @@ def delete_merged_local_branches( for b in gh_result.stdout.splitlines() if b.strip() ) - for branch in still_remaining: + for branch in gh_check_branches: if branch in gh_merged: force_merged.add(branch) except Exception: From 7355c89c1c1ebe40b06df3fff2c597c3b1f14f8f Mon Sep 17 00:00:00 2001 From: Shawn Koonin Date: Tue, 31 Mar 2026 12:08:33 -0700 Subject: [PATCH 4/5] Use positive merge confirmation instead of git branch --merged inference git branch --merged also falsely flags pushed branches without PRs. Demote it to deletion-flag selection only. Use git cherry (with output guard) and GH API merged PRs as the sole sources of merge evidence. --- git-cleanup/git-cleanup | 82 +++++++++++++---------------------------- 1 file changed, 26 insertions(+), 56 deletions(-) diff --git a/git-cleanup/git-cleanup b/git-cleanup/git-cleanup index b02b332..c26a32e 100755 --- a/git-cleanup/git-cleanup +++ b/git-cleanup/git-cleanup @@ -394,12 +394,12 @@ def delete_merged_local_branches( protected_branches = {"main", "master", base_branch_name} - # Track which phase detected each branch: -d is safe, -D for squash/rebase - safe_merged: set[str] = set() # Phase 1: git sees as merged, -d works - force_merged: set[str] = set() # Phase 2/3: squash/rebase, needs -D - - # Phase 1: Traditional merge detection (fast, catches ff/merge commits) - result = run_command( + # 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", @@ -409,50 +409,23 @@ def delete_merged_local_branches( "%(refname:short)", ] ) - git_merged = set(b.strip() for b in result.stdout.splitlines() if b.strip()) - - # Partition --merged candidates: branches with a remote tracking ref - # are confirmed (were pushed, so had real work). Branches without a - # remote ref may be newly created and never pushed — require GH API - # verification in Phase 3 before deletion. - upstream_result = run_command( - [ - "git", - "for-each-ref", - "--format", - "%(refname:short) %(upstream)", - "refs/heads/", - ] + git_merged = set( + b.strip() for b in git_merged_result.stdout.splitlines() if b.strip() ) - has_upstream = set() - for line in upstream_result.stdout.splitlines(): - parts = line.strip().split(None, 1) - if len(parts) == 2 and parts[1]: - has_upstream.add(parts[0]) - - needs_gh_verification: list[str] = [] - for branch in git_merged: - if branch in protected_branches: - continue - if branch in has_upstream: - safe_merged.add(branch) - else: - needs_gh_verification.append(branch) - - # Phase 2: Squash/rebase merge detection via git cherry + + confirmed_merged: set[str] = set() + + # Phase 1: Squash/rebase merge detection via git cherry all_branches_result = run_command( ["git", "branch", "--format", "%(refname:short)"] ) - remaining = [ + all_local = [ b.strip() for b in all_branches_result.stdout.splitlines() - if b.strip() - and b.strip() not in safe_merged - and b.strip() not in protected_branches - and b.strip() not in git_merged + if b.strip() and b.strip() not in protected_branches ] - for branch in remaining: + for branch in all_local: try: cherry = run_command( ["git", "cherry", base_branch_name, branch], check=False @@ -463,18 +436,14 @@ def delete_merged_local_branches( # Only mark merged if cherry produced output (has - lines). # Empty output means no unique commits, not "all merged". if lines and not unmerged: - force_merged.add(branch) + confirmed_merged.add(branch) except Exception: pass - # Phase 3: GitHub API merge detection - # Catches squash merges where cherry fails, and verifies --merged - # branches that lack a remote tracking ref (may be ff-merged or - # newly created with no work). - gh_check_branches = [ - b for b in remaining if b not in force_merged - ] + needs_gh_verification - if gh_check_branches: + # 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( [ @@ -498,15 +467,16 @@ def delete_merged_local_branches( for b in gh_result.stdout.splitlines() if b.strip() ) - for branch in gh_check_branches: + for branch in unchecked: if branch in gh_merged: - force_merged.add(branch) + confirmed_merged.add(branch) except Exception: pass - all_merged = safe_merged | force_merged branches_to_delete = [ - branch for branch in all_merged if branch not in protected_branches + branch + for branch in confirmed_merged + if branch not in protected_branches ] if not branches_to_delete: @@ -531,7 +501,7 @@ def delete_merged_local_branches( return for branch in branches_to_delete: - flag = "-d" if branch in safe_merged else "-D" + 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)") From f353b4ffa4f6fb06ffe0e0b10384f0bcc216786d Mon Sep 17 00:00:00 2001 From: Shawn Koonin Date: Tue, 31 Mar 2026 12:46:50 -0700 Subject: [PATCH 5/5] Address PR review: security, error handling, and code quality fixes SafeDumper for sort-yaml-key, error context in ec2-search futures, wasted API call in gh-search, Popen returncode check in testpod, file handle leak fix, dead code removal, docstring and help text corrections --- asm/asm | 2 +- ec2-search/ec2-search | 31 +++++++++++++------------------ ec2-state/ec2-state | 15 ++++++++------- gh-search/gh-search | 22 +++++++++++----------- git-cleanup/git-cleanup | 3 ++- sort-yaml-key/sort-yaml-key | 6 ++++-- testpod/testpod | 8 +++++++- tfplan-all/tfplan-all | 12 ++++++------ 8 files changed, 52 insertions(+), 47 deletions(-) diff --git a/asm/asm b/asm/asm index 96c1593..6a9b489 100755 --- a/asm/asm +++ b/asm/asm @@ -163,7 +163,7 @@ Examples: asm update my-secret -v newkey=newval Update secret asm update my-secret --force Force update unchanged secret asm delete my-secret Delete immediately - asm delete my-secret -R 7 Delete with 7-day recovery + asm delete my-secret -R 7 Delete with 7-day recovery asm search List all secrets asm list List all secrets (alias) asm search "prod.*database" Search by regex pattern diff --git a/ec2-search/ec2-search b/ec2-search/ec2-search index 2a6d9f7..73f2e42 100755 --- a/ec2-search/ec2-search +++ b/ec2-search/ec2-search @@ -78,17 +78,18 @@ def search_instances_by_profile(profile, args): instances = [] with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: - futures = [ - executor.submit(get_instances_by_region, region, profile) + future_to_region = { + executor.submit(get_instances_by_region, region, profile): region for region in regions - ] + } - for future in concurrent.futures.as_completed(futures): + for future in concurrent.futures.as_completed(future_to_region): + region_name = future_to_region[future] try: region_instances = future.result() instances += [(instance, profile) for instance in region_instances] except Exception as e: - print(f"An error occurred: {e}", file=sys.stderr) + print(f"Error in {profile}/{region_name}: {e}", file=sys.stderr) return instances @@ -127,14 +128,7 @@ def generate_data(instance_tuples, show_all_tags=False, keys=None): if show_all_tags: tag_row = [tag_values.get(header, "") for header in headers[5:]] else: - name_tag = next( - ( - tag["Value"] - for tag in instance.get("Tags", []) - if tag["Key"] == "Name" - ), - "", - ) + name_tag = tag_values.get("Name", "") row.append(name_tag) tag_row = [tag_values.get(header, "") for header in headers[6:]] row += tag_row @@ -214,17 +208,18 @@ def main(): # Run searches for each profile in parallel with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: - futures = [ - executor.submit(search_instances_by_profile, profile, args) + future_to_profile = { + executor.submit(search_instances_by_profile, profile, args): profile for profile in profiles - ] + } - for future in concurrent.futures.as_completed(futures): + 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"An error occurred: {e}", file=sys.stderr) + print(f"Error searching profile '{profile_name}': {e}", file=sys.stderr) # Generate the data headers, rows = generate_data( diff --git a/ec2-state/ec2-state b/ec2-state/ec2-state index 98502bb..c1fdf5b 100755 --- a/ec2-state/ec2-state +++ b/ec2-state/ec2-state @@ -129,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) diff --git a/gh-search/gh-search b/gh-search/gh-search index 756f1a5..209274b 100755 --- a/gh-search/gh-search +++ b/gh-search/gh-search @@ -83,7 +83,7 @@ def get_access_token(token=None, token_file=None): ) sys.exit(1) - token = token_file.open().read().strip() + token = token_file.read_text().strip() if not token: print(f"Error: token file '{token_file}' is empty.", file=sys.stderr) sys.exit(1) @@ -158,7 +158,6 @@ def clone_repos(repos, clone_directory): print(f"Cloning repositories to {clone_directory}...") print("-" * 80) - cloned_repos = [] all_repos = [] for repo_name, repo_url, clone_url in repos: @@ -177,7 +176,6 @@ def clone_repos(repos, clone_directory): if result.stderr: print(result.stderr, file=sys.stderr) continue - cloned_repos.append(str(repo_path)) else: print(f"Repo {repo_name} already exists. Skipping clone.") @@ -283,13 +281,10 @@ def main(): args = parser.parse_args() + if args.edit and not args.clone: + 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) - repos = search_repos( - query=args.query, - access_token=access_token, - organization=args.org, - github_api_base_url=args.url, - ) if args.repo: repos = search_repositories(args.query, access_token, args.org, args.url) @@ -303,14 +298,19 @@ def main(): print(f"\nFound {len(repos)} repositories\n") return + repos = search_repos( + query=args.query, + access_token=access_token, + organization=args.org, + github_api_base_url=args.url, + ) + if args.clone: clone_directory = Path(os.path.expanduser(args.dir)) repos = clone_repos(repos, clone_directory) print("-" * 80) if args.edit: - if not args.clone: - parser.error("The --edit flag requires the --clone flag to be specified.") if args.new: open_in_vscode(repos, new_instance=True) else: diff --git a/git-cleanup/git-cleanup b/git-cleanup/git-cleanup index c26a32e..f654d96 100755 --- a/git-cleanup/git-cleanup +++ b/git-cleanup/git-cleanup @@ -101,7 +101,7 @@ def confirm_action(message: str) -> bool: try: tty = open("/dev/tty") except OSError: - log("No TTY for confirmation — use -f to skip prompts", err=True) + log("No TTY for confirmation -- use -f to skip prompts", err=True) return False try: sys.stderr.write(f"[{_repo_name()}] {message} (y/N): ") @@ -169,6 +169,7 @@ def preserve_branch(): ) 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( diff --git a/sort-yaml-key/sort-yaml-key b/sort-yaml-key/sort-yaml-key index 5dcbbcc..0f3386a 100755 --- a/sort-yaml-key/sort-yaml-key +++ b/sort-yaml-key/sort-yaml-key @@ -19,7 +19,7 @@ def represent_ordereddict(dumper, data): return yaml.nodes.MappingNode("tag:yaml.org,2002:map", value) -yaml.add_representer(OrderedDict, represent_ordereddict) +yaml.add_representer(OrderedDict, represent_ordereddict, Dumper=yaml.SafeDumper) def main(): @@ -82,7 +82,9 @@ def main(): sorted(data[i].items(), key=lambda item: item[0] != args.key) ) - sorted_yaml = yaml.dump(loaded, 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) diff --git a/testpod/testpod b/testpod/testpod index 41958b8..0492fa3 100755 --- a/testpod/testpod +++ b/testpod/testpod @@ -58,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: @@ -66,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: + 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 13539e9..f48b01e 100755 --- a/tfplan-all/tfplan-all +++ b/tfplan-all/tfplan-all @@ -1,15 +1,15 @@ #!/usr/bin/env python3 """ -tfplan_all - Run terraform init+plan on all subdirectories containing .tf files. +tfplan-all - Run terraform init+plan on all subdirectories containing .tf files. Usage: - tfplan_all [target_dir] [filter_pattern] [options] + tfplan-all [target_dir] [filter_pattern] [options] Examples: - tfplan_all # All terraform dirs in current directory - tfplan_all /path/to/infra # All terraform dirs in /path/to/infra - tfplan_all . '*network*' # Only dirs matching *network* - tfplan_all . 'prod' # Only dirs containing 'prod' + tfplan-all # All terraform dirs in current directory + tfplan-all /path/to/infra # All terraform dirs in /path/to/infra + tfplan-all . '*network*' # Only dirs matching *network* + tfplan-all . 'prod' # Only dirs containing 'prod' Plans are saved as path-with-dashes.tfplan.txt in the output directory. """