Interactively apply Ruff autofixes per rule and optionally commit the changes.

autofix.py

#!/usr/bin/env python3

# Teal Dulcet
# Interactively apply Ruff autofixes per rule and optionally commit the changes
# Also see: https://github.com/astral-sh/ruff/issues/4361

# Run: python3 autofix.py [Ruff args]...

import json
import os
import re
import subprocess
import sys
from urllib.parse import urlparse

# Additional arguments to the first run of 'ruff check', such as --select or --ignore options
args = []

# Additional arguments to every run of 'ruff check'
ARGS = [
    "--preview",
    "--line-length",
    "320", # https://github.com/astral-sh/ruff/issues/8106
    "--unsafe-fixes",
    ".",
]

# Automatically commit any changes made
COMMIT = False

BOLD = "\033[1m"
RESET = "\033[22m"

# yes_regex = re.compile(locale.nl_langinfo(locale.YESEXPR))
yes_regex = re.compile(r"^[yY]")
# no_regex = re.compile(locale.nl_langinfo(locale.NOEXPR))
no_regex = re.compile(r"^[nN]")


def ask_yn(astr, val):
    """Prompt the user with a yes/no question and return their response as a boolean value."""
    while True:
        temp = input("{} ({}): ".format(astr, "Y" if val else "N")).strip()
        if not temp:
            return val
        yes_res = yes_regex.match(temp)
        no_res = no_regex.match(temp)
        if yes_res or no_res:
            return bool(yes_res)


def ask_ok():
    """Displays a message and waits for the user to acknowledge it."""
    input("\nHit enter to continue: ")


args += sys.argv[1:]

if COMMIT and subprocess.call(["git", "diff", "--quiet"]):
    print("Please commit or stash your changes before running this script.")
    sys.exit(1)

# print(f"\n{BOLD}Clearing cache{RESET}\n")
# subprocess.check_call(["ruff", "clean"])

print(f"\n{BOLD}Determining errors{RESET}\n")
with subprocess.Popen(
    ["ruff", "check", "--output-format", "json", *args, *ARGS], stdout=subprocess.PIPE
) as p:
    output, _ = p.communicate()
errors = json.loads(output)

if not errors:
    print("\nNo errors found!")
    sys.exit(0)

rules = {}
for error in errors:
    code = error["code"]
    if code not in rules:
        rules[code] = []
    rules[code].append(error)

for i, (code, rule) in enumerate(rules.items()):
    print(f"\n{BOLD}{i + 1} of {len(rules)}: Rule {code}{RESET}\n")
    subprocess.check_call(["ruff", "rule", code])

    print(f"\n{BOLD}Errors{RESET}\n")
    with subprocess.Popen(["ruff", "check", "--select", code, *ARGS]) as p:
        p.wait()
        if not p.poll():
            print("\nNo errors found")
            continue

    print()
    if any(error["fix"] for error in rule):
        with subprocess.Popen(
            ["ruff", "check", "--diff", "--select", code, *ARGS]
        ) as p:
            p.wait()

        print()
        if ask_yn("Apply the autofixes for the above errors?", True):
            with subprocess.Popen(
                ["ruff", "check", "--select", code, *ARGS, "--fix"]
            ) as p:
                p.wait()
                if p.poll():
                    print(
                        "\nSome errors for this rule were not autofixable, please manually fix the above errors"
                    )
    else:
        print("This rule is not autofixable, please manually fix the above errors")

    url = urlparse(rule[0]["url"])
    name = os.path.basename(url.path)
    messages = {error["message"] for error in rule}
    message = "Fixed {} ({}){}".format(
        code, name, f": {messages.pop()}" if len(messages) == 1 else ""
    )
    print(f"\nCommit message: {message!r}")

    ask_ok()

    if COMMIT:
        if subprocess.call(["git", "diff", "--quiet"]):
            subprocess.check_call(["git", "commit", "-am", message])
        else:
            print("No changes were made")