ruff/python/ruff-ecosystem/ruff_ecosystem/cli.py

import argparse
import asyncio
import logging
import os
import shutil
import sys
import sysconfig
import tempfile
from contextlib import nullcontext
from pathlib import Path
from signal import SIGINT, SIGTERM

from ruff_ecosystem import logger
from ruff_ecosystem.defaults import DEFAULT_TARGETS
from ruff_ecosystem.format import FormatComparison
from ruff_ecosystem.main import OutputFormat, main
from ruff_ecosystem.projects import RuffCommand


def excepthook(type, value, tb):
    if hasattr(sys, "ps1") or not sys.stderr.isatty():
        # we are in interactive mode or we don't have a tty so call the default
        sys.__excepthook__(type, value, tb)
    else:
        import pdb
        import traceback

        traceback.print_exception(type, value, tb)
        print()
        pdb.post_mortem(tb)


def entrypoint():
    args = parse_args()

    if args.pdb:
        sys.excepthook = excepthook

    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)
    else:
        logging.basicConfig(level=logging.INFO)

    # Use a temporary directory for caching if no cache is specified
    cache_context = (
        tempfile.TemporaryDirectory() if not args.cache else nullcontext(args.cache)
    )

    baseline_executable = args.baseline_executable
    if not args.baseline_executable.exists():
        baseline_executable = get_executable_path(str(args.baseline_executable))
        if not baseline_executable:
            print(
                f"Could not find ruff baseline executable: {args.baseline_executable}",
                sys.stderr,
            )
            exit(1)
        logger.info(
            "Resolved baseline executable %s to %s",
            args.baseline_executable,
            baseline_executable,
        )

    comparison_executable = args.comparison_executable
    if not args.comparison_executable.exists():
        comparison_executable = get_executable_path(str(args.comparison_executable))
        if not comparison_executable:
            print(
                f"Could not find ruff comparison executable: {args.comparison_executable}",
                sys.stderr,
            )
            exit(1)
        logger.info(
            "Resolved comparison executable %s to %s",
            args.comparison_executable,
            comparison_executable,
        )

    targets = DEFAULT_TARGETS
    if args.force_preview:
        targets = [target.with_preview_enabled() for target in targets]

    format_comparison = (
        FormatComparison(args.format_comparison)
        if args.ruff_command == RuffCommand.format.value
        else None
    )

    with cache_context as cache:
        loop = asyncio.get_event_loop()
        main_task = asyncio.ensure_future(
            main(
                command=RuffCommand(args.ruff_command),
                baseline_executable=baseline_executable,
                comparison_executable=comparison_executable,
                targets=targets,
                format=OutputFormat(args.output_format),
                project_dir=Path(cache),
                raise_on_failure=args.pdb,
                format_comparison=format_comparison,
            )
        )
        # https://stackoverflow.com/a/58840987/3549270
        for signal in [SIGINT, SIGTERM]:
            loop.add_signal_handler(signal, main_task.cancel)
        try:
            loop.run_until_complete(main_task)
        finally:
            loop.close()


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Check two versions of ruff against a corpus of open-source code.",
    )

    # TODO: Support non-default `--targets`
    # parser.add_argument(
    #     "--targets",
    #     type=Path,
    #     help=(
    #         "Optional JSON files to use over the default repositories. "
    #         "Supports both github_search_*.jsonl and known-github-tomls.jsonl."
    #     ),
    # )
    parser.add_argument(
        "--cache",
        type=Path,
        help="Location for caching cloned repositories",
    )
    parser.add_argument(
        "--output-format",
        choices=[option.value for option in OutputFormat],
        default="markdown",
        help="Location for caching cloned repositories",
    )
    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable debug logging",
    )
    parser.add_argument(
        "--pdb",
        action="store_true",
        help="Enable debugging on failure",
    )
    parser.add_argument(
        "--force-preview",
        action="store_true",
        help="Force preview mode to be enabled for all projects",
    )
    parser.add_argument(
        "--format-comparison",
        choices=[option.value for option in FormatComparison],
        default=FormatComparison.ruff_and_ruff,
        help="Type of comparison to make when checking formatting.",
    )
    parser.add_argument(
        "ruff_command",
        choices=[option.value for option in RuffCommand],
        help="The Ruff command to test",
    )
    parser.add_argument(
        "baseline_executable",
        type=Path,
    )
    parser.add_argument(
        "comparison_executable",
        type=Path,
    )

    return parser.parse_args()


def get_executable_path(name: str) -> Path | None:
    # Add suffix for Windows executables
    name += ".exe" if sys.platform == "win32" and not name.endswith(".exe") else ""

    path = os.path.join(sysconfig.get_path("scripts"), name)

    # The executable in the current interpreter's scripts directory.
    if os.path.exists(path):
        return Path(path)

    # The executable in the global environment.
    environment_path = shutil.which(name)
    if environment_path:
        return Path(environment_path)

    return None